End to end woes continue.

Signed-off-by: Cliff Hill <clifford.hill@gsa.gov>

app/config.py

@@ -183,6 +183,16 @@ class E2ETest(Staging):
         "notify-admin-logo-upload-bucket-staging"
     )
 
+    API_HOST_NAME = "http://localhost:6011"  # nosec B105 - only used in development
+    # credential overrides
+    DANGEROUS_SALT = "development-notify-salt"
+    SECRET_KEY = "dev-notify-secret-key"  # nosec B105 - only used in development
+    # ADMIN_CLIENT_USER_NAME is called ADMIN_CLIENT_ID in api repo, they should match
+    ADMIN_CLIENT_USER_NAME = "notify-admin"
+    ADMIN_CLIENT_SECRET = (
+        "dev-notify-secret-key"  # nosec B105 - only used in development
+    )
+
 
 class Demo(Staging):
     HEADER_COLOUR = "#6F72AF"  # $mauve

app/notify_client/__init__.py

@@ -85,8 +85,9 @@ class NotifyAdminAPIClient(BaseAPIClient):
             abort(403)
 
     def post(self, *args, **kwargs):
-        self.check_inactive_service()
-        self.check_inactive_user(args)
+        if os.getenv('NOTIFY_ENVIRONMENT') != "e2etest":
+            self.check_inactive_service()
+            self.check_inactive_user(args)
         return super().post(*args, **kwargs)
 
     def put(self, *args, **kwargs):

tests/end_to_end/conftest.py

@@ -1,11 +1,53 @@
+import datetime
 import os
+from contextlib import contextmanager
+from unittest.mock import patch
 
 import pytest
 from axe_core_python.sync_playwright import Axe
+from flask import Flask, current_app, url_for
+from flask import session as flask_session
+from flask_login import login_user
+from flask.testing import FlaskClient
+
+from app import create_app
+from app.models.service import Service
+from app.models.user import User
+from app.notify_client.service_api_client import service_api_client
+from app.notify_client.user_api_client import user_api_client
+from notifications_python_client.errors import HTTPError
+from .. import TestClient
+
 
 E2E_TEST_URI = os.getenv("NOTIFY_E2E_TEST_URI")
 
 
+class TestClient(FlaskClient):
+    def login(self, user, mocker=None, service=None):
+        # Skipping authentication here and just log them in
+        model_user = User(user)
+        with self.session_transaction() as session:
+            session["current_session_id"] = model_user.current_session_id
+            session["user_id"] = model_user.id
+        if mocker:
+            mocker.patch("app.user_api_client.get_user", return_value=user)
+        if mocker and service:
+            with self.session_transaction() as session:
+                session["service_id"] = service["id"]
+            mocker.patch(
+                "app.service_api_client.get_service", return_value={"data": service}
+            )
+
+        with patch("app.events_api_client.create_event"):
+            login_user(model_user, force=True)  # forces the user to be logged in.
+        with self.session_transaction() as test_session:
+            for key, value in flask_session.items():
+                test_session[key] = value
+
+    def logout(self, user):
+        self.get(url_for("main.sign_out"))
+
+
 @pytest.fixture
 def end_to_end_context(browser):
     context = browser.new_context()
@@ -36,3 +78,107 @@ def check_axe_report(page):
             "minor",
             "moderate",
         ], f"Accessibility violation: {violation}"
