WOR-1120 Add smoke tests to landing zone service

.gitignore

@@ -46,3 +46,6 @@ gha-creds-*.json
 
 ### jEnv
 .java-version
+
+### Python
+**/__pycache__/

smoke_tests/README.md

@@ -0,0 +1,31 @@
+
+
+## Purpose
+To provide a low overhead way to check that Landing Zone service is basically operational in a given environment.
+
+These tests are intended to be run either as the final step in the deployment check list for releasing the service independently, or on an ad-hoc basis if needed.
+They are currently set up to be run manually, but could be automated in the future.
+
+## Scope
+Verifying _basic_ functionality, based on arguments passed:
+* The status endpoint returns 200, and all subsystems specified by the service are 'OK'
+* The version endpoint returns 200, and version payload exists
+* If a user access token is passed, the following endpoints will be called:
+  * `GET /api/landingzones/definitions/v1/azure` to verify available landing zone definitions, and checked that it returns 200.
+  * `GET api/landingzones/v1/azure`  to get all landing zones available to user, and checked that it returns 200.
+    This provides basic verification that the database connection is intact, even if the user has no landing zones available.
+
+
+## Setup
+* From the smoke_tests directory
+* Install [poetry](https://python-poetry.org) or `brew install poetry` on a mac
+* Install dependencies: `poetry install`
+* Get a shell in the created venv to run the scripts: `poetry shell`
+* Run the test: `python smoke_tests.py <args>`
+* For full usage information: `python smoke_tests.py -h`
+
+
+## Running as part of post-deployment steps:
+* Do setup as described above above
+* Get a user access token. Eg: `gcloud auth print-access-token`
+* Run script in poetry shell: `python smoke_tests.py "${LZ_HOST}" "${USER_ACCESS_TOKEN}"`

smoke_tests/pyproject.toml

@@ -0,0 +1,20 @@
+[tool.poetry]
+name = "landingzone-smoke-tests"
+version = "0.1.0"
+description = "Verify basic LZ functionality for a given environment"
+authors = ["DSP Workspaces <[email protected]>"]
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.11"
+requests = "^2.31.0"
+google-auth = "^2.19.1"
+
+[tool.poetry.group.dev.dependencies]
+black = {extras = ["d"], version = "^23.3.0"}
+pdbpp = "^0.10.3"
+poetry-lock-package = "^0.5.0"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

smoke_tests/smoke_tests.py

@@ -0,0 +1,88 @@
+import argparse
+import sys
+import unittest
+from unittest import TestSuite
+import requests
+
+from tests.smoke_test_case import SmokeTestCase
+from tests.unauthenticated.status_tests import StatusTests
+from tests.unauthenticated.version_tests import VersionTests
+from tests.authenticated.landingzone_definitions_tests import LandingZoneDefinitionsTests
+from tests.authenticated.landingzone_list_tests import LandingZoneListTests
+
+"""
+Landing Zone Smoke Test
+Enter the host (domain and optional port) of the Landing Zone instance you want to to test. 
+This test will ensure that the Landing Zone instance running on that host is minimally functional.
+"""
+
+
+def gather_tests(is_authenticated: bool = False) -> TestSuite:
+    suite = unittest.TestSuite()
+    status_tests = unittest.defaultTestLoader.loadTestsFromTestCase(StatusTests)
+    suite.addTests(status_tests)
+    version_tests = unittest.defaultTestLoader.loadTestsFromTestCase(VersionTests)
+    suite.addTests(version_tests)
+    if is_authenticated:
+        lz_definitions_tests = unittest.defaultTestLoader.loadTestsFromTestCase(LandingZoneDefinitionsTests)
+        suite.addTests(lz_definitions_tests)
+        lz_list_tests = unittest.defaultTestLoader.loadTestsFromTestCase(LandingZoneListTests)
+        suite.addTests(lz_list_tests)
+    return suite
+
+
+def main(main_args):
+    SmokeTestCase.LZ_HOST = main_args.lz_host
+    SmokeTestCase.USER_TOKEN = main_args.user_token
+
+    valid_user_token = main_args.user_token is not None and verify_user_token(main_args.user_token)
+    test_suite = gather_tests(valid_user_token)
+
+    runner = unittest.TextTestRunner(verbosity=main_args.verbosity)
+    runner.run(test_suite)
+
+
+def verify_user_token(user_token: str) -> bool:
+    response = requests.get(f"https://www.googleapis.com/oauth2/v1/tokeninfo?access_token={user_token}")
+    assert response.status_code == 200, "User Token is no longer valid.  Please generate a new token and try again."
+    return True
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        prog="Landing Zone Smoke Tests",
+        description=__doc__,
+    )
+    parser.add_argument(
+        "lz_host",
+        type=str,
+        help="Required domain with optional port number of the LandingZone host you want to test"
+    )
+    parser.add_argument(
+        "user_token",
+        nargs='?',
+        default=None,
+        type=str,
+        help="Optional. If present, will test additional authenticated endpoints using the specified token"
+    )
+    parser.add_argument(
+        "-v",
+        "--verbosity",
+        type=int,
+        choices=[0, 1, 2],
+        default=1,
+        help="""Python unittest verbosity setting: 
+    0: Quiet - Prints only number of tests executed
+    1: Minimal - (default) Prints number of tests executed plus a dot for each success and an F for each failure
+    2: Verbose - Help string and its result will be printed for each test"""
+    )
+    parsed_args = parser.parse_args()
+    # Need to pop off sys.argv values to avoid messing with args passed to unittest.main()
+    for _ in range(len(sys.argv[1:])):
+        sys.argv.pop()
+    return parsed_args
+
+
+if __name__ == "__main__":
+    args = parse_args()
+    main(args)

smoke_tests/tests/authenticated/landingzone_definitions_tests.py

@@ -0,0 +1,25 @@
+import json
+
+from tests.smoke_test_case import SmokeTestCase
+
+"""Validates landing zone definitions endpoint"""
+
+
+class LandingZoneDefinitionsTests(SmokeTestCase):
+
+    @staticmethod
+    def status_url() -> str:
+        return SmokeTestCase.build_lz_url("/api/landingzones/definitions/v1/azure")
+
+    def test_status_code_is_401_when_no_token_provided(self):
+        """Validates that definitions endpoint return 401 when no token provided"""
+        response = SmokeTestCase.call_lz(self.status_url())
+        self.assertEqual(response.status_code, 401)
+
+    def test_definitions(self):
+        """Validates lz definition response"""
+        response = SmokeTestCase.call_lz(self.status_url(), user_token=SmokeTestCase.USER_TOKEN)
+        self.assertEqual(response.status_code, 200)
+        lz_def_response = json.loads(response.text)
+        self.assertIsNotNone(lz_def_response["landingzones"], "landingzones property is not defined.")
+        self.assertEqual(len(lz_def_response["landingzones"]), 2, "unexpected number of landing zone definitions")

smoke_tests/tests/authenticated/landingzone_list_tests.py

@@ -0,0 +1,19 @@
+from tests.smoke_test_case import SmokeTestCase
+
+"""Validates landing zone endpoint which returns all landing zones available to user."""
+
+
+class LandingZoneListTests(SmokeTestCase):
+
+    @staticmethod
+    def status_url() -> str:
+        return SmokeTestCase.build_lz_url("/api/landingzones/v1/azure")
+
+    def test_status_code_is_401_when_no_token_provided(self):
+        """Validates that lz create endpoint return 401 when no payload provided"""
+        response = SmokeTestCase.call_lz(self.status_url())
+        self.assertEqual(response.status_code, 401)
+
+    def test_available_landingzones(self):
+        response = SmokeTestCase.call_lz(self.status_url(), user_token=SmokeTestCase.USER_TOKEN)
+        self.assertEqual(response.status_code, 200)

smoke_tests/tests/smoke_test_case.py

@@ -0,0 +1,25 @@
+import re
+from unittest import TestCase
+from urllib.parse import urljoin
+
+import requests
+from requests import Response
+
+
+class SmokeTestCase(TestCase):
+    LZ_HOST = None
+    USER_TOKEN = None
+
+    @staticmethod
+    def build_lz_url(path: str) -> str:
+        assert SmokeTestCase.LZ_HOST, "ERROR - BPMSmokeTests.LZ_HOST not properly set"
+        if re.match(r"^\s*https?://", SmokeTestCase.LZ_HOST):
+            return urljoin(SmokeTestCase.LZ_HOST, path)
+        else:
+            return urljoin(f"https://{SmokeTestCase.LZ_HOST}", path)
+
+    @staticmethod
+    def call_lz(url: str, params: dict = None, user_token: str = None) -> Response:
+        """Function is memoized so that we only make the call once"""
+        headers = {"Authorization": f"Bearer {user_token}"} if user_token else {}
+        return requests.get(url, params=params, headers=headers)

smoke_tests/tests/unauthenticated/status_tests.py

@@ -0,0 +1,21 @@
+import json
+
+from tests.smoke_test_case import SmokeTestCase
+
+"""Validates 'status' endpoint"""
+
+
+class StatusTests(SmokeTestCase):
+    @staticmethod
+    def status_url() -> str:
+        return SmokeTestCase.build_lz_url("/status")
+
+    def test_status_code_is_200(self):
+        response = SmokeTestCase.call_lz(self.status_url())
+        self.assertEqual(response.status_code, 200)
+
+    def test_subsystems(self):
+        response = SmokeTestCase.call_lz(self.status_url())
+        status = json.loads(response.text)
+        for system in status["systems"]:
+            self.assertEqual(status["systems"][system]["ok"], True, f"{system} is not OK")

smoke_tests/tests/unauthenticated/version_tests.py

@@ -0,0 +1,23 @@
+import json
+
+from tests.smoke_test_case import SmokeTestCase
+
+"""Validates 'version' endpoint"""
+
+
+class VersionTests(SmokeTestCase):
+    @staticmethod
+    def status_url() -> str:
+        return SmokeTestCase.build_lz_url("/version")
+
+    def test_status_code_is_200(self):
+        response = SmokeTestCase.call_lz(self.status_url())
+        self.assertEqual(response.status_code, 200)
+
+    def test_version_details(self):
+        response = SmokeTestCase.call_lz(self.status_url())
+        version_details = json.loads(response.text)
+        self.assertIsNotNone(version_details["gitTag"], "gitTag is not defined.")
+        self.assertIsNotNone(version_details["gitHash"], "gitHash is not defined.")
+        self.assertIsNotNone(version_details["github"], "github is not defined")
+        self.assertIsNotNone(version_details["build"], "build is not defined")