test: add partner errors and import requests Response as RequestsResp…

…onse

test-engineering/contract-tests/client/tests/conftest.py

@@ -4,12 +4,18 @@
 
 import os
 import pathlib
-from typing import Dict, Any
+from typing import Dict, Union, Type
 
+import pytest
 import yaml
 
 from models import Records, Scenario, Service, Tiles
 
+SERVICE_MODEL: Dict[Service, Union[Type[Records], Type[Tiles]]] = {
+    Service.PARTNER: Records,
+    Service.CONTILE: Tiles,
+}
+
 
 def pytest_configure(config):
     """Load test scenarios from file."""
@@ -25,21 +31,18 @@ def pytest_configure(config):
 
     # Check that all 200 OK responses in test scenarios contain correct
     # information and FastAPI model instances were created for them.
-    SERVICE_MODEL: Dict[Service, Any] = {
-        Service.PARTNER: Records,
-        Service.CONTILE: Tiles,
-    }
-
     for scenario in config.contile_scenarios:
         for i, step in enumerate(scenario.steps):
 
             if step.response.status_code != 200:
                 continue
 
-            expected_model: Any = SERVICE_MODEL.get(step.request.service)
+            expected_model: Union[Type[Records], Type[Tiles]] = SERVICE_MODEL.get(
+                step.request.service
+            )
 
             if not isinstance(step.response.content, expected_model):
-                raise RuntimeError(
+                raise pytest.UsageError(
                     f"Failed to create {expected_model.__name__} "
                     f"model for '200 OK' response content in "
                     f"step {i} of scenario {scenario.name}"

test-engineering/contract-tests/client/tests/models.py

@@ -6,6 +6,7 @@
 from typing import Any, List, Optional, Union
 
 from pydantic import BaseModel, Extra
+from requests import Response as RequestsResponse
 
 
 class Service(Enum):
@@ -99,3 +100,20 @@ class Scenario(BaseModel):
     name: str
     description: str
     steps: List[Step]
+
+
+class PartnerError(Exception):
+    """Error specific to partner service interactions."""
+
+
+class PartnerRecordsNotClearedError(PartnerError):
+    """Error clearing partner records."""
+
+    def __init__(self, response: RequestsResponse):
+        error_message: str = (
+            f"The Partner records may not have cleared after the test execution.\n"
+            f"Response details:\n"
+            f"Status Code: {response.status_code}\n"
+            f"Content: '{response.text}'"
+        )
+        super().__init__(error_message)

test-engineering/contract-tests/client/tests/test_contile.py

@@ -7,9 +7,9 @@
 
 import pytest
 import requests
-from requests.models import Response
+from requests import Response as RequestsResponse
 
-from models import Step
+from models import PartnerRecordsNotClearedError, Step
 
 
 @pytest.fixture(name="hosts", scope="session")
@@ -29,16 +29,10 @@ def fixture_clear_partner_records(hosts: Dict[str, str]) -> Callable:
     partner_host = hosts["partner"]
 
     def clear_partner_records():
-        r: Response = requests.delete(f"{partner_host}/records/")
+        r: RequestsResponse = requests.delete(f"{partner_host}/records/")
 
         if r.status_code != 204:
-            error_message: str = (
-                f"The Partner records may not have cleared after the test execution.\n"
-                f"Response details:\n"
-                f"Status Code: {r.status_code}\n"
-                f"Content: '{r.text}'"
-            )
-            raise Exception(error_message)
+            raise PartnerRecordsNotClearedError(r)
 
     return clear_partner_records
 
@@ -65,7 +59,7 @@ def test_contile(hosts: Dict[str, str], steps: List[Step]):
             header.name: header.value for header in step.request.headers
         }
 
-        r: Response = requests.request(method, url, headers=headers)
+        r: RequestsResponse = requests.request(method, url, headers=headers)
 
         error_message: str = (
             f"Expected status code {step.response.status_code},\n"