diff --git a/.github/workflows/stage_ci.yml b/.github/workflows/stage_ci.yml index 44965ca..7093d3e 100644 --- a/.github/workflows/stage_ci.yml +++ b/.github/workflows/stage_ci.yml @@ -23,8 +23,7 @@ jobs: uses: ./.github/actions - name: Run tests - run: | - poetry run pytest + run: make test pre-commit: runs-on: ubuntu-latest diff --git a/.gitignore b/.gitignore index a25237b..78d5341 100644 --- a/.gitignore +++ b/.gitignore @@ -52,10 +52,11 @@ coverage.xml *.mo *.pot -# Django stuff: +# Miscellaneuos configuration stuff: *.log local_settings.py db.sqlite3 +*.db # Flask stuff: instance/ diff --git a/Makefile b/Makefile index 2c1fd79..b42d702 100644 --- a/Makefile +++ b/Makefile @@ -11,7 +11,10 @@ pip-install: pip install --prefer-binary --use-pep517 --check-build-dependencies .[dev] test: - pytest -s xcov19/tests/ + APP_ENV=test APP_DB_ENGINE_URL="sqlite+aiosqlite://" pytest -s xcov19/tests/ -m "not integration" + +test-integration: + APP_ENV=test APP_DB_ENGINE_URL="sqlite+aiosqlite://" pytest -s xcov19/tests/ -m "integration" todos: @grep -rn "TODO:" xcov19/ --exclude-dir=node_modules --include="*.py" \ No newline at end of file diff --git a/poetry.lock b/poetry.lock index 4eae204..a752ed1 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand. +# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand. [[package]] name = "aiosqlite" diff --git a/pyproject.toml b/pyproject.toml index 9558bf2..516ea05 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,6 +44,17 @@ testpaths = [ "xcov19/tests", ] asyncio_mode = "auto" +markers = [ + "slow: marks tests as slow (deselect with '-m \"not slow\"')", + "integration: marks tests as integration tests", + "unit: marks tests as unit tests", + # Add more markers as needed +] +# Add env vars when running pytest +env = [ + "APP_ENV=test", + "APP_DB_ENGINE_URL=sqlite+aiosqlite://" +] [tool.pyright] pythonVersion = "3.12" diff --git a/run.sh b/run.sh index c9ea51b..e7ca90c 100755 --- a/run.sh +++ b/run.sh @@ -1,3 +1,3 @@ #!/bin/bash -poetry run python3 -m xcov19.dev \ No newline at end of file +APP_ENV=dev APP_DB_ENGINE_URL="sqlite+aiosqlite:///xcov19.db" poetry run python3 -m xcov19.dev \ No newline at end of file diff --git a/xcov19/app/settings.py b/xcov19/app/settings.py index 1cd9520..2059753 100644 --- a/xcov19/app/settings.py +++ b/xcov19/app/settings.py @@ -7,8 +7,9 @@ https://docs.pydantic.dev/latest/usage/settings/ """ +from typing import Annotated from blacksheep import FromHeader -from pydantic import BaseModel +from pydantic import BaseModel, Field from pydantic_settings import BaseSettings, SettingsConfigDict @@ -26,6 +27,8 @@ class Site(BaseModel): class Settings(BaseSettings): + db_engine_url: Annotated[str, "database connection string"] = Field(default=...) + # to override info: # export app_info='{"title": "x", "version": "0.0.2"}' info: APIInfo = APIInfo() @@ -34,13 +37,14 @@ class Settings(BaseSettings): # export app_app='{"show_error_details": True}' app: App = App() - db_engine_url: str = "sqlite+aiosqlite:///" # "sqlite+aiosqlite:///xcov19.db" - model_config = SettingsConfigDict(env_prefix="APP_") def load_settings() -> Settings: - return Settings() + settings = Settings() + if not settings.db_engine_url: + raise ValueError("Missing environment variable: APP_DB_ENGINE_URL") + return settings class FromOriginMatchHeader(FromHeader[str]): diff --git a/xcov19/tests/conftest.py b/xcov19/tests/conftest.py index c7051d9..5026af0 100644 --- a/xcov19/tests/conftest.py +++ b/xcov19/tests/conftest.py @@ -1,8 +1,10 @@ from collections.abc import Callable from typing import List +from blacksheep.testing import TestClient import pytest +from blacksheep import Application from xcov19.dto import ( AnonymousId, GeoLocation, @@ -14,6 +16,10 @@ from xcov19.services.geolocation import LocationQueryServiceInterface from xcov19.utils.mixins import InterfaceProtocolCheckMixin +import random + +RANDOM_SEED = random.seed(1) + # Same as using @pytest.mark.anyio pytestmark = pytest.mark.anyio @@ -29,12 +35,12 @@ def anyio_backend(request): @pytest.fixture(scope="class") -def dummy_coordinates(): +def dummy_coordinates() -> GeoLocation: return GeoLocation(lat=0, lng=0) @pytest.fixture(scope="class") -def dummy_geolocation_query_json(dummy_coordinates): +def dummy_geolocation_query_json(dummy_coordinates) -> LocationQueryJSON: return LocationQueryJSON( location=dummy_coordinates, cust_id=AnonymousId(cust_id="test_cust_id"), @@ -43,8 +49,43 @@ def dummy_geolocation_query_json(dummy_coordinates): @pytest.fixture(scope="class") -def stub_location_srvc(): - return StubLocationQueryServiceImpl +def dummy_reverse_geo_lookup_svc() -> Callable[[LocationQueryJSON], dict]: + def callback(query: LocationQueryJSON) -> dict: + return {} + + return callback + + +@pytest.fixture(scope="class") +def dummy_patient_query_lookup_svc_none() -> ( + Callable[[Address, LocationQueryJSON], list] +): + def callback(address: Address, query: LocationQueryJSON) -> list: + return [] + + return callback + + +@pytest.fixture(scope="class") +def dummy_patient_query_lookup_svc() -> Callable[[Address, LocationQueryJSON], list]: + def callback(address: Address, query: LocationQueryJSON) -> list: + return [ + FacilitiesResult( + name="Test facility", + address=Address(), + geolocation=GeoLocation(lat=0.0, lng=0.0), + contact="+919999999999", + facility_type="nursing", + ownership="charity", + specialties=["surgery", "pediatrics"], + stars=4, + reviews=120, + rank=random.randint(1, 20), + estimated_time=20, + ) + ] + + return callback class StubLocationQueryServiceImpl( @@ -82,3 +123,17 @@ async def fetch_facilities( estimated_time=20, ) ] + + +@pytest.fixture(scope="class") +def stub_location_srvc() -> LocationQueryServiceInterface: + return StubLocationQueryServiceImpl + + +@pytest.fixture(scope="function", name="client") +async def test_client(): + # Create a test client + async def start_client(app: Application) -> TestClient: + return TestClient(app) + + return start_client diff --git a/xcov19/tests/start_server.py b/xcov19/tests/start_server.py new file mode 100644 index 0000000..f1a7634 --- /dev/null +++ b/xcov19/tests/start_server.py @@ -0,0 +1,13 @@ +from collections.abc import AsyncGenerator +from xcov19.app.main import app +from blacksheep import Application + + +async def start_server() -> AsyncGenerator[Application, None]: + """Start a test server for automated testing.""" + try: + await app.start() + yield app + finally: + if app.started: + await app.stop() diff --git a/xcov19/tests/test_geolocation_api.py b/xcov19/tests/test_geolocation_api.py new file mode 100644 index 0000000..083b13e --- /dev/null +++ b/xcov19/tests/test_geolocation_api.py @@ -0,0 +1,37 @@ +import json +import pytest +from xcov19.dto import LocationQueryJSON, GeoLocation, AnonymousId, QueryId +from blacksheep import Content, Response + + +@pytest.mark.integration +@pytest.mark.usefixtures("client") +class TestGeolocationAPI: + async def test_location_query_endpoint(self, client): + # Prepare the request payload + location_query = LocationQueryJSON( + location=GeoLocation(lat=0, lng=0), + cust_id=AnonymousId(cust_id="test_cust_id"), + query_id=QueryId(query_id="test_query_id"), + ) + + # Send a POST request to the /geo endpoint + query = location_query.model_dump(round_trip=True) + binary_data = json.dumps(query).encode("utf-8") + print("binary data", binary_data, type(binary_data)) + response: Response = await client.post( + "/geo", + content=Content(b"application/json", binary_data), + # Add the required header + headers={ + "X-Origin-Match-Header": "secret", + }, + ) + + # The current implementation returns ok(), which is null in JSON + # response_text = await response.text() + # assert response_text.lower() == "resource not found" + # Assert the response + assert response.content_type() == b"text/plain; charset=utf-8" + # assert response.content == b'' + assert response.status == 200 diff --git a/xcov19/tests/test_services.py b/xcov19/tests/test_services.py index 3dafb04..d23b2c4 100644 --- a/xcov19/tests/test_services.py +++ b/xcov19/tests/test_services.py @@ -1,6 +1,10 @@ +from collections.abc import Callable from typing import List import pytest import unittest + +from rodi import ContainerProtocol +from xcov19.tests.start_server import start_server from xcov19.domain.models.provider import ( Contact, FacilityEstablishment, @@ -20,35 +24,9 @@ import random -RANDOM_SEED = random.seed(1) - - -def dummy_reverse_geo_lookup_svc(query: LocationQueryJSON) -> dict: - return {} - - -def dummy_patient_query_lookup_svc_none( - address: Address, query: LocationQueryJSON -) -> list: - return [] - +from sqlalchemy.ext.asyncio import AsyncSession -def dummy_patient_query_lookup_svc(address: Address, query: LocationQueryJSON) -> list: - return [ - FacilitiesResult( - name="Test facility", - address=Address(), - geolocation=GeoLocation(lat=0.0, lng=0.0), - contact="+919999999999", - facility_type="nursing", - ownership="charity", - specialties=["surgery", "pediatrics"], - stars=4, - reviews=120, - rank=random.randint(1, 20), - estimated_time=20, - ) - ] +RANDOM_SEED = random.seed(1) class DummyProviderRepo(IProviderRepository[Provider], InterfaceProtocolCheckMixin): @@ -118,28 +96,37 @@ def stub_get_facilities_by_patient_query( return facilities_result -@pytest.mark.usefixtures("dummy_geolocation_query_json", "stub_location_srvc") +@pytest.mark.usefixtures( + "dummy_geolocation_query_json", + "dummy_reverse_geo_lookup_svc", + "dummy_patient_query_lookup_svc", + "stub_location_srvc", +) class GeoLocationServiceInterfaceTest(unittest.IsolatedAsyncioTestCase): @pytest.fixture(autouse=True) def autouse( self, dummy_geolocation_query_json: LocationQueryJSON, + dummy_reverse_geo_lookup_svc: Callable[[LocationQueryJSON], dict], + dummy_patient_query_lookup_svc: Callable[[Address, LocationQueryJSON], list], stub_location_srvc: LocationQueryServiceInterface, ): self.dummy_geolocation_query_json = dummy_geolocation_query_json + self.dummy_reverse_geo_lookup_svc = dummy_reverse_geo_lookup_svc + self.dummy_patient_query_lookup_svc = dummy_patient_query_lookup_svc self.stub_location_srvc = stub_location_srvc async def test_resolve_coordinates(self): result = await self.stub_location_srvc.resolve_coordinates( - dummy_reverse_geo_lookup_svc, self.dummy_geolocation_query_json + self.dummy_reverse_geo_lookup_svc, self.dummy_geolocation_query_json ) self.assertEqual(Address(), result) async def test_fetch_facilities(self): result = await self.stub_location_srvc.fetch_facilities( - dummy_reverse_geo_lookup_svc, + self.dummy_reverse_geo_lookup_svc, self.dummy_geolocation_query_json, - dummy_patient_query_lookup_svc, + self.dummy_patient_query_lookup_svc, ) self.assertIsInstance(result, list) assert result @@ -148,24 +135,40 @@ async def test_fetch_facilities(self): ) -@pytest.mark.usefixtures("dummy_geolocation_query_json") +@pytest.mark.usefixtures( + "dummy_geolocation_query_json", + "dummy_reverse_geo_lookup_svc", + "dummy_patient_query_lookup_svc", + "dummy_patient_query_lookup_svc_none", +) class GeoLocationServiceTest(unittest.IsolatedAsyncioTestCase): @pytest.fixture(autouse=True) - def autouse(self, dummy_geolocation_query_json: LocationQueryJSON): + def autouse( + self, + dummy_geolocation_query_json: LocationQueryJSON, + dummy_reverse_geo_lookup_svc: Callable[[LocationQueryJSON], dict], + dummy_patient_query_lookup_svc: Callable[[Address, LocationQueryJSON], list], + dummy_patient_query_lookup_svc_none: Callable[ + [Address, LocationQueryJSON], list + ], + ): self.dummy_geolocation_query_json = dummy_geolocation_query_json + self.dummy_reverse_geo_lookup_svc = dummy_reverse_geo_lookup_svc + self.dummy_patient_query_lookup_svc = dummy_patient_query_lookup_svc + self.dummy_patient_query_lookup_svc_none = dummy_patient_query_lookup_svc_none async def test_resolve_coordinates(self): result = await GeolocationQueryService.resolve_coordinates( - dummy_reverse_geo_lookup_svc, self.dummy_geolocation_query_json + self.dummy_reverse_geo_lookup_svc, self.dummy_geolocation_query_json ) expected = Address() self.assertEqual(expected, result, f"Got {result}, expected {expected}") async def test_fetch_facilities(self): result = await GeolocationQueryService.fetch_facilities( - dummy_reverse_geo_lookup_svc, + self.dummy_reverse_geo_lookup_svc, self.dummy_geolocation_query_json, - dummy_patient_query_lookup_svc, + self.dummy_patient_query_lookup_svc, ) self.assertIsNotNone(result) record = None @@ -175,23 +178,66 @@ async def test_fetch_facilities(self): async def test_fetch_facilities_no_results(self): result = await GeolocationQueryService.fetch_facilities( - dummy_reverse_geo_lookup_svc, + self.dummy_reverse_geo_lookup_svc, self.dummy_geolocation_query_json, - dummy_patient_query_lookup_svc_none, + self.dummy_patient_query_lookup_svc_none, ) self.assertIsNone(result) -# @pytest.mark.skip("WIP") -@pytest.mark.usefixtures("dummy_geolocation_query_json") +@pytest.mark.skip(reason="WIP") +@pytest.mark.integration +@pytest.mark.usefixtures("dummy_reverse_geo_lookup_svc", "dummy_geolocation_query_json") +class GeoLocationServiceSqlRepoDBTest(unittest.IsolatedAsyncioTestCase): + """Test case for Sqlite Repository to test Geolocation Service. + + Before testing, ensure to: + 1. Setup Database + 2. For fetch_facilities, relevant services are configured. + 3. patient_query_lookup_svc is configured to call sqlite repository. + """ + + async def asyncSetUp(self) -> None: + app = await anext(start_server()) + self._container: ContainerProtocol = app.services + self._seed_db(self._container.resolve(AsyncSession)) + await super().asyncSetUp() + + def _seed_db(self, session: AsyncSession) -> None: + # TODO: add data to sqlite tables based on dummy_geolocation_query_json + # and add providers data. + ... + + def _patient_query_lookup_svc_using_repo( + self, address: Address, query: LocationQueryJSON + ) -> Callable[[Address, LocationQueryJSON], List[FacilitiesResult]]: ... + + async def test_fetch_facilities( + self, dummy_reverse_geo_lookup_svc, dummy_geolocation_query_json + ): + # TODO Implement test_fetch_facilities like this: + # providers = await GeolocationQueryService.fetch_facilities( + # dummy_reverse_geo_lookup_svc, + # dummy_geolocation_query_json, + # self._patient_query_lookup_svc_using_repo + # ) + ... + + +@pytest.mark.usefixtures("dummy_geolocation_query_json", "dummy_reverse_geo_lookup_svc") class PatientQueryLookupSvcTest(unittest.IsolatedAsyncioTestCase): @pytest.fixture(autouse=True) - def autouse(self, dummy_geolocation_query_json: LocationQueryJSON): + def autouse( + self, + dummy_geolocation_query_json: LocationQueryJSON, + dummy_reverse_geo_lookup_svc: Callable[[LocationQueryJSON], dict], + ): self.dummy_geolocation_query_json = dummy_geolocation_query_json + self.dummy_reverse_geo_lookup_svc = dummy_reverse_geo_lookup_svc async def test_patient_query_lookup_svc(self): providers = await GeolocationQueryService.fetch_facilities( - dummy_reverse_geo_lookup_svc, + self.dummy_reverse_geo_lookup_svc, self.dummy_geolocation_query_json, stub_get_facilities_by_patient_query, )