feat: return 409 conflict when the device already has an active session (08ad57cb) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/database/repos/qod_sessions.py

+31 −3

Changes for src/open_exposure_gateway/adapters/database/repos/qod_sessions.py: 31 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		from typing import Any
		from uuid import UUID

		from sqlalchemy import select
		from sqlalchemy import or_, select, type_coerce
		from sqlalchemy.dialects.postgresql import JSONB
		from sqlalchemy.ext.asyncio import AsyncSession

		from open_exposure_gateway.adapters.database.mappers import QodSessionMapper
		from open_exposure_gateway.adapters.database.sql import QodSessionRow
		from open_exposure_gateway.domain.models import QodSession
		from open_exposure_gateway.adapters.database.sql import OperationRow, QodSessionRow
		from open_exposure_gateway.domain.models import QodSession, QodSessionState
		from open_exposure_gateway.ports.database.qod_sessions import QodSessionRepository

		_ACTIVE_STATES = (QodSessionState.REQUESTED, QodSessionState.AVAILABLE)
		_DEVICE_IDENTIFIERS = ("phoneNumber", "ipv4Address", "ipv6Address")


		class SqlQodSessionRepository(QodSessionRepository):
		def __init__(self, session: AsyncSession) -> None:
		@@ -23,6 +28,29 @@ class SqlQodSessionRepository(QodSessionRepository):
		row = await self._session.scalar(stmt)
		return QodSessionMapper.to_domain(row) if row is not None else None

		async def find_active_by_device(
		self, device: dict[str, Any], app_provider_id: str
		) -> QodSession \| None:
		identifier_matches = [
		QodSessionRow.device[key] == type_coerce(device[key], JSONB)
		for key in _DEVICE_IDENTIFIERS
		if device.get(key) is not None
		]
		if not identifier_matches:
		return None
		stmt = (
		select(QodSessionRow)
		.join(OperationRow, OperationRow.operation_id == QodSessionRow.operation_id)
		.where(
		OperationRow.app_provider_id == app_provider_id,
		QodSessionRow.state.in_(_ACTIVE_STATES),
		or_(*identifier_matches),
		)
		.limit(1)
		)
		row = await self._session.scalar(stmt)
		return QodSessionMapper.to_domain(row) if row is not None else None

		async def save(self, qod_session: QodSession) -> QodSession:
		merged = await self._session.merge(QodSessionMapper.to_row(qod_session))
		await self._session.flush()

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+9 −1

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 9 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -28,6 +28,7 @@ from open_exposure_gateway.core.config import (
		)
		from open_exposure_gateway.core.exceptions import (
		BadRequestException,
		ConflictException,
		DownstreamServiceException,
		ErrorCode,
		NotFoundException,
		@@ -184,6 +185,13 @@ class QualityOnDemandService:
		"receive no notifications"
		)

		device_json = device.model_dump(mode="json", exclude_none=True)
		if await self._qod_session_repo.find_active_by_device(device_json, app_provider_id):
		raise ConflictException(
		error_code=ErrorCode.CONFLICT,
		message="Conflict with an existing session for the same device.",
		)

		operation_id, correlation_id, requested_at = self._new_operation_metadata(x_correlator)
		session_id = uuid4()
		# The platform-seeded QoD specification, identical on every session (ADR-0035).
		@@ -235,7 +243,7 @@ class QualityOnDemandService:
		if request.applicationServerPorts is not None
		else None
		),
		device=device.model_dump(mode="json", exclude_none=True),
		device=device_json,
		application_server=request.applicationServer.model_dump(
		mode="json", exclude_none=True
		),

src/open_exposure_gateway/ports/database/qod_sessions.py

+7 −0

Changes for src/open_exposure_gateway/ports/database/qod_sessions.py: 7 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""QoD session repository ports."""

		from abc import ABC, abstractmethod
		from typing import Any
		from uuid import UUID

		from open_exposure_gateway.domain.models import QodSession
		@@ -15,6 +16,12 @@ class QodSessionRepository(ABC):
		async def get_by_operation_id(self, operation_id: UUID) -> QodSession \| None:
		pass

		@abstractmethod
		async def find_active_by_device(
		self, device: dict[str, Any], app_provider_id: str
		) -> QodSession \| None:
		pass

		@abstractmethod
		async def save(self, qod_session: QodSession) -> QodSession:
		pass

tests/conformance/conftest.py

+1 −1

Changes for tests/conformance/conftest.py: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -83,7 +83,7 @@ def service_overrides() -> Generator[None, None, None]:
		)

		qod_operation_repo = FakeOperationRepository()
		qod_session_repo = FakeQodSessionRepository()
		qod_session_repo = FakeQodSessionRepository(qod_operation_repo)
		qod_callback_registration_repo = FakeCallbackRegistrationRepository()
		qod_callback_delivery_repo = FakeCallbackDeliveryRepository()
		qod_callback_delivery_port = FakeQodCallbackDeliveryPort()

tests/integration/test_postgres.py

+28 −0

Changes for tests/integration/test_postgres.py: 28 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -573,6 +573,34 @@ async def test_qod_session_repo_get_by_operation_id_returns_none_when_missing(
		assert await repo.get_by_operation_id(uuid4()) is None


		async def test_qod_session_repo_find_active_by_device_scopes_by_identifier_state_and_owner(
		db_session: AsyncSession,
		) -> None:
		operation = await SqlOperationRepository(db_session).save(_operation())
		repo = SqlQodSessionRepository(db_session)
		ipv4 = {"publicAddress": "203.0.113.7", "publicPort": 59765}
		qod_session = _qod_session(operation_id=operation.operation_id)
		qod_session.device = {"phoneNumber": "+123456789", "ipv4Address": ipv4}
		saved = await repo.save(qod_session)

		by_phone = await repo.find_active_by_device({"phoneNumber": "+123456789"}, "tenant-a")
		by_ipv4 = await repo.find_active_by_device({"ipv4Address": ipv4}, "tenant-a")
		assert by_phone is not None and by_phone.session_id == saved.session_id
		assert by_ipv4 is not None and by_ipv4.session_id == saved.session_id
		assert await repo.find_active_by_device({"phoneNumber": "+123456789"}, "tenant-b") is None
		assert await repo.find_active_by_device({"phoneNumber": "+987654321"}, "tenant-a") is None
		assert (
		await repo.find_active_by_device(
		{"ipv4Address": {"publicAddress": "203.0.113.7", "publicPort": 1}}, "tenant-a"
		)
		is None
		)

		saved.state = QodSessionState.DELETED
		await repo.save(saved)
		assert await repo.find_active_by_device({"phoneNumber": "+123456789"}, "tenant-a") is None


		async def test_qod_session_repo_updates_updated_at_on_second_save(db_engine: AsyncEngine) -> None:
		session_factory = async_sessionmaker(bind=db_engine, expire_on_commit=False)