Commit eb19048c authored by Dimitrios Gogos's avatar Dimitrios Gogos Committed by Paris Stentoumis
Browse files

feat: add failing probe adapter tests for network capability activation

parent fef74486

tests/unit/test_device_targeted_control_path.py

+38 −0
Changes for tests/unit/test_device_targeted_control_path.py: 38 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -37,6 +37,7 @@ from srm.domain.models.topology import ( 
from srm.domain.ports.database.topology import DomainRepository

from srm.domain.ports.transformation_functions import (

    ControlPathRegistry,

    DomainTimeoutError,

    TargetServeability,

    TargetServeabilityRequest,

    TransformationFunctionExecutionPort,
@@ -60,6 +61,11 @@ class _ProbingAdapter(FakeTransformationFunctionAdapter): 
        return self._answer





class _FailingProbeAdapter(FakeTransformationFunctionAdapter):

    def can_serve_target(self, request: TargetServeabilityRequest) -> TargetServeability:

        raise DomainTimeoutError("nef_timeout", "no answer")





class _Registry(ControlPathRegistry):

    def __init__(self, adapters: Mapping[str, TransformationFunctionExecutionPort]) -> None:

        self._adapters = adapters
@@ -361,6 +367,38 @@ async def test_an_unprobed_candidate_prevents_a_verdict_about_the_subscriber() - 
    assert resolution.failure is TargetResolutionFailure.CONTROL_PATH_UNAVAILABLE





async def test_a_probe_that_raises_prevents_a_verdict_about_the_subscriber() -> None:

    failing = _binding("nef-01", is_default=True)

    probed = _binding("nef-02")



    resolution = await _resolver(

        _zone(failing, probed),

        {

            "nef-01": _FailingProbeAdapter(),

            "nef-02": _ProbingAdapter(TargetServeability.UNKNOWN_SUBSCRIBER),

        },

    ).resolve(_request())



    assert resolution.placement is None

    assert resolution.failure is TargetResolutionFailure.CONTROL_PATH_UNAVAILABLE





async def test_a_probe_that_raises_does_not_block_one_that_can_serve() -> None:

    failing = _binding("nef-01", is_default=True)

    working = _binding("nef-02")



    resolution = await _resolver(

        _zone(failing, working),

        {

            "nef-01": _FailingProbeAdapter(),

            "nef-02": _ProbingAdapter(TargetServeability.CAN_SERVE),

        },

    ).resolve(_request())



    assert resolution.placement is not None

    assert resolution.placement.control_path_ref == "nef-02"





async def test_no_topology_candidate_at_all() -> None:

    empty = Zone(

        id=uuid4(),

tests/unit/test_location_query_use_case.py

+19 −1
Changes for tests/unit/test_location_query_use_case.py: 19 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -57,6 +57,8 @@ from srm.domain.ports.transformation_functions import ( 
    ServiceNotApplicable,

    SubscriberNotFound,

    TargetNotLocatable,

    TargetServeability,

    TargetServeabilityRequest,

    TFSDKError,

    TransformationFunctionExecutionPort,

    UnknownControlPathRef,
@@ -93,6 +95,11 @@ class _Adapter(FakeTransformationFunctionAdapter): 
        )





class _FailingProbeAdapter(_Adapter):

    def can_serve_target(self, request: TargetServeabilityRequest) -> TargetServeability:

        raise DomainTimeoutError("nef_timeout", "no answer")





class _Registry(ControlPathRegistry):

    def __init__(self, adapter: TransformationFunctionExecutionPort | None) -> None:

        self._adapter = adapter
@@ -180,6 +187,7 @@ async def _use_case( 
    *,

    adapter: TransformationFunctionExecutionPort | None = None,

    zone: Zone | None = None,

    extra_zones: tuple[Zone, ...] = (),

    requirements: list[ServiceCapabilityRequirement] | None = None,

    specification_state: ServiceSpecificationState = ServiceSpecificationState.ACTIVE,

) -> LocationQueryUseCase:
@@ -196,7 +204,8 @@ async def _use_case( 
    )

    zones = InMemoryZoneRepository()

    resolved_zone = zone if zone is not None else _zone()

    zones.rows[resolved_zone.id] = resolved_zone

    for row in (resolved_zone, *extra_zones):

        zones.rows[row.id] = row



    return LocationQueryUseCase(

        specifications,
@@ -359,3 +368,12 @@ async def test_an_unconfigured_control_path_is_a_fault_not_a_refusal() -> None: 
    outcome = await (await _use_case(adapter=None)).execute(_command())



    assert outcome.failure is LocationQueryFailure.CAPABILITY_EXECUTION_FAILED





async def test_a_probe_that_raises_is_a_fault_not_a_refusal() -> None:

    use_case = await _use_case(adapter=_FailingProbeAdapter(), extra_zones=(_zone(),))



    outcome = await use_case.execute(_command())



    assert outcome.failure is LocationQueryFailure.CAPABILITY_EXECUTION_FAILED

    assert outcome.result is None

tests/unit/test_network_capability_activate_use_case.py

+71 −1
Changes for tests/unit/test_network_capability_activate_use_case.py: 71 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -3,8 +3,12 @@ from datetime import datetime 
from unittest.mock import AsyncMock

from uuid import UUID, uuid4



from srm.application.services.capability_placement import CapabilityPlacement

from srm.application.services.capability_placement import (

    CapabilityPlacement,

    CapabilityPlacementPlanner,

)

from srm.application.services.device_targeted_control_path import (

    DeviceTargetedControlPathResolver,

    DeviceTargetedResolution,

    TargetResolutionFailure,

)
@@ -48,6 +52,27 @@ from srm.domain.models.topology import ( 
    ZoneKind,

    ZoneState,

)

from srm.domain.ports.transformation_functions import (

    ControlPathRegistry,

    DomainTimeoutError,

    TargetServeability,

    TargetServeabilityRequest,

    TransformationFunctionExecutionPort,

)

from tests.unit.fakes import FakeTransformationFunctionAdapter, InMemoryZoneRepository





class _FailingProbeAdapter(FakeTransformationFunctionAdapter):

    def can_serve_target(self, request: TargetServeabilityRequest) -> TargetServeability:

        raise DomainTimeoutError("nef_timeout", "no answer")





class _SingleAdapterRegistry(ControlPathRegistry):

    def __init__(self, adapter: TransformationFunctionExecutionPort) -> None:

        self._adapter = adapter



    def get_adapter(self, control_path_ref: str) -> TransformationFunctionExecutionPort:

        return self._adapter





def _command(
@@ -519,6 +544,51 @@ async def test_accept_fails_before_start_when_network_placement_cannot_be_resolv 
    )





async def test_accept_fails_before_start_when_the_probe_raises() -> None:

    command = _command(uuid4())

    service_orders = AsyncMock()

    service_orders.get_by_operation_id.return_value = None

    service_instances = AsyncMock()

    service_instances.get_by_id.return_value = None

    service_specifications = AsyncMock()

    service_specifications.get_by_id.return_value = _service_specification(

        command.service_specification_id

    )

    requirements = AsyncMock()

    requirements.list_by_service_specification_id.return_value = [

        _network_requirement(command.service_specification_id)

    ]

    # Two zones -> two candidates, so the resolver has to probe.

    zones = InMemoryZoneRepository()

    for zone_id in (uuid4(), uuid4()):

        zones.rows[zone_id] = _zone(zone_id, domains=[_network_domain(uuid4(), zone_id=zone_id)])

    resolver = DeviceTargetedControlPathResolver(

        CapabilityPlacementPlanner(zones),

        _SingleAdapterRegistry(_FailingProbeAdapter()),

    )

    publisher = AsyncMock()



    result = await _use_case(

        service_orders=service_orders,

        service_instances=service_instances,

        service_specifications=service_specifications,

        service_capability_requirements=requirements,

        resolver=resolver,  # type: ignore[arg-type]

        publisher=publisher,

    ).accept(command)



    assert result.accepted_order_id is None

    service_orders.create.assert_not_awaited()

    service_instances.create.assert_not_awaited()

    (status_subject, status), (completed_subject, completed) = [

        call.args for call in publisher.publish.await_args_list

    ]

    assert status_subject == "event.srm.operation.status"

    assert status["state"] == "failed_before_start"

    assert completed_subject == "event.srm.operation.completed"

    assert completed["status"] == "failed"





async def test_complete_creates_capability_instance_and_completed_event() -> None:

    command = _command(uuid4())

    accepted_order = _service_order(command.operation_id)