Merge pull request #33 from foarsitter/development

feat: tests with mocked API responses

Author: foarsitter

noxfile.py

@@ -184,7 +184,7 @@ def coverage(session: Session) -> None:
 
 @session(python=python_versions[0])
 def typeguard(session: Session) -> None:
-    """Runtime type checking using Typeguard."""
+    """Runtime model checking using Typeguard."""
     session.install(".")
     session.install("pytest", "typeguard", "pygments")
     session.run("pytest", f"--typeguard-packages={package}", *session.posargs)

poetry.lock

@@ -1,16 +1,31 @@
 from json import JSONDecodeError
+from typing import Dict
 from typing import List
 from typing import Optional
-from typing import Union
+from typing import Type
+from typing import TypeVar
 
 import httpx
 from httpx import Request
 from httpx import Response
 
 from . import models
+from .errors import CheckedIDAuthenticationError
+from .errors import CheckedIDError
+from .errors import CheckedIDNotFoundError
+from .errors import CheckedIDValidationError
+
+
+_T = TypeVar("_T")
 
 
 class Client:
+    ERROR_RESPONSE_MAPPING: Dict[int, Type[CheckedIDError]] = {
+        422: CheckedIDValidationError,
+        403: CheckedIDAuthenticationError,
+        404: CheckedIDNotFoundError,
+    }
+
     def __init__(self, customer_code: str, base_url: str = "https://api.checkedid.eu/"):
         self.httpx = httpx.Client(base_url=base_url, auth=self.authenticate_request)
         self.access_token: Optional[str] = None
@@ -21,39 +36,43 @@ def authenticate_request(self, request: Request) -> Request:
             request.headers["Authorization"] = f"Bearer {self.access_token}"
         return request
 
+    def process_response(
+        self, response: Response, model: Type[_T], status_code_success: int = 200
+    ) -> Optional[_T]:
+        if response.status_code == status_code_success:
+            return model(**response.json())
+
+        self.handle_error_response(response)
+
+        return None
+
     def oauth_token(
         self, grant_type: str, username: str, password: str
-    ) -> Union[models.OAuthToken, models.ErrorResponse]:
+    ) -> Optional[models.OAuthToken]:
         response = self.httpx.post(
             "/oauth/token",
             data={"grant_type": grant_type, "username": username, "password": password},
         )
 
-        if response.status_code == 200:
-            typed_response = models.OAuthToken(**response.json())
+        typed_response = self.process_response(response, models.OAuthToken)
 
+        if typed_response:
             self.access_token = typed_response.access_token
 
             return typed_response
-        else:
-            return self.handle_error_response(response)
+        return None
 
-    def invitation_status(
-        self, invitation_code: str
-    ) -> Union[models.Invitation, models.ErrorResponse]:
+    def invitation_status(self, invitation_code: str) -> Optional[models.Invitation]:
         response: Response = self.httpx.get(
             f"/result/status/{invitation_code}",
             headers={"Accept": "application/json"},
         )
 
-        if response.status_code == 200:
-            return models.Invitation(**response.json())
-        else:
-            return self.handle_error_response(response)
+        return self.process_response(response, models.Invitation)
 
     def invitations_create(
         self, invitations: List[models.CreateInvitationRequest]
-    ) -> Union[models.CustomerDetails, models.ErrorResponse]:
+    ) -> Optional[models.CustomerDetails]:
         obj = models.CreateInvitationDetails(
             CustomerCode=self.customer_code, Invitations=invitations
         )
@@ -64,47 +83,53 @@ def invitations_create(
             headers={"Accept": "application/json", "Content-Type": "application/json"},
         )
 
-        if response.status_code == 200:
-            return models.CustomerDetails(**response.json())
-        else:
-            return self.handle_error_response(response)
+        return self.process_response(response, models.CustomerDetails)
 
-    def invitation_delete(
-        self, invitation_code: str
-    ) -> Union[models.ErrorResponse, bool]:
+    def invitation_delete(self, invitation_code: str) -> bool:
         response: Response = self.httpx.delete(
             f"/invitation/{self.customer_code}/{invitation_code}",
             headers={"Accept": "application/json"},
         )
 
         if response.status_code == 200:
             return True
-        else:
-            return self.handle_error_response(response)
 
-    def dossier(
-        self, dossier_number: str
-    ) -> Union[models.ReportResponse, models.ErrorResponse]:
+        self.handle_error_response(response)
+
+        return False
+
+    def dossier(self, dossier_number: str) -> Optional[models.ReportResponse]:
         response = self.httpx.get(f"/report/{dossier_number}")
 
-        if response.status_code == 200:
-            return models.ReportResponse(**response.json())
-        return self.handle_error_response(response)
+        return self.process_response(response, models.ReportResponse)
 
     def dossier_with_scope(
         self, dossier_number: str, scope: str
-    ) -> Union[models.ReportDataV3, models.ErrorResponse]:
+    ) -> Optional[models.ReportDataV3]:
         response = self.httpx.get(f"/reportdata/{dossier_number}/{scope}")
 
-        if response.status_code == 200:
-            return models.ReportDataV3(**response.json())
-        return self.handle_error_response(response)
+        return self.process_response(response, models.ReportDataV3)
+
+    def handle_error_response(self, response: Response) -> None:
+        if response.status_code == 400:
+            raise CheckedIDValidationError(
+                response.text, status_code=response.status_code
+            )
 
-    def handle_error_response(self, response: Response) -> models.ErrorResponse:
         try:
             json = response.json()
         except JSONDecodeError:
             json = {"message": response.text}
 
         json["status_code"] = response.status_code
-        return models.ErrorResponse(**json)
+
+        exception_type = self.map_exception(response)
+        raise exception_type(
+            status_code=response.status_code, json=json, message="Error from server"
+        )
+
+    def map_exception(self, response: Response) -> Type[CheckedIDError]:
+        exception_type = self.ERROR_RESPONSE_MAPPING.get(
+            response.status_code, CheckedIDError
+        )
+        return exception_type

src/checkedid/client.py

@@ -0,0 +1,31 @@
+from typing import Any
+from typing import Dict
+from typing import Optional
+
+
+class CheckedIDError(Exception):
+    status_code: int
+    json: Optional[Dict[str, Any]] = None
+
+    def __init__(
+        self,
+        message: str,
+        status_code: int,
+        json: Optional[Dict[str, Any]] = None,
+        *args: str
+    ):
+        super().__init__(message, *args)
+        self.status_code = status_code
+        self.json = json
+
+
+class CheckedIDValidationError(CheckedIDError):
+    pass
+
+
+class CheckedIDNotFoundError(CheckedIDError):
+    pass
+
+
+class CheckedIDAuthenticationError(CheckedIDError):
+    pass

src/checkedid/errors.py

@@ -1,6 +1,7 @@
 import os
 
 import pytest
+from httpx import Response
 
 from checkedid.client import Client
 from checkedid.models import ErrorResponse
@@ -22,8 +23,23 @@ def client(customer_code) -> Client:
     return client
 
 
+@pytest.fixture
+def access_token_mock(respx_mock):
+    respx_mock.post("").mock(
+        return_value=Response(
+            status_code=200,
+            json={
+                "access_token": "abc",
+                "expires_in": 3600,
+                "token_type": "Bearer",
+                "refresh_token": "def",
+            },
+        )
+    )
+
+
 @pytest.fixture()
-def auth_client(client: Client) -> Client:
+def auth_client(client: Client, access_token_mock) -> Client:
     response = client.oauth_token(
         "password", os.getenv("CHECKEDID_USERNAME"), os.getenv("CHECKEDID_PASSWORD")
     )

tests/conftest.py

@@ -1,28 +1,47 @@
+import pytest
 from httpx import Response
 
+from checkedid import errors
 
-def test_dossier(auth_client):
-    response = auth_client.dossier("100029-0000031")
 
-    assert response.DossierNumber == "100029-0000031"
+def test_dossier(auth_client, respx_mock):
+    dossier_number = "999999-8888800"
+    respx_mock.get("").mock(
+        return_value=Response(
+            status_code=200, json={"DossierNumber": dossier_number, "ReportPDF": ""}
+        )
+    )
+    response = auth_client.dossier(dossier_number)
+
+    assert response.DossierNumber == dossier_number
 
 
 def test_dossier_with_error(auth_client, respx_mock):
     respx_mock.get("").mock(return_value=Response(status_code=404))