+
+
+@pytest.fixture
+def notify_admin_e2e():
+    os.environ["NOTIFY_ENVIRONMENT"] = "e2etest"
+
+    application = Flask("app")
+    create_app(application)
+
+    application.test_client_class = TestClient
+
+    with application.app_context():
+        yield application
+
+
+@pytest.fixture
+def default_user(notify_admin_e2e):
+    user_data = user_api_client.get_user_by_email(os.getenv("NOTIFY_E2E_TEST_EMAIL"))
+    return user_data
+
+
+@pytest.fixture
+def client(notify_admin_e2e, default_user):
+    """
+    Do not use this fixture directly – use `client_request` instead
+    """
+    with notify_admin_e2e.test_request_context(), notify_admin_e2e.test_client() as client:
+        client.allow_subdomain_redirects = True
+        try:
+            client.login(default_user)
+            yield client
+        finally:
+            client.logout(default_user)
+
+
+# Need e2e service defined here?
+@pytest.fixture()
+def default_service(browser, client, default_user):
+    current_date_time = datetime.datetime.now()
+    now = current_date_time.strftime("%m/%d/%Y %H:%M:%S")
+    browser_type = browser.browser_type.name
+    service_name = f"E2E Federal Test Service {now} - {browser_type}"
+
+    service_id = service_api_client.create_service(
+        service_name=service_name,
+        organization_type="federal",
+        message_limit=current_app.config["DEFAULT_SERVICE_LIMIT"],
+        restricted=True,
+        user_id=default_user["id"],
+        email_from=default_user["email_address"],
+    )
+
+    print("*" * 80)
+    print(service_id)
+
+    service_data = service_api_client.get_service(service_id)
+
+    service = Service(service_data)
+
+    from pprint import pprint
+    pprint(dir(service))
+
+    try:
+        yield service
+    finally:
+        service_api_client.archive_service(service.id, None)
+
+
+@contextmanager
+def _set_up_user(
+    default_service,
+    name,
+    email_addr,
+    phone,
+    password,
+    auth_type,
+    permissions,
+    folder_permissions,
+):
+    user = user_api_client.get_user_by_email_or_none(email_addr)
+    if user is None:
+        user = user_api_client.register_user(
+            name, email_addr, phone, password, auth_type
+        )
+    user_api_client.add_user_to_service(
+        default_service.id, user.id, permissions, folder_permissions
+    )
+    user_api_client.activate_user(user.id)
+    yield user
+    user_api_client.deactivate_user(user.id)
+    service_api_client.remove_user_from_service(user.id, default_service.id)
+
+
+@pytest.fixture
+def admin_user(default_service):
+    with _set_up_user(
+        default_service,
+        "E2E Admin Test",
+        "admin@nowhere.huh",
+        "1234567890",
+        "password",
+        "sms",
+    ) as user:
+        yield user

tests/end_to_end/test_campaign_manager_story.py

