APM-6136 re-write python tests using apim pytest library

Author: sophieclayton12-nhs

e2e/tests/api_tests.py

@@ -49,88 +49,67 @@ def test_check_status_is_secured(nhsd_apim_proxy_url):
     resp = requests.get(f"{nhsd_apim_proxy_url}/_status")
     assert resp.status_code == 401
 
-# @pytest.mark.e2e
-# @pytest.mark.smoketest
-# @pytest.mark.asyncio
-# async def test_wait_for_status(
-#         api_client: APISessionClient, api_test_config: APITestSessionConfig
-# ):
-#     async def is_deployed(resp: ClientResponse):
-#         if resp.status != 200:
-#             return False
-#         body = await resp.json()
-#
-#         if body.get("commitId") != api_test_config.commit_id:
-#             return False
-#
-#         backend = dict_path(body, ["checks", "healthcheck", "outcome", "version"])
-#
-#         if type(backend) != dict:
-#             return False
-#
-#         return backend.get("commitId") == api_test_config.commit_id
-#
-#     await poll_until(
-#         make_request=lambda: api_client.get(
-#             "_status", headers={"apikey": env.status_endpoint_api_key()}
-#         ),
-#         until=is_deployed,
-#         timeout=120,
-#     )
-#
-#
-# @pytest.mark.e2e
-# @pytest.mark.asyncio
-# async def test_api_status_with_service_header_another_service(api_client: APISessionClient):
-#
-#     r = await api_client.get(
-#         "_status", allow_retries=True, max_retries=5,
-#         headers={
-#             'x-apim-service': 'async-slowapp',
-#             'apikey': env.status_endpoint_api_key()
-#         }
-#     )
-#     assert r.status == 200, (r.status, r.reason, (await r.text())[:2000])
-#     body = await r.json()
-#
-#     service = dict_path(body, ["checks", "healthcheck", "outcome", "service"])
-#
-#     assert service == 'sync-wrap'
-#
-#
-# @pytest.mark.e2e
-# @pytest.mark.asyncio
-# async def test_api_status_with_service_header(api_client: APISessionClient):
-#
-#     r = await api_client.get(
-#         "_status", allow_retries=True, max_retries=5,
-#         headers={
-#             'x-apim-service': 'sync-wrap',
-#             'apikey': env.status_endpoint_api_key()
-#         }
-#     )
-#     assert r.status == 200, (r.status, r.reason, (await r.text())[:2000])
-#     body = await r.json()
-#
-#     service = dict_path(body, ["checks", "healthcheck", "outcome", "service"])
-#
-#     assert service == 'sync-wrap'
-#
-#
-# @pytest.mark.e2e
-# @pytest.mark.asyncio
-# async def test_api_slowapp_slower_than_sync_wait(api_client: APISessionClient):
-#
-#     r = await api_client.get(
-#         "async-slowapp/slow?delay=5", allow_retries=True, max_retries=5, headers={'x-sync-wait': '0.25'}
-#     )
-#     assert r.status == 504, (r.status, r.reason, (await r.text())[:2000])
-#
-#
-# @pytest.mark.e2e
-# @pytest.mark.asyncio
-# async def test_api_slowapp_responds_test_final_status(api_client: APISessionClient):
-#
-#     r = await api_client.get("async-slowapp/slow?final_status=418&complete_in=0.5", allow_retries=True, max_retries=5)
-#     assert r.status == 418, (r.status, r.reason, (await r.text())[:2000])
-#     assert r.reason == "I'm a Teapot"
+
+@pytest.mark.e2e
+@pytest.mark.smoketest
+@pytest.mark.parametrize("service_header", ["", "async-slowapp", "sync-wrap"])
+def test_wait_for_status(nhsd_apim_proxy_url, status_endpoint_auth_headers, service_header):
+    def _container_not_ready(resp: requests.Response):
+        """
+        Requests to ECS containers which are still starting up return with a
+        HTTP 503 (service unavailable).
+        """
+        return resp.json().get("checks", {}) \
+            .get("healthcheck", {}) \
+            .get("responseCode") == 503
+
+    retries = 0
+    headers = status_endpoint_auth_headers
+    if service_header:
+        headers["x-apim-service"] = service_header
+    resp = requests.get(
+        f"{nhsd_apim_proxy_url}/_status", headers=headers, timeout=30
+    )
+
+    if resp.status_code != 200:
+        # Status should always be 200 we don't need to wait for it
+        pytest.fail(f"Status code {resp.status_code}, expecting 200")
+    if not resp.json().get("version"):
+        pytest.fail("version not found")
+
+    deployed_commit_id = resp.json().get("commitId")
+
+    while deployed_commit_id != getenv("SOURCE_COMMIT_ID") or _container_not_ready(resp) and retries <= 45:
+        resp = requests.get(
+            f"{nhsd_apim_proxy_url}/_status", headers=status_endpoint_auth_headers, timeout=30
+        )
+
+        deployed_commit_id = resp.json().get("commitId")
+        retries += 1
+        sleep(1)
+
+    if retries >= 45:
+        pytest.fail("Timeout Error - max retries")
+
+    body = resp.json()
+    assert body.get("status") == "pass"
+
+    service = dict_path(body, ["checks", "healthcheck", "outcome", "service"])
+    assert service == 'sync-wrap'
+
+
+@pytest.mark.e2e
+def test_api_slowapp_slower_than_sync_wait(nhsd_apim_proxy_url):
+    resp = requests.get(
+        f"{nhsd_apim_proxy_url}/async-slowapp/slow?delay=5", headers={'x-sync-wait': '0.25'}, timeout=45
+    )
+    assert resp.status_code == 504, (resp.status_code, resp.reason, resp.text[:2000])
+
+
+@pytest.mark.e2e
+def test_api_slowapp_responds_test_final_status(nhsd_apim_proxy_url):
+    resp = requests.get(
+        f"{nhsd_apim_proxy_url}/async-slowapp/slow?final_status=418&complete_in=0.5", timeout=30
+    )
+    assert resp.status_code == 418, (resp.status_code, resp.reason, resp.text[:2000])
+    assert resp.reason == "I'm a Teapot"

poetry.lock

@@ -28,7 +28,6 @@ requests = "^2.32.3"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^8.3.5"
-pytest-asyncio = "^0.26.0"
 coverage = "^7.8.0"
 flake8 = "^7.2.0"
 black = "^25.1.0"