local smoke tests

.devcontainer/scripts/notify-dev-entrypoint.sh

@@ -25,6 +25,7 @@ echo -e "complete -F __start_kubectl k" >> ~/.zshrc
 
 # Smoke test
 # requires adding files .env_staging and .env_prod to the root of the project
+echo -e "alias smoke-local='cd /workspace && cp .env_smoke_local tests_smoke/.env && poetry run make smoke-test-local'" >> ~/.zshrc
 echo -e "alias smoke-staging='cd /workspace && cp .env_smoke_staging tests_smoke/.env && poetry run make smoke-test'" >> ~/.zshrc
 echo -e "alias smoke-prod='cd /workspace && cp .env_smoke_prod tests_smoke/.env && poetry run make smoke-test'" >> ~/.zshrc
 

Makefile

@@ -46,6 +46,10 @@ format:
 smoke-test:
 	cd tests_smoke && poetry run python smoke_test.py
 
+.PHONY: smoke-test-local
+smoke-test-local:
+	cd tests_smoke && poetry run python smoke_test.py --local --nofiles
+
 .PHONY: run
 run: ## Run the web app
 	poetry run flask run -p 6011 --host=0.0.0.0

tests_smoke/README.md

@@ -0,0 +1,14 @@
+# Smoke Tests
+
+This repository contains a set of smoke tests for our application. Smoke testing, also known as "Build Verification Testing", is a type of software testing that comprises of a non-exhaustive set of tests that aim at ensuring that the most important functions work. The phrase 'smoke testing' comes from the hardware testing, where you plug in a new piece of hardware and turn it on for the first time. If it starts smoking, you know you have a problem.
+
+## Getting Started
+
+These smoke tests are designed to run in the api devcontainer.
+
+in the root of the repo create `.env` files for the environments you with to smoke test, for example `.env_smoke_local`, `.env_smoke_staging`, and `.env_smoke_prod`. For required values see the [.env.example](.env.example) file).
+
+## Running the tests
+
+in the devcontainer run the aliases `smoke-local`, `smoke-staging`, or `smoke-prod` to run the tests.
+

tests_smoke/smoke/common.py

@@ -2,8 +2,6 @@
 import json
 import os
 import time
-
-# from notifications_utils.s3 import s3upload as utils_s3upload
 import urllib
 import uuid
 from enum import Enum
@@ -41,6 +39,7 @@ class Config:
     EMAIL_TEMPLATE_ID = os.environ.get("SMOKE_EMAIL_TEMPLATE_ID")
     SMS_TEMPLATE_ID = os.environ.get("SMOKE_SMS_TEMPLATE_ID")
     API_KEY = os.environ.get("SMOKE_API_KEY", "")
+    JOB_SIZE = int(os.environ.get("SMOKE_JOB_SIZE", 2))
 
 
 boto_session = Session(
@@ -67,8 +66,8 @@ def rows_to_csv(rows: List[List[str]]):
     return output.getvalue()
 
 
-def job_line(data: str, number_of_lines: int) -> Iterator[List[str]]:
-    return map(lambda n: [data, f"var{n}"], range(0, number_of_lines))
+def job_line(data: str, number_of_lines: int, prefix: str = "") -> Iterator[List[str]]:
+    return map(lambda n: [data, f"{prefix} {n}"], range(0, number_of_lines))
 
 
 def pretty_print(data: Any):
@@ -120,7 +119,6 @@ def job_succeeded(service_id: str, job_id: str) -> bool:
     return success
 
 
-# from notifications_utils.s3 import s3upload as utils_s3upload
 def utils_s3upload(filedata, region, bucket_name, file_location, content_type="binary/octet-stream", tags=None):
     _s3 = boto_session.resource("s3")
 

tests_smoke/smoke/test_admin_csv.py

@@ -13,13 +13,13 @@
 )
 
 
-def test_admin_csv(notification_type: Notification_type):
+def test_admin_csv(notification_type: Notification_type, local: bool = False):
     print(f"test_admin_csv ({notification_type.value})... ", end="", flush=True)
 
     if notification_type == Notification_type.EMAIL:
