kitos
f13764d9e2
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
send_webhook_email() only returned False on a network exception — a webhook that responded with a non-2xx status (bad URL, wrong API key, misconfigured Power Automate flow) was still reported as delivered. Now checks response.ok and returns False on rejection too. send_password_setup_email() also never checked the return value at all, so the admin's 'Send Email' action (including resending a password-reset email after the user already set their password) always showed success even when nothing was actually sent. Now raises a clear error instead.
211 lines
8.6 KiB
Python
211 lines
8.6 KiB
Python
"""End-to-end coverage for the passwordless user creation + emailed
|
|
set-password-link flow (also reused for password resets).
|
|
"""
|
|
|
|
import uuid
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def new_user_id(api, auth_headers):
|
|
resp = api(
|
|
"post", "/api/v1/users", auth_headers,
|
|
json={"full_name": "Set Password User", "email": "setpw@test.com", "role": "viewer"},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
return resp.json()["id"]
|
|
|
|
|
|
def test_send_password_email_requires_webhook_configured(api, auth_headers, new_user_id):
|
|
resp = api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
|
assert resp.status_code == 400
|
|
assert "webhook" in resp.text.lower()
|
|
|
|
|
|
def test_admin_can_configure_password_webhook(api, auth_headers):
|
|
resp = api(
|
|
"patch", "/api/v1/system/email-webhook-config", auth_headers,
|
|
json={"url": "https://example.com/power-automate-hook"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["configured"] is True
|
|
|
|
get_resp = api("get", "/api/v1/system/email-webhook-config", auth_headers)
|
|
assert get_resp.json()["url"] == "https://example.com/power-automate-hook"
|
|
|
|
|
|
def test_admin_can_configure_webhook_api_key(api, auth_headers):
|
|
resp = api(
|
|
"patch", "/api/v1/system/email-webhook-config", auth_headers,
|
|
json={"url": "https://example.com/power-automate-hook", "api_key": "super-secret-key"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["api_key_set"] is True
|
|
# The key itself is never echoed back.
|
|
assert "api_key" not in resp.json()
|
|
assert "super-secret-key" not in resp.text
|
|
|
|
|
|
def test_send_password_email_sends_api_key_header(api, db, auth_headers, new_user_id):
|
|
api(
|
|
"patch", "/api/v1/system/email-webhook-config", auth_headers,
|
|
json={"url": "https://example.com/power-automate-hook", "api_key": "super-secret-key"},
|
|
)
|
|
with patch("app.services.webhook_email_service.requests.post") as mock_post:
|
|
api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
|
call_kwargs = mock_post.call_args
|
|
assert call_kwargs.kwargs["headers"] == {"x-api-key": "super-secret-key"}
|
|
|
|
|
|
def test_send_password_email_posts_to_webhook_and_issues_token(api, db, auth_headers, new_user_id):
|
|
api(
|
|
"patch", "/api/v1/system/email-webhook-config", auth_headers,
|
|
json={"url": "https://example.com/power-automate-hook"},
|
|
)
|
|
|
|
with patch("app.services.webhook_email_service.requests.post") as mock_post:
|
|
resp = api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
|
assert resp.status_code == 200, resp.text
|
|
mock_post.assert_called_once()
|
|
call_kwargs = mock_post.call_args
|
|
assert call_kwargs.args[0] == "https://example.com/power-automate-hook"
|
|
payload = call_kwargs.kwargs["json"]
|
|
assert payload["to"] == "setpw@test.com"
|
|
assert payload["subject"] == "Set Your Password"
|
|
assert "Hi Set Password User" in payload["body"]
|
|
assert "token=" in payload["body"]
|
|
assert "Purple Team Engineering" in payload["body"]
|
|
|
|
from app.models.password_setup_token import PasswordSetupToken
|
|
token_row = db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == uuid.UUID(new_user_id)).first()
|
|
assert token_row is not None
|
|
assert token_row.used_at is None
|
|
|
|
|
|
def test_send_password_email_fails_loudly_when_webhook_rejects(api, auth_headers, new_user_id):
|
|
"""A webhook that's reachable but rejects the request (bad URL/API key,
|
|
wrong Power Automate config, etc.) must not be reported as a success —
|
|
previously the HTTP response status was never checked."""
|
|
api(
|
|
"patch", "/api/v1/system/email-webhook-config", auth_headers,
|
|
json={"url": "https://example.com/power-automate-hook"},
|
|
)
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
fake_response = MagicMock(ok=False, status_code=404, text="not found")
|
|
with patch("app.services.webhook_email_service.requests.post", return_value=fake_response):
|
|
resp = api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
|
assert resp.status_code == 400
|
|
assert "failed to send" in resp.text.lower()
|
|
|
|
|
|
def test_send_password_email_resend_after_password_set_is_a_reset_email(api, db, auth_headers, new_user_id):
|
|
"""Once a user has already set their password, hitting 'Send Email'
|
|
again must still actually send — as a password-reset email — not
|
|
silently no-op."""
|
|
api(
|
|
"patch", "/api/v1/system/email-webhook-config", auth_headers,
|
|
json={"url": "https://example.com/power-automate-hook"},
|
|
)
|
|
|
|
with patch("app.services.webhook_email_service.requests.post") as mock_post:
|
|
resp = api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
|
assert resp.status_code == 200, resp.text
|
|
first_payload = mock_post.call_args.kwargs["json"]
|
|
assert first_payload["subject"] == "Set Your Password"
|
|
|
|
from app.models.password_setup_token import PasswordSetupToken
|
|
token_row = db.query(PasswordSetupToken).filter(
|
|
PasswordSetupToken.user_id == uuid.UUID(new_user_id),
|
|
).first()
|
|
api(
|
|
"post", "/api/v1/auth/set-password", {},
|
|
json={"token": token_row.token, "new_password": "BrandNewPass123!@#"},
|
|
)
|
|
|
|
# Now the user has must_change_password=False — resending must still
|
|
# go through, this time as a reset email.
|
|
with patch("app.services.webhook_email_service.requests.post") as mock_post_2:
|
|
resp2 = api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
|
assert resp2.status_code == 200, resp2.text
|
|
mock_post_2.assert_called_once()
|
|
second_payload = mock_post_2.call_args.kwargs["json"]
|
|
assert second_payload["subject"] == "Reset Your Password"
|
|
|
|
def test_set_password_with_valid_token_succeeds(api, db, auth_headers, new_user_id):
|
|
api(
|
|
"patch", "/api/v1/system/email-webhook-config", auth_headers,
|
|
json={"url": "https://example.com/power-automate-hook"},
|
|
)
|
|
with patch("app.services.webhook_email_service.requests.post"):
|
|
api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
|
|
|
from app.models.password_setup_token import PasswordSetupToken
|
|
token_row = db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == uuid.UUID(new_user_id)).first()
|
|
|
|
validate = api("get", f"/api/v1/auth/set-password/validate?token={token_row.token}", {})
|
|
assert validate.status_code == 200
|
|
assert validate.json()["valid"] is True
|
|
assert validate.json()["full_name"] == "Set Password User"
|
|
|
|
resp = api(
|
|
"post", "/api/v1/auth/set-password", {},
|
|
json={"token": token_row.token, "new_password": "BrandNewPass123!@#"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
login = api(
|
|
"post", "/api/v1/auth/login", {},
|
|
data={"username": "setpw@test.com", "password": "BrandNewPass123!@#"},
|
|
)
|
|
assert login.status_code == 200, login.text
|
|
|
|
|
|
def test_set_password_token_cannot_be_reused(api, db, auth_headers, new_user_id):
|
|
api(
|
|
"patch", "/api/v1/system/email-webhook-config", auth_headers,
|
|
json={"url": "https://example.com/power-automate-hook"},
|
|
)
|
|
with patch("app.services.webhook_email_service.requests.post"):
|
|
api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
|
|
|
from app.models.password_setup_token import PasswordSetupToken
|
|
token_row = db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == uuid.UUID(new_user_id)).first()
|
|
|
|
first = api(
|
|
"post", "/api/v1/auth/set-password", {},
|
|
json={"token": token_row.token, "new_password": "BrandNewPass123!@#"},
|
|
)
|
|
assert first.status_code == 200
|
|
|
|
second = api(
|
|
"post", "/api/v1/auth/set-password", {},
|
|
json={"token": token_row.token, "new_password": "AnotherPass456!@#"},
|
|
)
|
|
assert second.status_code == 400
|
|
|
|
|
|
def test_set_password_invalid_token_rejected(api):
|
|
validate = api("get", "/api/v1/auth/set-password/validate?token=not-a-real-token", {})
|
|
assert validate.status_code == 200
|
|
assert validate.json()["valid"] is False
|
|
|
|
resp = api(
|
|
"post", "/api/v1/auth/set-password", {},
|
|
json={"token": "not-a-real-token", "new_password": "BrandNewPass123!@#"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_password_webhook_config_requires_admin(api, red_lead_headers):
|
|
resp = api("get", "/api/v1/system/email-webhook-config", red_lead_headers)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
def test_send_password_email_requires_admin(api, red_lead_headers, new_user_id):
|
|
resp = api("post", f"/api/v1/users/{new_user_id}/send-password-email", red_lead_headers)
|
|
assert resp.status_code == 403
|