fix: fixed type errors and added some tests (7d1f5b12) · Commits · OOP / code / service-resource-manager

src/srm/api/databus/dependencies.py

+2 −1

Changes for src/srm/api/databus/dependencies.py: 2 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -15,6 +15,7 @@ from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.adapters.databus.nats_publisher import NatsPublisher
		from srm.application.command_handlers.deploy_service import DeployServiceCommandCoordinator
		from srm.application.use_cases.deploy_service import DeployServiceUseCase
		from srm.domain.ports.databus.publisher import DataBusPublisher


		def get_deploy_service_use_case(
		@@ -26,7 +27,7 @@ def get_deploy_service_use_case(

		def build_deploy_service_use_case(
		session: AsyncSession,
		publisher: NatsPublisher,
		publisher: DataBusPublisher,
		) -> DeployServiceUseCase:
		return DeployServiceUseCase(
		service_specifications=SqlServiceSpecificationRepository(session),

tests/api/databus/test_nats_subscriber.py

+173 −31

Changes for tests/api/databus/test_nats_subscriber.py: 173 added lines, 31 removed lines.
Original line number	Diff line number	Diff line
		from __future__ import annotations

		from dataclasses import dataclass, field
		from unittest.mock import AsyncMock, MagicMock
		from unittest.mock import AsyncMock, MagicMock, patch

		import pytest
		import structlog.testing
		@@ -12,7 +12,7 @@ from srm.api.databus.nats_subscriber import (
		NatsSubscriber,
		subscribe_to_subjects,
		)
		from srm.api.databus.schemas import InboundMessage
		from srm.api.databus.schemas import ValidatedEnvelopeV1


		@dataclass
		@@ -22,6 +22,27 @@ class FakeMsg:
		headers: dict[str, str] \| None = field(default=None)


		class FakeSessionContext:
		def __init__(self) -> None:
		self.session = AsyncMock()

		async def __aenter__(self) -> AsyncMock:
		return self.session

		async def __aexit__(self, *_args: object) -> None:
		return None


		class FakeSessionMaker:
		def __init__(self) -> None:
		self.contexts: list[FakeSessionContext] = []

		def __call__(self) -> FakeSessionContext:
		context = FakeSessionContext()
		self.contexts.append(context)
		return context


		@pytest.fixture
		def connection_manager() -> MagicMock:
		manager = MagicMock(spec=NatsConnectionManager)
		@@ -32,13 +53,13 @@ def connection_manager() -> MagicMock:
		def make_subscriber(
		connection_manager: MagicMock, subject: str = "command.srm.service.deploy"
		) -> tuple[NatsSubscriber, AsyncMock]:
		router = AsyncMock()
		callback = AsyncMock()
		subscriber = NatsSubscriber(
		connection_manager=connection_manager,
		subject=subject,
		router=router,
		callback=callback,
		)
		return subscriber, router
		return subscriber, callback


		async def test_start_subscribes_via_connection_manager_client(
		@@ -55,65 +76,144 @@ async def test_start_subscribes_via_connection_manager_client(
		assert subscriber._subscription is not None


		async def test_handle_message_builds_inbound_message_and_invokes_router(
		async def test_handle_message_builds_inbound_message_and_invokes_callback(
		connection_manager: MagicMock,
		) -> None:
		subscriber, router = make_subscriber(connection_manager)
		subscriber, callback = make_subscriber(connection_manager)
		msg = FakeMsg(
		subject="command.srm.service.deploy",
		data=b'{"operation_id": "abc-123"}',
		data=(
		b'{"schema_version": "1.0",'
		b' "operation_id": "11111111-2222-3333-4444-555555555555",'
		b' "correlation_id": "corr-1",'
		b' "requested_at": "2026-06-19T12:00:00Z",'
		b' "app_provider_id": "VideoAppsCo",'
		b' "source": "nbi_camara"}'
		),
		headers={"x-correlation-id": "corr-1"},
		)

		await subscriber._handle_message(msg) # type: ignore[arg-type]

		router.assert_awaited_once()
		assert router.await_args is not None
		(inbound,) = router.await_args.args
		assert isinstance(inbound, InboundMessage)
		assert inbound.subject == "command.srm.service.deploy"
		assert inbound.payload == b'{"operation_id": "abc-123"}'
		assert inbound.headers == {"x-correlation-id": "corr-1"}
		callback.assert_awaited_once()
		assert callback.await_args is not None
		(envelope,) = callback.await_args.args
		assert isinstance(envelope, ValidatedEnvelopeV1)
		assert envelope.correlation_id == "corr-1"
		assert envelope.model_extra == {}


		async def test_handle_message_preserves_command_payload_fields_on_validated_envelope(
		connection_manager: MagicMock,
		) -> None:
		subscriber, callback = make_subscriber(connection_manager)
		msg = FakeMsg(
		subject="command.srm.service.deploy",
		data=b"""
		{
		"schema_version": "1.0",
		"operation_id": "11111111-2222-3333-4444-555555555555",
		"correlation_id": "corr-1",
		"requested_at": "2026-06-19T12:00:00Z",
		"app_provider_id": "VideoAppsCo",
		"source": "nbi_camara",
		"service_specification_id": "d4e5f6a7-0000-4000-8000-000000000001",
		"targets": [
		{
		"app_instance_id": "9a3f1c22-0000-4000-8000-000000000002"
		}
		],
		"deploy": {}
		}
		""",
		)

		await subscriber._handle_message(msg) # type: ignore[arg-type]

		callback.assert_awaited_once()
		assert callback.await_args is not None
		(envelope,) = callback.await_args.args
		assert envelope.model_extra is not None
		assert envelope.model_extra["service_specification_id"] == (
		"d4e5f6a7-0000-4000-8000-000000000001"
		)
		assert envelope.model_extra["targets"] == [
		{"app_instance_id": "9a3f1c22-0000-4000-8000-000000000002"}
		]
		assert envelope.model_extra["deploy"] == {}


		async def test_handle_message_defaults_headers_to_empty_dict_when_none(
		connection_manager: MagicMock,
		) -> None:
		subscriber, router = make_subscriber(connection_manager)
		msg = FakeMsg(subject="command.srm.service.deploy", data=b"{}", headers=None)
		subscriber, callback = make_subscriber(connection_manager)
		msg = FakeMsg(
		subject="command.srm.service.deploy",
		data=(
		b'{"schema_version": "1.0",'
		b' "operation_id": "11111111-2222-3333-4444-555555555555",'
		b' "correlation_id": "corr-1",'
		b' "requested_at": "2026-06-19T12:00:00Z",'
		b' "app_provider_id": "VideoAppsCo",'
		b' "source": "nbi_camara"}'
		),
		headers=None,
		)

		await subscriber._handle_message(msg) # type: ignore[arg-type]

		assert router.await_args is not None
		(inbound,) = router.await_args.args
		assert inbound.headers == {}
		assert callback.await_args is not None
		(envelope,) = callback.await_args.args
		assert envelope.correlation_id == "corr-1"


		async def test_handle_message_passes_malformed_payload_through_unparsed(
		connection_manager: MagicMock,
		) -> None:
		subscriber, router = make_subscriber(connection_manager)
		subscriber, callback = make_subscriber(connection_manager)
		msg = FakeMsg(subject="command.srm.service.deploy", data=b'{"operation_id": ')

		with structlog.testing.capture_logs() as logs:
		await subscriber._handle_message(msg) # type: ignore[arg-type]

		router.assert_awaited_once()
		assert router.await_args is not None
		(inbound,) = router.await_args.args
		assert inbound.payload == b'{"operation_id": '
		callback.assert_not_awaited()
		assert [(entry["event"], entry["subject"], entry["log_level"]) for entry in logs] == [
		("databus_stage1_unanswerable", "command.srm.service.deploy", "error")
		]


		async def test_handle_message_logs_dropped_command_when_router_fails(
		async def test_handle_message_logs_stage1_unanswerable_when_envelope_is_invalid(
		connection_manager: MagicMock,
		) -> None:
		subscriber, router = make_subscriber(connection_manager)
		router.side_effect = RuntimeError("handler blew up")
		msg = FakeMsg(subject="command.srm.service.deploy", data=b"{}")
		subscriber, callback = make_subscriber(connection_manager)
		msg = FakeMsg(subject="command.srm.service.deploy", data=b'{"operation_id": "abc-123"}')

		await subscriber._handle_message(msg) # type: ignore[arg-type]

		callback.assert_not_awaited()


		async def test_handle_message_logs_dropped_command_when_callback_fails(
		connection_manager: MagicMock,
		) -> None:
		subscriber, callback = make_subscriber(connection_manager)
		callback.side_effect = RuntimeError("handler blew up")
		msg = FakeMsg(
		subject="command.srm.service.deploy",
		data=(
		b'{"schema_version": "1.0",'
		b' "operation_id": "11111111-2222-3333-4444-555555555555",'
		b' "correlation_id": "corr-1",'
		b' "requested_at": "2026-06-19T12:00:00Z",'
		b' "app_provider_id": "VideoAppsCo",'
		b' "source": "nbi_camara"}'
		),
		)

		with structlog.testing.capture_logs() as logs:
		await subscriber._handle_message(msg) # type: ignore[arg-type]

		router.assert_awaited_once()
		callback.assert_awaited_once()
		assert [(entry["event"], entry["subject"], entry["log_level"]) for entry in logs] == [
		("databus_handler_failed", "command.srm.service.deploy", "error")
		]
		@@ -133,10 +233,52 @@ def test_command_subjects_matches_the_contract() -> None:
		assert sorted(COMMAND_SUBJECTS) == sorted(EXPECTED_COMMAND_SUBJECTS)


		async def test_deploy_subscription_callback_delegates_to_command_coordinator(
		connection_manager: MagicMock,
		) -> None:
		envelope = ValidatedEnvelopeV1.model_validate(
		{
		"schema_version": "1.0",
		"operation_id": "11111111-2222-3333-4444-555555555555",
		"correlation_id": "corr-appinst-001",
		"requested_at": "2026-06-19T12:00:00Z",
		"app_provider_id": "VideoAppsCo",
		"source": "nbi_camara",
		"service_specification_id": "d4e5f6a7-0000-4000-8000-000000000001",
		"targets": [
		{
		"app_instance_id": "9a3f1c22-0000-4000-8000-000000000002",
		"zone_id": "642f6105-7015-4af1-a4d1-e1ecb8437abc",
		}
		],
		"deploy": {},
		}
		)
		session_maker = FakeSessionMaker()
		with patch(
		"srm.api.databus.nats_subscriber.get_deploy_service_command_coordinator"
		) as coordinator_factory:
		coordinator = MagicMock()
		coordinator.handle = AsyncMock()
		coordinator_factory.return_value = coordinator

		subscribers = await subscribe_to_subjects(connection_manager, session_maker) # type: ignore[arg-type]
		deploy_subscriber = next(
		subscriber
		for subscriber in subscribers
		if subscriber._subject == "command.srm.service.deploy"
		)
		await deploy_subscriber._callback(envelope)

		coordinator_factory.assert_called_once_with(session_maker, connection_manager)
		coordinator.handle.assert_awaited_once_with(envelope)
		assert session_maker.contexts == []


		async def test_subscribe_to_subjects_registers_all_command_subjects(
		connection_manager: MagicMock,
		) -> None:
		subscribers = await subscribe_to_subjects(connection_manager)
		subscribers = await subscribe_to_subjects(connection_manager, FakeSessionMaker()) # type: ignore[arg-type]

		assert sorted(sub._subject for sub in subscribers) == sorted(EXPECTED_COMMAND_SUBJECTS)
		assert connection_manager.client.subscribe.await_count == len(EXPECTED_COMMAND_SUBJECTS)

tests/integration/test_databus.py

+30 −10

Changes for tests/integration/test_databus.py: 30 added lines, 10 removed lines.
Original line number	Diff line number	Diff line
		import asyncio
		import json
		from collections.abc import AsyncIterator, Generator
		from unittest.mock import AsyncMock

		import nats
		import pytest
		@@ -11,7 +12,7 @@ from testcontainers.nats import NatsContainer
		from srm.adapters.databus.nats_connection_manager import NatsConnectionManager
		from srm.adapters.databus.nats_publisher import NatsPublisher
		from srm.api.databus.nats_subscriber import NatsSubscriber, subscribe_to_subjects
		from srm.api.databus.schemas import InboundMessage
		from srm.api.databus.schemas import ValidatedEnvelopeV1
		from srm.config import NatsSettings
		from tests.api.databus.test_nats_subscriber import EXPECTED_COMMAND_SUBJECTS

		@@ -79,40 +80,59 @@ async def test_publisher_publishes_json_message_to_subject(
		assert received == [{"operation_id": "abc-123"}]


		async def test_subscriber_invokes_router_when_message_arrives(
		async def test_subscriber_invokes_callback_when_message_arrives(
		connection_manager: NatsConnectionManager,
		raw_client: Client,
		) -> None:
		received: list[InboundMessage] = []
		received: list[ValidatedEnvelopeV1] = []
		ready = asyncio.Event()

		async def router(message: InboundMessage) -> None:
		received.append(message)
		async def callback(envelope: ValidatedEnvelopeV1) -> None:
		received.append(envelope)
		ready.set()

		subscriber = NatsSubscriber(
		connection_manager=connection_manager,
		subject="command.srm.service.deploy",
		router=router,
		callback=callback,
		)
		await subscriber.start()
		await connection_manager.client.flush()

		await raw_client.publish(
		"command.srm.service.deploy",
		json.dumps({"operation_id": "abc-123"}).encode("utf-8"),
		json.dumps(
		{
		"schema_version": "1.0",
		"operation_id": "11111111-2222-3333-4444-555555555555",
		"correlation_id": "corr-1",
		"requested_at": "2026-06-19T12:00:00Z",
		"app_provider_id": "VideoAppsCo",
		"source": "nbi_camara",
		}
		).encode("utf-8"),
		)
		await asyncio.wait_for(ready.wait(), timeout=2.0)

		assert len(received) == 1
		assert received[0].subject == "command.srm.service.deploy"
		assert json.loads(received[0].payload) == {"operation_id": "abc-123"}
		assert received[0].correlation_id == "corr-1"


		async def test_subscribe_to_subjects_registers_all_command_subjects(
		connection_manager: NatsConnectionManager,
		) -> None:
		subscribers = await subscribe_to_subjects(connection_manager)
		class FakeSessionContext:
		async def __aenter__(self) -> AsyncMock:
		return AsyncMock()

		async def __aexit__(self, *_args: object) -> None:
		return None

		class FakeSessionMaker:
		def __call__(self) -> FakeSessionContext:
		return FakeSessionContext()

		subscribers = await subscribe_to_subjects(connection_manager, FakeSessionMaker()) # type: ignore[arg-type]

		assert sorted(sub._subject for sub in subscribers) == sorted(EXPECTED_COMMAND_SUBJECTS)
		for sub in subscribers: