"""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_create_user_automatically_sends_set_password_email(api, db, auth_headers): 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", "/api/v1/users", auth_headers, json={"full_name": "Auto Send User", "email": "autosend@test.com", "role": "viewer"}, ) assert resp.status_code == 201, resp.text mock_post.assert_called_once() payload = mock_post.call_args.kwargs["json"] assert payload["to"] == "autosend@test.com" assert payload["subject"] == "Set Your Password" assert "token=" in payload["body"] from app.models.password_setup_token import PasswordSetupToken token_row = db.query(PasswordSetupToken).filter( PasswordSetupToken.user_id == uuid.UUID(resp.json()["id"]), ).first() assert token_row is not None def test_create_user_succeeds_even_if_auto_send_fails(api, auth_headers): """No webhook configured yet — the automatic first-email attempt fails, but user creation itself must still succeed.""" resp = api( "post", "/api/v1/users", auth_headers, json={"full_name": "No Webhook User", "email": "nowebhook@test.com", "role": "viewer"}, ) assert resp.status_code == 201, resp.text 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