refactor (tests): move shared test code to ... shared.py

Author: pjhartzell

tests/application.py

@@ -1,88 +1,21 @@
-from collections import defaultdict
+import os
+import sys
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
-from typing import Any, Literal, Self
+from typing import Any
 
 from fastapi import FastAPI
-from pydantic import BaseModel, Field, model_validator
+
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
 
 from stapi_fastapi.models.conformance import CORE
-from stapi_fastapi.models.opportunity import (
-    OpportunityProperties,
-)
-from stapi_fastapi.models.order import (
-    Order,
-    OrderParameters,
-    OrderStatus,
-)
-from stapi_fastapi.models.product import (
-    Product,
-    Provider,
-    ProviderRole,
-)
 from stapi_fastapi.routers.root_router import RootRouter
-
-from .backends import (
-    mock_create_order,
+from tests.backends import (
     mock_get_order,
     mock_get_order_statuses,
     mock_get_orders,
-    mock_search_opportunities,
-)
-
-
-class InMemoryOrderDB:
-    def __init__(self) -> None:
-        self._orders: dict[str, Order] = {}
-        self._statuses: dict[str, list[OrderStatus]] = defaultdict(list)
-
-
-class MyProductConstraints(BaseModel):
-    off_nadir: int
-
-
-class OffNadirRange(BaseModel):
-    minimum: int = Field(ge=0, le=45)
-    maximum: int = Field(ge=0, le=45)
-
-    @model_validator(mode="after")
-    def validate_range(self) -> Self:
-        if self.minimum > self.maximum:
-            raise ValueError("range minimum cannot be greater than maximum")
-        return self
-
-
-class MyOpportunityProperties(OpportunityProperties):
-    off_nadir: OffNadirRange
-    vehicle_id: list[Literal[1, 2, 5, 7, 8]]
-    platform: Literal["platform_id"]
-
-
-class MyOrderParameters(OrderParameters):
-    s3_path: str | None = None
-
-
-provider = Provider(
-    name="Test Provider",
-    description="A provider for Test data",
-    roles=[ProviderRole.producer],  # Example role
-    url="https://test-provider.example.com",  # Must be a valid URL
-)
-
-mock_product_test_spotlight = Product(
-    id="test-spotlight",
-    title="Test Spotlight Product",
-    description="Test product for test spotlight",
-    license="CC-BY-4.0",
-    keywords=["test", "satellite"],
-    providers=[provider],
-    links=[],
-    create_order=mock_create_order,
-    search_opportunities=mock_search_opportunities,
-    constraints=MyProductConstraints,
-    opportunity_properties=MyOpportunityProperties,
-    order_parameters=MyOrderParameters,
 )
+from tests.shared import InMemoryOrderDB, mock_product_test_spotlight
 
 
 @asynccontextmanager

tests/conftest.py

@@ -1,17 +1,11 @@
 from collections.abc import AsyncIterator, Iterator
 from contextlib import asynccontextmanager
-from datetime import datetime, timedelta, timezone
 from typing import Any, Callable
 from urllib.parse import urljoin
-from uuid import uuid4
 
 import pytest
-from fastapi import FastAPI, status
+from fastapi import FastAPI
 from fastapi.testclient import TestClient
-from geojson_pydantic import Point
-from geojson_pydantic.types import Position2D
-from httpx import Response
-from pytest import fail
 
 from stapi_fastapi.models.opportunity import (
     Opportunity,
@@ -21,49 +15,35 @@
 )
 from stapi_fastapi.routers.root_router import RootRouter
 
-from .application import (
-    InMemoryOrderDB,
-    MyOpportunityProperties,
-    MyOrderParameters,
-    MyProductConstraints,
-    OffNadirRange,
-    mock_create_order,
+from .backends import (
     mock_get_order,
     mock_get_order_statuses,
     mock_get_orders,
+)
+from .shared import (
+    InMemoryOrderDB,
+    create_mock_opportunity,
+    find_link,
+    mock_product_test_satellite_provider,
     mock_product_test_spotlight,
-    mock_search_opportunities,
-    provider,
 )
-from .shared import find_link
 
 
 @pytest.fixture(scope="session")
 def base_url() -> Iterator[str]:
     yield "http://stapiserver"
 
 
-mock_product_test_satellite_provider = Product(
-    id="test-satellite-provider",
-    title="Satellite Product",
-    description="A product by a satellite provider",
-    license="CC-BY-4.0",
-    keywords=["test", "satellite", "provider"],
-    providers=[provider],
-    links=[],
-    create_order=mock_create_order,
-    search_opportunities=mock_search_opportunities,
-    constraints=MyProductConstraints,
-    opportunity_properties=MyOpportunityProperties,
-    order_parameters=MyOrderParameters,
-)
-
-
 @pytest.fixture
 def mock_products() -> list[Product]:
     return [mock_product_test_spotlight, mock_product_test_satellite_provider]
 
 
+@pytest.fixture
+def mock_opportunities() -> list[Opportunity]:
+    return [create_mock_opportunity()]
+
+
 @pytest.fixture
 def stapi_client(
     mock_products: list[Product],
@@ -134,99 +114,3 @@ def _assert_link(
         assert link["href"] == url_for(path)
 
     return _assert_link
-
-
-@pytest.fixture
-def mock_opportunities() -> list[Opportunity]:
-    """Fixture to create mock data for Opportunities for `test-spotlight-1`."""
-    now = datetime.now(timezone.utc)  # Use timezone-aware datetime
-    start = now
-    end = start + timedelta(days=5)
-
-    # Create a list of mock opportunities for the given product
-    return [
-        Opportunity(
-            id=str(uuid4()),
-            type="Feature",
-            geometry=Point(
-                type="Point",
-                coordinates=Position2D(longitude=0.0, latitude=0.0),
-            ),
-            properties=MyOpportunityProperties(
-                product_id="xyz123",
-                datetime=(start, end),
-                off_nadir=OffNadirRange(minimum=20, maximum=22),
-                vehicle_id=[1],
-                platform="platform_id",
-                other_thing="abcd1234",
-            ),
-        ),
-    ]
-
-
-def pagination_tester(
-    stapi_client: TestClient,
-    endpoint: str,
-    method: str,
-    limit: int,
-    target: str,
-    expected_returns: list,
-    body: dict | None = None,
-) -> None:
-    retrieved = []
-
-    res = make_request(stapi_client, endpoint, method, body, None, limit)
-    assert res.status_code == status.HTTP_200_OK
-    resp_body = res.json()
-
-    assert len(resp_body[target]) <= limit
-    retrieved.extend(resp_body[target])
-    next_url = next((d["href"] for d in resp_body["links"] if d["rel"] == "next"), None)
-
-    while next_url:
-        url = next_url
-        if method == "POST":
-            body = next(
-                (d["body"] for d in resp_body["links"] if d["rel"] == "next"), None
-            )
-
-        res = make_request(stapi_client, url, method, body, next_url, limit)
-        assert res.status_code == status.HTTP_200_OK
-        assert len(resp_body[target]) <= limit
-        resp_body = res.json()
-        retrieved.extend(resp_body[target])
-
-        # get url w/ query params for next call if exists, and POST body if necessary
-        if resp_body["links"]:
-            next_url = next(
-                (d["href"] for d in resp_body["links"] if d["rel"] == "next"), None
-            )
-        else:
-            next_url = None
-
-    assert len(retrieved) == len(expected_returns)
-    assert retrieved == expected_returns
-
-
-def make_request(
-    stapi_client: TestClient,
-    endpoint: str,
-    method: str,
-    body: dict | None,
-    next_token: str | None,
-    limit: int,
-) -> Response:
-    """request wrapper for pagination tests"""
-
-    match method:
-        case "GET":
-            if next_token:  # extract pagination token
-                next_token = next_token.split("next=")[1]
-            params = {"next": next_token, "limit": limit}
-            res = stapi_client.get(endpoint, params=params)
-        case "POST":
-            res = stapi_client.post(endpoint, json=body)
-        case _:
-            fail(f"method {method} not supported in make request")
-
-    return res