-        data = rows_to_csv([["email address", "var"], *job_line(Config.EMAIL_TO, 2)])
+        data = rows_to_csv([["email address", "var"], *job_line(Config.EMAIL_TO, Config.JOB_SIZE, prefix="smoke test admin csv")])
     else:
-        data = rows_to_csv([["phone number", "var"], *job_line(Config.SMS_TO, 2)])
+        data = rows_to_csv([["phone number", "var"], *job_line(Config.SMS_TO, Config.JOB_SIZE, prefix="smoke test admin csv")])
 
     upload_id = s3upload(Config.SERVICE_ID, data)
     metadata_kwargs = {
@@ -42,8 +42,11 @@ def test_admin_csv(notification_type: Notification_type):
         print("FAILED: post to send_notification failed")
         exit(1)
 
-    success = job_succeeded(Config.SERVICE_ID, upload_id)
-    if not success:
-        print("FAILED: job didn't finish successfully")
-        exit(1)
-    print("Success")
+    if local:
+        print(f"Check manually for {Config.JOB_SIZE} {notification_type.value}s")
+    else:
+        success = job_succeeded(Config.SERVICE_ID, upload_id)
+        if not success:
+            print("FAILED: job didn't finish successfully")
+            exit(1)
+        print("Success")

tests_smoke/smoke/test_admin_one_off.py

@@ -4,7 +4,7 @@
 from .common import Config, Notification_type, pretty_print, single_succeeded
 
 
-def test_admin_one_off(notification_type: Notification_type):
+def test_admin_one_off(notification_type: Notification_type, local: bool = False):
     print(f"test_admin_one_off ({notification_type.value})... ", end="", flush=True)
 
     token = create_jwt_token(Config.ADMIN_CLIENT_SECRET, client_id=Config.ADMIN_CLIENT_USER_NAME)
@@ -17,7 +17,7 @@ def test_admin_one_off(notification_type: Notification_type):
             "to": to,
             "template_id": template_id,
             "created_by": Config.USER_ID,
-            "personalisation": {"var": "var"},
+            "personalisation": {"var": "smoke test admin one off"},
         },
         headers={"Authorization": f"Bearer {token}"},
     )
@@ -28,9 +28,12 @@ def test_admin_one_off(notification_type: Notification_type):
         print("FAILED: post to send_notification failed")
         exit(1)
 
-    uri = f"{Config.API_HOST_NAME}/service/{Config.SERVICE_ID}/notifications/{body['id']}"
-    success = single_succeeded(uri, use_jwt=True)
-    if not success:
-        print("FAILED: job didn't finish successfully")
-        exit(1)
-    print("Success")
+    if local:
+        print(f"Check manually for 1 {notification_type.value}")
+    else:
+        uri = f"{Config.API_HOST_NAME}/service/{Config.SERVICE_ID}/notifications/{body['id']}"
+        success = single_succeeded(uri, use_jwt=True)
+        if not success:
+            print("FAILED: job didn't finish successfully")
+            exit(1)
+        print("Success")

tests_smoke/smoke/test_api_bulk.py

@@ -12,7 +12,7 @@
 )
 
 
-def test_api_bulk(notification_type: Notification_type):
+def test_api_bulk(notification_type: Notification_type, local: bool = False):
     print(f"test_api_bulk ({notification_type.value})... ", end="", flush=True)
     template_id = Config.EMAIL_TEMPLATE_ID if notification_type == Notification_type.EMAIL else Config.SMS_TEMPLATE_ID
     to = Config.EMAIL_TO if notification_type == Notification_type.EMAIL else Config.SMS_TO
@@ -23,7 +23,7 @@ def test_api_bulk(notification_type: Notification_type):
         json={
             "name": f"My bulk name {datetime.utcnow().isoformat()}",
             "template_id": template_id,
-            "csv": rows_to_csv([[header, "var"], *job_line(to, 2)]),
+            "csv": rows_to_csv([[header, "var"], *job_line(to, Config.JOB_SIZE, prefix="smoke test api bulk")]),
         },
         headers={"Authorization": f"ApiKey-v1 {Config.API_KEY}"},
     )