-    response = auth_client.dossier("does-not-exist")
-
-    assert response.status_code == 404
-
-
-def test_dossier_with_scope(auth_client):
+    with pytest.raises(errors.CheckedIDError):
+        auth_client.dossier("does-not-exist")
+
+
+def test_dossier_with_scope(auth_client, respx_mock):
+    dossier_number = "999999-8888800"
+    respx_mock.get("").mock(
+        return_value=Response(
+            status_code=200,
+            json={
+                "DossierNumber": dossier_number,
+                "ReportPDF": "",
+                "Authority": "Burg. van Groningen",
+            },
+        )
+    )
     response = auth_client.dossier_with_scope("100029-0000031", "10")
 
-    assert response.DossierNumber == "100029-0000031"
+    assert response.DossierNumber == dossier_number
     assert response.Authority == "Burg. van Groningen"
 
 
 def test_dossier_with_scope_with_error(auth_client, respx_mock):
     respx_mock.get("").mock(return_value=Response(status_code=404))
-    response = auth_client.dossier_with_scope("does-not-exist", "10")
 
-    assert response.status_code == 404
+    with pytest.raises(errors.CheckedIDError):
+        auth_client.dossier_with_scope("does-not-exist", "10")

tests/test_dossiers.py

@@ -1,14 +1,27 @@
+import pytest
 from httpx import Response
 
+from checkedid import errors
 
-def test_invitation_status(auth_client):
-    response = auth_client.invitation_status("15IMN4")
+
+def test_invitation_status(auth_client, respx_mock):
+    invitation_code = "4ZCNXF"
+
+    respx_mock.get("").mock(
+        return_value=Response(
+            status_code=200,
+            json={"CustomerCode": 100029, "InvitationCode": invitation_code},
+        )
+    )
+
+    response = auth_client.invitation_status(invitation_code)
 
     assert response.CustomerCode == 100029
+    assert response.InvitationCode == invitation_code
 
 
 def test_invitation_status_not_found(auth_client, respx_mock):
     respx_mock.get("").mock(return_value=Response(status_code=404))
-    response = auth_client.invitation_status("15IMN4")
 
-    assert response.status_code == 404
+    with pytest.raises(errors.CheckedIDNotFoundError):
+        auth_client.invitation_status("15IMN4")

tests/test_invitation_status.py

@@ -1,7 +1,14 @@
+import pytest
+from httpx import Response
+
+from checkedid import errors
+from checkedid.models import CreateInvitationDetails
 from checkedid.models import CreateInvitationRequest
 
 
-def test_invitations_create_and_delete(auth_client, employee_code):
+def test_invitations_create_and_delete(
+    auth_client, employee_code, respx_mock, customer_code
+):
     details: CreateInvitationRequest = CreateInvitationRequest.construct()
     details.EmployeeCode = employee_code
     details.InviteeEmail = "[email protected]"
@@ -11,6 +18,17 @@ def test_invitations_create_and_delete(auth_client, employee_code):
     details.AppFlow = "10"
     details.PreferredLanguage = "nl"
 
+    respx_mock.post("").mock(
+        return_value=Response(
+            status_code=200,
+            json=CreateInvitationDetails(
+                CustomerCode=customer_code, Invitations=[details]
+            ).dict(),
+        )
+    )
+
+    respx_mock.delete("").mock(return_value=Response(status_code=200))
+
     response = auth_client.invitations_create([details])
 
     assert response.CustomerCode == 100029
@@ -19,14 +37,14 @@ def test_invitations_create_and_delete(auth_client, employee_code):
     assert auth_client.invitation_delete(response.Invitations[0].InvitationCode) is True
 
 
-def test_invitations_create_with_error(auth_client):
-    response = auth_client.invitations_create([CreateInvitationRequest.construct()])
-
-    assert response.status_code == 422
-    assert len(response.Errors) == 4
+def test_invitations_create_with_error(auth_client, respx_mock):
+    respx_mock.post("").mock(return_value=Response(status_code=422))
 
+    with pytest.raises(errors.CheckedIDValidationError):
+        auth_client.invitations_create([CreateInvitationRequest.construct()])
 
-def test_invitation_delete_with_error(auth_client):
-    response = auth_client.invitation_delete("xyz")
 
-    assert response.status_code == 404
+def test_invitation_delete_with_error(auth_client, respx_mock):
+    respx_mock.delete("").mock(return_value=Response(status_code=404))
+    with pytest.raises(errors.CheckedIDNotFoundError):
+        auth_client.invitation_delete("xyz")