@@ -0,0 +1,247 @@
+import datetime
+import os
+import re
+import uuid
+
+from playwright.sync_api import expect
+
+from tests.conftest import create_user
+from tests.end_to_end.conftest import check_axe_report
+
+E2E_TEST_URI = os.getenv("NOTIFY_E2E_TEST_URI")
+
+
+def create_new_template(page):
+
+    current_service_link = page.get_by_text("Current service")
+    expect(current_service_link).to_be_visible()
+    current_service_link.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    send_messages_button = page.get_by_role("link", name="Send messages")
+    expect(send_messages_button).to_be_visible()
+    send_messages_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    create_template_button = page.get_by_role("button", name="New template")
+    expect(create_template_button).to_be_visible()
+    create_template_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    start_with_a_blank_template_radio = page.get_by_text("Start with a blank template")
+    expect(start_with_a_blank_template_radio).to_be_visible()
+    start_with_a_blank_template_radio.click()
+
+    continue_button = page.get_by_role("button", name="Continue")
+
+    # continue_button = page.get_by_text("Continue")
+    expect(continue_button).to_be_visible()
+    continue_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    template_name_input = page.get_by_text("Template name")
+    expect(template_name_input).to_be_visible()
+    template_name = str(uuid.uuid4())
+    template_name_input.fill(template_name)
+    message_input = page.get_by_role("textbox", name="Message")
+    expect(message_input).to_be_visible()
+    message = "Test message for e2e test"
+    message_input.fill(message)
+
+    save_button = page.get_by_text("Save")
+    expect(save_button).to_be_visible()
+    save_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    assert template_name in page.content()
+    assert message in page.content()
+
+
+def update_template(page):
+    # Check update template button is visible
+    update_template_button = page.get_by_text("Edit this template")
+    expect(update_template_button).to_be_visible()
+    update_template_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    assert "Edit text message template" in page.content()
+
+    # Update template name and message
+    template_name_input = page.get_by_text("Template name")
+    expect(template_name_input).to_be_visible()
+    template_name = str(uuid.uuid4())
+    template_name_input.fill(template_name)
+    message_input = page.get_by_role("textbox", name="Message")
+    expect(message_input).to_be_visible()
+    message = "Test message is updated for e2e test"
+    message_input.fill(message)
+
+    # Save new template info
+    save_button = page.get_by_text("Save")
+    expect(save_button).to_be_visible()
+    save_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    # Check new template name, message and last edited text are on page.
+    assert template_name in page.content()
+    assert message in page.content()
+    # assert "Last edited just now" in page.content()
+
+
+def delete_template(page):
+    # Check delete template link is visible
+    delete_template_link = page.get_by_text("Delete this template")
+    expect(delete_template_link).to_be_visible()
+    delete_template_link.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    # Check confirmation text is visible
+    assert "Are you sure you want to delete" in page.content()
+
+    # Check that the delete button is visible
+    delete_button = page.get_by_text("Yes, delete")
+    expect(delete_button).to_be_visible()
+    delete_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    assert "Select or create a template" in page.content()
+
+
+def test_campaign_manager_core_user_story(default_service, end_to_end_context):
+    create_user()
+    page = end_to_end_context.new_page()
+    # page.goto(f"{E2E_TEST_URI}/sign-in")
+    # Wait for the next page to fully load.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    current_date_time = datetime.datetime.now()
+    # new_service_name = "E2E Federal Test Service {now} - {browser_type}".format(
+    #    now=current_date_time.strftime("%m/%d/%Y %H:%M:%S"),
+    #    browser_type=page.context.browser.browser_type.name,
+    # )
+    # page.goto(f"{E2E_TEST_URI}/accounts")
+
+    # Check to make sure that we've arrived at the next page.
+    # page.wait_for_load_state("domcontentloaded")
+    # check_axe_report(page)
+
+    # Check to make sure that we've arrived at the next page.
+    # Check the page title exists and matches what we expect.
+    # expect(page).to_have_title(re.compile("Choose service"))
+
+    # Check for the sign in heading.
+    # sign_in_heading = page.get_by_role("heading", name="Choose service")
+    # expect(sign_in_heading).to_be_visible()
+
+    # Retrieve some prominent elements on the page for testing.
+    # add_service_button = page.get_by_role(
+    #     "button", name=re.compile("Add a new service")
+    # )
+
+    # expect(add_service_button).to_be_visible()
+
+    # existing_service_link = page.get_by_role("link", name=new_service_name)
+
+    # Check to see if the service was already created - if so, we should fail.
+    # TODO:  Figure out how to make this truly isolated, and/or work in a
+    #        delete service workflow.
+    # expect(existing_service_link).to_have_count(0)
+
+    # Click on add a new service.
+    a  # dd_service_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    # page.wait_for_load_state("domcontentloaded")
+    # check_axe_report(page)
+
+    # Check for the sign in heading.
+    # about_heading = page.get_by_role("heading", name="About your service")
+    # expect(about_heading).to_be_visible()
+
+    # Retrieve some prominent elements on the page for testing.
+    # service_name_input = page.locator('xpath=//input[@name="name"]')
+    # add_service_button = page.get_by_role("button", name=re.compile("Add service"))
+
+    # expect(service_name_input).to_be_visible()
+    # expect(add_service_button).to_be_visible()
+
+    # Fill in the form.
+    # service_name_input.fill(new_service_name)
+
+    # Click on add service.
+    # add_service_button.click()
+
+    # Check to make sure that we've arrived at the next page.
+    # page.wait_for_load_state("domcontentloaded")
+    # check_axe_report(page)
+
+    # TODO this fails on staging due to duplicate results on 'get_by_text'
+    # Check for the service name title and heading.
+    # service_heading = page.get_by_text(new_service_name, exact=True)
+    # expect(service_heading).to_be_visible()
+
+    print("*" * 80)
+    print(default_service.name)
+    print("*" * 80)
+
+    expect(page).to_have_title(re.compile(default_service.name))
+
+    create_new_template(page)
+
+    update_template(page)
+
+    delete_template(page)
+
+    _teardown(page)
+
+
+def _teardown(page):
+    page.click("text='Settings'")
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    page.click("text='Delete this service'")
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    page.click("text='Yes, delete'")
+
+    # Check to make sure that we've arrived at the next page.
+    page.wait_for_load_state("domcontentloaded")
+    check_axe_report(page)
+
+    # Check to make sure that we've arrived at the next page.
+    # Check the page title exists and matches what we expect.
+    expect(page).to_have_title(re.compile("Choose service"))