from unittest.mock import ANY, Mock, call import pytest from flask import url_for from freezegun import freeze_time from notifications_python_client.errors import HTTPError import app from tests import service_json from tests.conftest import ( SERVICE_ONE_ID, create_active_caseworking_user, create_active_user_with_permissions, normalize_spaces, ) @pytest.fixture() def _mock_no_users_for_service(mocker): mocker.patch("app.models.user.Users.client_method", return_value=[]) @pytest.fixture() def mock_get_existing_user_by_email(mocker, api_user_active): return mocker.patch( "app.user_api_client.get_user_by_email", return_value=api_user_active ) @pytest.fixture() def mock_check_invite_token(mocker, sample_invite): return mocker.patch("app.invite_api_client.check_token", return_value=sample_invite) @pytest.mark.usefixtures("_mock_no_users_for_service") @freeze_time("2021-12-12 12:12:12") def test_existing_user_accept_invite_calls_api_and_redirects_to_dashboard( client_request, service_one, api_user_active, mock_check_invite_token, mock_get_existing_user_by_email, mock_accept_invite, mock_add_user_to_service, mock_get_service, mocker, mock_events, mock_get_user, mock_update_user_attribute, ): client_request.logout() expected_service = service_one["id"] expected_permissions = { "view_activity", "send_messages", "manage_service", "manage_api_keys", } client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_redirect=url_for( "main.service_dashboard", service_id=expected_service ), ) mock_check_invite_token.assert_called_with("thisisnotarealtoken") mock_get_existing_user_by_email.assert_called_with("invited_user@test.gsa.gov") assert mock_accept_invite.call_count == 1 mock_add_user_to_service.assert_called_with( expected_service, api_user_active["id"], expected_permissions, [], ) @pytest.mark.usefixtures("_mock_no_users_for_service") def test_existing_user_with_no_permissions_or_folder_permissions_accept_invite( client_request, mocker, service_one, api_user_active, sample_invite, mock_check_invite_token, mock_get_existing_user_by_email, mock_add_user_to_service, mock_get_service, mock_events, mock_get_user, mock_update_user_attribute, ): client_request.logout() expected_service = service_one["id"] sample_invite["permissions"] = "" expected_permissions = set() expected_folder_permissions = [] mocker.patch("app.invite_api_client.accept_invite", return_value=sample_invite) client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, ) mock_add_user_to_service.assert_called_with( expected_service, api_user_active["id"], expected_permissions, expected_folder_permissions, ) def test_if_existing_user_accepts_twice_they_redirect_to_sign_in( client_request, mocker, sample_invite, mock_check_invite_token, mock_get_service, mock_update_user_attribute, ): client_request.logout() # Logging out updates the current session ID to `None` mock_update_user_attribute.reset_mock() sample_invite["status"] = "accepted" page = client_request.get( "main.accept_invite", token="thisisnotarealtoken", _follow_redirects=True, ) assert ( page.h1.string, page.select("main p")[0].text.strip(), ) == ( "You need to sign in again", "We signed you out because you have not used Notify for a while.", ) # We don’t let people update `email_access_validated_at` using an # already-accepted invite assert mock_update_user_attribute.called is False @pytest.mark.usefixtures("_mock_no_users_for_service") def test_invite_goes_in_session( client_request, mocker, sample_invite, mock_get_service, api_user_active, mock_check_invite_token, mock_get_user_by_email, mock_add_user_to_service, mock_accept_invite, ): sample_invite["email_address"] = "test@user.gsa.gov" client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, _expected_redirect=url_for( "main.service_dashboard", service_id=SERVICE_ONE_ID, ), _follow_redirects=False, ) with client_request.session_transaction() as session: assert session["invited_user_id"] == sample_invite["id"] @pytest.mark.usefixtures("_mock_no_users_for_service") @pytest.mark.parametrize( ("user", "landing_page_title"), [ (create_active_user_with_permissions(), "Dashboard"), (create_active_caseworking_user(), "Templates"), ], ) def test_accepting_invite_removes_invite_from_session( client_request, mocker, sample_invite, mock_get_service, service_one, mock_check_invite_token, mock_get_user_by_email, mock_add_user_to_service, mock_accept_invite, mock_get_service_templates, mock_get_template_statistics, mock_has_no_jobs, mock_get_service_statistics, mock_get_template_folders, mock_get_annual_usage_for_service, mock_get_monthly_usage_for_service, mock_get_free_sms_fragment_limit, mock_get_inbound_sms_summary, mock_get_api_keys, fake_uuid, user, landing_page_title, ): sample_invite["email_address"] = user["email_address"] client_request.login(user) page = client_request.get( "main.accept_invite", token="thisisnotarealtoken", _follow_redirects=True, ) assert normalize_spaces(page.select_one("h1").text) == landing_page_title with client_request.session_transaction() as session: assert "invited_user_id" not in session @freeze_time("2021-12-12T12:12:12") def test_existing_user_of_service_get_redirected_to_signin( client_request, mocker, api_user_active, sample_invite, mock_get_service, mock_get_user_by_email, mock_check_invite_token, mock_accept_invite, mock_update_user_attribute, ): client_request.logout() sample_invite["email_address"] = api_user_active["email_address"] mocker.patch("app.models.user.Users.client_method", return_value=[api_user_active]) page = client_request.get( "main.accept_invite", token="thisisnotarealtoken", _follow_redirects=True, ) assert ( page.h1.string, page.select("main p")[0].text.strip(), ) == ( "You need to sign in again", "We signed you out because you have not used Notify for a while.", ) assert mock_accept_invite.call_count == 1 @pytest.mark.usefixtures("_mock_no_users_for_service") def test_accept_invite_redirects_if_api_raises_an_error_that_they_are_already_part_of_the_service( client_request, mocker, api_user_active, sample_invite, mock_get_existing_user_by_email, mock_check_invite_token, mock_accept_invite, mock_get_service, mock_get_user, mock_update_user_attribute, ): client_request.logout() mocker.patch( "app.user_api_client.add_user_to_service", side_effect=HTTPError( response=Mock( status_code=400, json={ "result": "error", "message": { f"User id: {api_user_active['id']} already part of service id: {SERVICE_ONE_ID}" }, }, ), message=f"User id: {api_user_active['id']} already part of service id: {SERVICE_ONE_ID}", ), ) client_request.get( "main.accept_invite", token="thisisnotarealtoken", _follow_redirects=False, _expected_redirect=url_for("main.service_dashboard", service_id=SERVICE_ONE_ID), ) @pytest.mark.usefixtures("_mock_no_users_for_service") def test_existing_signed_out_user_accept_invite_redirects_to_sign_in( client_request, service_one, api_user_active, sample_invite, mock_check_invite_token, mock_get_existing_user_by_email, mock_add_user_to_service, mock_accept_invite, mock_get_service, mocker, mock_events, mock_get_user, mock_update_user_attribute, ): client_request.logout() expected_service = service_one["id"] expected_permissions = { "view_activity", "send_messages", "manage_service", "manage_api_keys", } page = client_request.get( "main.accept_invite", token="thisisnotarealtoken", _follow_redirects=True, ) mock_check_invite_token.assert_called_with("thisisnotarealtoken") mock_get_existing_user_by_email.assert_called_with("invited_user@test.gsa.gov") mock_add_user_to_service.assert_called_with( expected_service, api_user_active["id"], expected_permissions, sample_invite["folder_permissions"], ) assert mock_accept_invite.call_count == 1 assert ( page.h1.string, page.select("main p")[0].text.strip(), ) == ( "You need to sign in again", "We signed you out because you have not used Notify for a while.", ) @pytest.mark.usefixtures("_mock_no_users_for_service") def test_new_user_accept_invite_calls_api_and_redirects_to_registration( client_request, service_one, mock_check_invite_token, mock_dont_get_user_by_email, mock_add_user_to_service, mock_get_service, mocker, ): client_request.logout() client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_redirect="/register-from-invite", ) mock_check_invite_token.assert_called_with("thisisnotarealtoken") mock_dont_get_user_by_email.assert_called_with("invited_user@test.gsa.gov") @pytest.mark.usefixtures("_mock_no_users_for_service") def test_new_user_accept_invite_calls_api_and_views_registration_page( client_request, service_one, sample_invite, mock_check_invite_token, mock_dont_get_user_by_email, mock_get_invited_user_by_id, mock_add_user_to_service, mock_get_service, mocker, ): client_request.logout() page = client_request.get( "main.accept_invite", token="thisisnotarealtoken", _follow_redirects=True, ) mock_check_invite_token.assert_called_with("thisisnotarealtoken") mock_dont_get_user_by_email.assert_called_with("invited_user@test.gsa.gov") mock_get_invited_user_by_id.assert_called_once_with(sample_invite["id"]) assert page.h1.string.strip() == "Create an account" assert normalize_spaces(page.select_one("main p").text) == ( "Your account will be created with this email address: " "invited_user@test.gsa.gov" ) form = page.find("form") name = form.find("input", id="name") password = form.find("input", id="password") service = form.find("input", type="hidden", id="service") email = form.find("input", type="hidden", id="email_address") assert email assert email.attrs["value"] == "invited_user@test.gsa.gov" assert name assert password assert service assert service.attrs["value"] == service_one["id"] def test_cancelled_invited_user_accepts_invited_redirect_to_cancelled_invitation( client_request, mock_get_user, mock_get_service, sample_invite, mock_check_invite_token, mock_update_user_attribute, ): client_request.logout() mock_update_user_attribute.reset_mock() sample_invite["status"] = "cancelled" page = client_request.get("main.accept_invite", token="thisisnotarealtoken") app.invite_api_client.check_token.assert_called_with("thisisnotarealtoken") assert page.h1.string.strip() == "The invitation you were sent has been cancelled" # We don’t let people update `email_access_validated_at` using an # cancelled invite assert mock_update_user_attribute.called is False @pytest.mark.parametrize( ("admin_endpoint", "api_endpoint"), [ ("main.accept_invite", "app.invite_api_client.check_token"), ("main.accept_org_invite", "app.org_invite_api_client.check_token"), ], ) def test_new_user_accept_invite_with_malformed_token( admin_endpoint, api_endpoint, client_request, service_one, mocker, ): client_request.logout() mocker.patch( api_endpoint, side_effect=HTTPError( response=Mock( status_code=400, json={ "result": "error", "message": { "invitation": { "Something’s wrong with this link. Make sure you’ve copied the whole thing." } }, }, ), message={ "invitation": "Something’s wrong with this link. Make sure you’ve copied the whole thing." }, ), ) page = client_request.get( admin_endpoint, token="thisisnotarealtoken", _follow_redirects=True ) assert ( normalize_spaces(page.select_one(".banner-dangerous").text) == "Something’s wrong with this link. Make sure you’ve copied the whole thing." ) @pytest.mark.usefixtures("_mock_no_users_for_service") def test_new_user_accept_invite_completes_new_registration_redirects_to_verify( client_request, service_one, sample_invite, api_user_active, mock_check_invite_token, mock_dont_get_user_by_email, mock_email_is_not_already_in_use, mock_register_user, mock_send_verify_code, mock_get_invited_user_by_id, mock_accept_invite, mock_add_user_to_service, mock_get_service, mocker, ): client_request.logout() expected_redirect_location = "/register-from-invite" client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_redirect=expected_redirect_location, ) with client_request.session_transaction() as session: assert session.get("invited_user_id") == sample_invite["id"] data = { "service": sample_invite["service"], "email_address": sample_invite["email_address"], "from_user": sample_invite["from_user"], "password": "longpassword", "mobile_number": "+12027890123", "name": "Invited User", "auth_type": "email_auth", } expected_redirect_location = "/verify" client_request.post( "main.register_from_invite", _data=data, _expected_redirect=expected_redirect_location, ) mock_send_verify_code.assert_called_once_with(ANY, "sms", data["mobile_number"]) mock_get_invited_user_by_id.assert_called_once_with(sample_invite["id"]) mock_register_user.assert_called_with( data["name"], data["email_address"], data["mobile_number"], data["password"], data["auth_type"], ) assert mock_accept_invite.call_count == 1 def test_signed_in_existing_user_cannot_use_anothers_invite( client_request, mocker, api_user_active, mock_check_invite_token, sample_invite, mock_get_user, mock_accept_invite, mock_get_service, ): mocker.patch( "app.user_api_client.get_users_for_service", return_value=[api_user_active] ) page = client_request.get( "main.accept_invite", token="thisisnotarealtoken", _follow_redirects=True, _expected_status=403, ) assert page.h1.string.strip() == "You’re not allowed to see this page" flash_banners = page.find_all("div", class_="banner-dangerous") assert len(flash_banners) == 1 banner_contents = normalize_spaces(flash_banners[0].text) assert "You’re signed in as test@user.gsa.gov." in banner_contents assert "This invite is for another email address." in banner_contents assert "Sign out and click the link again to accept this invite." in banner_contents assert mock_accept_invite.call_count == 0 def test_accept_invite_does_not_treat_email_addresses_as_case_sensitive( client_request, mocker, api_user_active, sample_invite, mock_accept_invite, mock_check_invite_token, mock_get_user_by_email, ): # the email address of api_user_active is 'test@user.gsa.gov' sample_invite["email_address"] = "TEST@user.gsa.gov" mocker.patch("app.models.user.Users.client_method", return_value=[api_user_active]) client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, _expected_redirect=url_for( "main.service_dashboard", service_id=SERVICE_ONE_ID, ), ) @pytest.mark.usefixtures("_mock_no_users_for_service") def test_new_invited_user_verifies_and_added_to_service( client_request, service_one, sample_invite, api_user_active, mock_check_invite_token, mock_dont_get_user_by_email, mock_email_is_not_already_in_use, mock_register_user, mock_send_verify_code, mock_check_verify_code, mock_get_user, mock_update_user_attribute, mock_add_user_to_service, mock_accept_invite, mock_get_service, mock_get_invited_user_by_id, mock_get_service_templates, mock_get_template_statistics, mock_has_no_jobs, mock_has_permissions, mock_get_service_statistics, mock_get_annual_usage_for_service, mock_get_free_sms_fragment_limit, mock_create_event, mocker, ): client_request.logout() # visit accept token page client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_redirect=url_for("main.register_from_invite"), ) # get redirected to register from invite data = { "service": sample_invite["service"], "email_address": sample_invite["email_address"], "from_user": sample_invite["from_user"], "password": "longpassword", "mobile_number": "+12027890123", "name": "Invited User", "auth_type": "sms_auth", } client_request.post( "main.register_from_invite", _data=data, _expected_redirect=url_for("main.verify"), ) # that sends user on to verify page = client_request.post( "main.verify", _data={"sms_code": "123456"}, _follow_redirects=True, ) # when they post codes back to admin user should be added to # service and sent on to dash board expected_permissions = { "view_activity", "send_messages", "manage_service", "manage_api_keys", } with client_request.session_transaction() as session: assert "invited_user_id" not in session new_user_id = session["user_id"] mock_add_user_to_service.assert_called_with( data["service"], new_user_id, expected_permissions, [] ) mock_accept_invite.assert_called_with(data["service"], sample_invite["id"]) mock_check_verify_code.assert_called_once_with(new_user_id, "123456", "sms") assert service_one["id"] == session["service_id"] assert page.find("h1").text == "Dashboard" @pytest.mark.parametrize( ("service_permissions", "trial_mode", "expected_endpoint", "extra_args"), [ ([], True, "main.service_dashboard", {}), ([], False, "main.service_dashboard", {}), ], ) def test_new_invited_user_is_redirected_to_correct_place( mocker, client_request, sample_invite, mock_check_invite_token, mock_check_verify_code, mock_get_user, mock_dont_get_user_by_email, mock_add_user_to_service, mock_get_invited_user_by_id, mock_events, mock_get_service, service_permissions, trial_mode, expected_endpoint, extra_args, ): client_request.logout() mocker.patch( "app.service_api_client.get_service", return_value={ "data": service_json( sample_invite["service"], restricted=trial_mode, permissions=service_permissions, ) }, ) client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, ) with client_request.session_transaction() as session: session["user_details"] = { "email": sample_invite["email_address"], "id": sample_invite["id"], } client_request.post( "main.verify", _data={"sms_code": "123456"}, _expected_redirect=url_for( expected_endpoint, service_id=sample_invite["service"], **extra_args ), ) @pytest.mark.usefixtures("_mock_no_users_for_service") @freeze_time("2021-12-12 12:12:12") def test_existing_user_accepts_and_sets_email_auth( client_request, api_user_active, service_one, sample_invite, mock_get_existing_user_by_email, mock_accept_invite, mock_check_invite_token, mock_update_user_attribute, mock_add_user_to_service, mocker, ): sample_invite["email_address"] = api_user_active["email_address"] service_one["permissions"].append("email_auth") sample_invite["auth_type"] = "email_auth" client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, _expected_redirect=url_for( "main.service_dashboard", service_id=service_one["id"] ), ) mock_get_existing_user_by_email.assert_called_once_with("test@user.gsa.gov") assert mock_update_user_attribute.call_args_list == [ call(api_user_active["id"], email_access_validated_at="2021-12-12T12:12:12"), call(api_user_active["id"], auth_type="email_auth"), ] mock_add_user_to_service.assert_called_once_with( ANY, api_user_active["id"], ANY, ANY ) @pytest.mark.usefixtures("_mock_no_users_for_service") @freeze_time("2021-12-12 12:12:12") def test_platform_admin_user_accepts_and_preserves_auth( client_request, platform_admin_user, service_one, sample_invite, mock_check_invite_token, mock_accept_invite, mock_add_user_to_service, mocker, ): sample_invite["email_address"] = platform_admin_user["email_address"] sample_invite["auth_type"] = "email_auth" service_one["permissions"].append("email_auth") mocker.patch( "app.user_api_client.get_user_by_email", return_value=platform_admin_user ) mock_update_user_attribute = mocker.patch( "app.user_api_client.update_user_attribute", return_value=platform_admin_user, ) client_request.login(platform_admin_user) client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, _expected_redirect=url_for( "main.service_dashboard", service_id=service_one["id"] ), ) mock_update_user_attribute.assert_called_once_with( platform_admin_user["id"], email_access_validated_at="2021-12-12T12:12:12", ) assert mock_add_user_to_service.called @pytest.mark.usefixtures("_mock_no_users_for_service") @freeze_time("2021-12-12 12:12:12") def test_existing_user_doesnt_get_auth_changed_by_service_without_permission( client_request, api_user_active, service_one, sample_invite, mock_get_user_by_email, mock_check_invite_token, mock_accept_invite, mock_update_user_attribute, mock_add_user_to_service, mocker, ): sample_invite["email_address"] = api_user_active["email_address"] assert "email_auth" not in service_one["permissions"] sample_invite["auth_type"] = "email_auth" client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, _expected_redirect=url_for( "main.service_dashboard", service_id=service_one["id"] ), ) mock_update_user_attribute.assert_called_once_with( api_user_active["id"], email_access_validated_at="2021-12-12T12:12:12", ) @pytest.mark.usefixtures("_mock_no_users_for_service") @freeze_time("2021-12-12 12:12:12") def test_existing_email_auth_user_without_phone_cannot_set_sms_auth( client_request, api_user_active, service_one, sample_invite, mock_check_invite_token, mock_accept_invite, mock_update_user_attribute, mock_add_user_to_service, mocker, ): sample_invite["email_address"] = api_user_active["email_address"] service_one["permissions"].append("email_auth") api_user_active["auth_type"] = "email_auth" api_user_active["mobile_number"] = None sample_invite["auth_type"] = "sms_auth" mocker.patch("app.user_api_client.get_user_by_email", return_value=api_user_active) client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, _expected_redirect=url_for( "main.service_dashboard", service_id=service_one["id"] ), ) mock_update_user_attribute.assert_called_once_with( api_user_active["id"], email_access_validated_at="2021-12-12T12:12:12", ) @pytest.mark.usefixtures("_mock_no_users_for_service") @freeze_time("2021-12-12 12:12:12") def test_existing_email_auth_user_with_phone_can_set_sms_auth( client_request, api_user_active, service_one, sample_invite, mock_get_existing_user_by_email, mock_check_invite_token, mock_accept_invite, mock_update_user_attribute, mock_add_user_to_service, mocker, ): sample_invite["email_address"] = api_user_active["email_address"] service_one["permissions"].append("email_auth") sample_invite["auth_type"] = "sms_auth" client_request.get( "main.accept_invite", token="thisisnotarealtoken", _expected_status=302, _expected_redirect=url_for( "main.service_dashboard", service_id=service_one["id"] ), ) mock_get_existing_user_by_email.assert_called_once_with( sample_invite["email_address"] ) assert mock_update_user_attribute.call_args_list == [ call(api_user_active["id"], email_access_validated_at="2021-12-12T12:12:12"), call(api_user_active["id"], auth_type="sms_auth"), ]