fix mypy (7d9ce6f9) · Commits · OOP / code / open-exposure-gateway

.pre-commit-config.yaml

+1 −1

Original line number	Diff line number	Diff line
		@@ -23,4 +23,4 @@ repos:
		- id: mypy
		files: ^src/open_exposure_gateway/\|^tests/
		args: [--strict, --ignore-missing-imports, --cache-dir, .cache/mypy]
		additional_dependencies: ["fastapi[standard]>=0.135.1", "pydantic>=2.0", "pydantic-settings>=2.0", "httpx>=0.27", "pytest>=9.0.2", "sqlalchemy>=2.0.48", "pytest-asyncio>=0.24", "testcontainers>=4.0.0", "types-PyYAML>=6.0.1"]
		additional_dependencies: ["fastapi[standard]>=0.135.1", "pydantic>=2.0", "pydantic-settings>=2.0", "httpx>=0.27", "pytest>=9.0.2", "sqlalchemy>=2.0.48", "pytest-asyncio>=0.24", "testcontainers==4.14.2", "types-PyYAML>=6.0.1"]

src/open_exposure_gateway/application/services/traffic_influence_service.py

+27 −3

Original line number	Diff line number	Diff line
		@@ -21,6 +21,7 @@ from open_exposure_gateway.application.mappers.traffic_influence_mapper import (
		build_traffic_influence_response,
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		NotFoundException,
		@@ -65,6 +66,17 @@ _TERMINAL_OPERATION_STATUSES = frozenset(
		# CallbackRegistration.api_family discriminator for Traffic Influence's subscriptionRequest.
		_API_FAMILY = "traffic_influence"

		# Blocks re-deleting an already-terminal resource (same purpose as QoD's _TERMINAL_STATES
		# guard on delete_session). GET intentionally does NOT use this -- TI's own CAMARA spec
		# documents 'deletion in progress'/'deleted' as visible GET states, unlike QoD.
		_TERMINAL_STATES = frozenset(
		{
		RecordTrafficInfluenceState.DELETION_IN_PROGRESS,
		RecordTrafficInfluenceState.DELETED,
		RecordTrafficInfluenceState.ERROR,
		}
		)


		class TrafficInfluenceService:
		def __init__(
		@@ -81,6 +93,14 @@ class TrafficInfluenceService:
		self._traffic_influence_repo = traffic_influence_repo
		self._callback_registration_repo = callback_registration_repo

		def _parse_traffic_influence_id(self, traffic_influence_id: str) -> UUID:
		try:
		return UUID(traffic_influence_id)
		except ValueError as exc:
		raise BadRequestException(
		message=f"Malformed traffic influence ID: {traffic_influence_id}"
		) from exc

		def _new_operation_metadata(self, x_correlator: Optional[str]) -> tuple[UUID, str, str]:
		operation_id = uuid4()
		correlation_id = x_correlator or str(uuid4())
		@@ -394,7 +414,9 @@ class TrafficInfluenceService:
		if self._traffic_influence_repo is None:
		raise RuntimeError("TrafficInfluence repository is not available")

		traffic_influence = await self._traffic_influence_repo.get_by_id(UUID(traffic_influence_id))
		traffic_influence = await self._traffic_influence_repo.get_by_id(
		self._parse_traffic_influence_id(traffic_influence_id)
		)
		if traffic_influence is None:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")

		@@ -451,8 +473,10 @@ class TrafficInfluenceService:
		if self._operation_repo is None or self._traffic_influence_repo is None:
		raise RuntimeError("Operation/TrafficInfluence repositories are not available")

		traffic_influence = await self._traffic_influence_repo.get_by_id(UUID(traffic_influence_id))
		if traffic_influence is None:
		traffic_influence = await self._traffic_influence_repo.get_by_id(
		self._parse_traffic_influence_id(traffic_influence_id)
		)
		if traffic_influence is None or traffic_influence.state in _TERMINAL_STATES:
		raise NotFoundException(message=f"Traffic influence {traffic_influence_id} not found")

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)

tests/integration/conftest.py

+29 −2

Original line number	Diff line number	Diff line
		import re
		from collections.abc import AsyncGenerator, AsyncIterator, Generator
		from contextlib import asynccontextmanager

		@@ -16,7 +17,8 @@ from sqlalchemy.ext.asyncio import (
		create_async_engine,
		)
		from testcontainers.core.container import DockerContainer
		from testcontainers.core.wait_strategies import ExecWaitStrategy, LogMessageWaitStrategy
		from testcontainers.core.wait_strategies import LogMessageWaitStrategy
		from testcontainers.core.waiting_utils import WaitStrategy, WaitStrategyTarget

		from open_exposure_gateway.adapters.database.core import (
		build_engine_and_session_maker,
		@@ -65,6 +67,31 @@ async def publisher(nats_url: str) -> AsyncGenerator[NatsMessagePublisher, None]
		await p.close()


		class _PostgresReadyTwiceWaitStrategy(WaitStrategy): # type: ignore[misc]
		"""Waits for postgres's "ready to accept connections" log line twice.

		The postgres image starts a temporary Unix-socket-only server to run
		initdb/init scripts, then restarts as the real TCP-reachable server; both
		log the same "ready to accept connections" line. `LogMessageWaitStrategy`'s
		`times` argument only checks presence, not count (testcontainers 4.14.2),
		so it returns as soon as the temporary server's line appears -- racing
		with the real server's restart and surfacing as
		asyncpg.exceptions.CannotConnectNowError: "the database system is
		starting up". Counting occurrences directly closes that window.
		"""

		_PATTERN = re.compile(r"database system is ready to accept connections")

		def wait_until_ready(self, container: WaitStrategyTarget) -> None:
		def _seen_twice() -> bool:
		stdout, stderr = container.get_logs()
		text = stdout.decode() + stderr.decode()
		return len(self._PATTERN.findall(text)) >= 2

		if not self._poll(_seen_twice):
		raise TimeoutError("Postgres did not report ready twice within the startup timeout")


		@pytest.fixture(scope="session")
		def postgres_url() -> Generator[str, None, None]:
		container = (
		@@ -73,7 +100,7 @@ def postgres_url() -> Generator[str, None, None]:
		.with_env("POSTGRES_PASSWORD", "postgres")
		.with_env("POSTGRES_DB", "oeg_test")
		.with_exposed_ports(5432)
		.waiting_for(ExecWaitStrategy(["pg_isready", "-U", "postgres"]))
		.waiting_for(_PostgresReadyTwiceWaitStrategy())
		)
		with container:
		host = container.get_container_host_ip()

tests/unit/test_traffic_influence_flows.py

+27 −0

Original line number	Diff line number	Diff line
		@@ -252,6 +252,12 @@ class TestTrafficInfluenceGetFlow:
		response = api_client.get(f"{TI_BASE}/traffic-influences/{uuid4()}")
		assert response.status_code == 404

		def test_get_traffic_influence_returns_400_for_malformed_id(
		self, api_client: TestClient
		) -> None:
		response = api_client.get(f"{TI_BASE}/traffic-influences/not-a-uuid")
		assert response.status_code == 400

		def test_list_traffic_influences_returns_empty_by_default(self, api_client: TestClient) -> None:
		response = api_client.get(f"{TI_BASE}/traffic-influences")
		assert response.status_code == 200
		@@ -281,6 +287,27 @@ class TestTrafficInfluenceDeleteFlow:
		response = api_client.delete(f"{TI_BASE}/traffic-influences/{uuid4()}")
		assert response.status_code == 404

		def test_delete_returns_400_for_malformed_id(self, api_client: TestClient) -> None:
		response = api_client.delete(f"{TI_BASE}/traffic-influences/not-a-uuid")
		assert response.status_code == 400

		def test_delete_returns_404_when_already_in_progress(
		self, api_client: TestClient, fake_bus: FakeDataBus
		) -> None:
		"""A repeat DELETE on a resource already being torn down must not re-publish
		another deactivate command (same guard as QoD's delete_session)."""
		traffic_influence_id = api_client.post(
		f"{TI_BASE}/traffic-influences", json=TI_BODY
		).json()["trafficInfluenceID"]
		first = api_client.delete(f"{TI_BASE}/traffic-influences/{traffic_influence_id}")
		assert first.status_code == 202

		second = api_client.delete(f"{TI_BASE}/traffic-influences/{traffic_influence_id}")

		assert second.status_code == 404
		deactivates = [p for s, p in fake_bus.published if s == Subject.TASK_DEACTIVATE]
		assert len(deactivates) == 1

		async def test_delete_publishes_deactivate_command_once_available(
		self,
		api_client: TestClient,