@@ -32,8 +32,11 @@ def test_api_bulk(notification_type: Notification_type):
         print("FAILED: post failed")
         exit(1)
 
-    success = job_succeeded(Config.SERVICE_ID, response.json()["data"]["id"])
-    if not success:
-        print("FAILED: job didn't finish successfully")
-        exit(1)
-    print("Success")
+    if local:
+        print(f"Check manually for {Config.JOB_SIZE} {notification_type.value}s")
+    else:
+        success = job_succeeded(Config.SERVICE_ID, response.json()["data"]["id"])
+        if not success:
+            print("FAILED: job didn't finish successfully")
+            exit(1)
+        print("Success")

tests_smoke/smoke/test_api_one_off.py

@@ -11,7 +11,7 @@
 )
 
 
-def test_api_one_off(notification_type: Notification_type, attachment_type: Attachment_type = Attachment_type.NONE):
+def test_api_one_off(notification_type: Notification_type, attachment_type: Attachment_type = Attachment_type.NONE, local: bool = False):
     if attachment_type is Attachment_type.NONE:
         print(f"test_api_oneoff ({notification_type.value})... ", end="", flush=True)
     else:
@@ -51,7 +51,7 @@ def test_api_one_off(notification_type: Notification_type, attachment_type: Atta
         }
     else:
         data["personalisation"] = {
-            "var": "var",
+            "var": "smoke test api one off",
         }
 
     response = requests.post(
@@ -64,10 +64,12 @@ def test_api_one_off(notification_type: Notification_type, attachment_type: Atta
         print(f"FAILED: post to v2/notifications/{notification_type.value} failed")
         exit(1)
 
-    uri = response.json()["uri"]
-
-    success = single_succeeded(uri, use_jwt=False)
-    if not success:
-        print("FAILED: job didn't finish successfully")
-        exit(1)
-    print("Success")
+    if local:
+        print(f"Check manually for 1 {notification_type.value}")
+    else:
+        uri = response.json()["uri"]
+        success = single_succeeded(uri, use_jwt=False)
+        if not success:
+            print("FAILED: job didn't finish successfully")
+            exit(1)
+        print("Success")

tests_smoke/smoke_test.py

@@ -1,19 +1,28 @@
+import argparse
+
 from smoke.common import Attachment_type, Config, Notification_type  # type: ignore
 from smoke.test_admin_csv import test_admin_csv  # type: ignore
 from smoke.test_admin_one_off import test_admin_one_off  # type: ignore
 from smoke.test_api_bulk import test_api_bulk  # type: ignore
 from smoke.test_api_one_off import test_api_one_off  # type: ignore
 
 if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-l", "--local", default=False, action='store_true', help="run locally, do not check for delivery success (default false)")
+    parser.add_argument("--nofiles", default=False, action='store_true', help="do not send files (default false)")
+    args = parser.parse_args()
+
     print("API Smoke test\n")
     for key in ["API_HOST_NAME", "SERVICE_ID", "EMAIL_TEMPLATE_ID", "SMS_TEMPLATE_ID", "EMAIL_TO", "SMS_TO"]:
         print(f"{key:>17}: {Config.__dict__[key]}")
     print("")
 
     for notification_type in [Notification_type.EMAIL, Notification_type.SMS]:
-        test_admin_one_off(notification_type)
-        test_admin_csv(notification_type)
-        test_api_one_off(notification_type)
-        test_api_bulk(notification_type)
-    test_api_one_off(Notification_type.EMAIL, Attachment_type.ATTACHED)
-    test_api_one_off(Notification_type.EMAIL, Attachment_type.LINK)
+        test_admin_one_off(notification_type, local=args.local)
+        test_admin_csv(notification_type, local=args.local)
+        test_api_one_off(notification_type, local=args.local)
+        test_api_bulk(notification_type, local=args.local)
+
+    if not args.nofiles:
+        test_api_one_off(Notification_type.EMAIL, attachment_type=Attachment_type.ATTACHED, local=args.local)
+        test_api_one_off(Notification_type.EMAIL, attachment_type=Attachment_type.LINK, local=args.local)