Skip to content

Testing Reference

nautilus_adapter_sdk.testing.compliance

Adapter compliance test suite for validating adapter implementations.

AdapterComplianceSuite

Compliance test harness for adapter implementations.

Parameterised via adapter_factory (a callable returning an Adapter instance) and source_config (the SourceConfig to connect with).

Usage::

suite = AdapterComplianceSuite(
    adapter_factory=lambda: MyAdapter(),
    source_config=SourceConfig(id="s1", type="mytype"),
)
await suite.test_connect_execute_close_lifecycle()
Source code in packages/nautilus-adapter-sdk/src/nautilus_adapter_sdk/testing/compliance.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class AdapterComplianceSuite:
    """Compliance test harness for adapter implementations.

    Parameterised via *adapter_factory* (a callable returning an Adapter
    instance) and *source_config* (the SourceConfig to connect with).

    Usage::

        suite = AdapterComplianceSuite(
            adapter_factory=lambda: MyAdapter(),
            source_config=SourceConfig(id="s1", type="mytype"),
        )
        await suite.test_connect_execute_close_lifecycle()
    """

    def __init__(
        self,
        adapter_factory: Callable[[], Any],
        source_config: SourceConfig,
    ) -> None:
        self.adapter_factory = adapter_factory
        self.source_config = source_config

    # -- helpers --------------------------------------------------------

    def _make_intent(self) -> IntentAnalysis:
        return IntentAnalysis(
            raw_intent="test query",
            data_types_needed=["generic"],
            entities=[],
        )

    def _make_context(self) -> dict[str, Any]:
        """The third argument as the broker builds it.

        The suite used to pass ``{}``, so an adapter that reads its stated
        purpose or session id -- as every shipped adapter does -- was never
        exercised on the path it actually runs on.
        """
        return {"purpose": "testing", "session_id": "compliance", "clearance": "unclassified"}

    def _make_scope(self, operator: str = "=") -> list[ScopeConstraint]:
        return [
            ScopeConstraint(
                source_id=self.source_config.id,
                operator=operator,
                field="id",
                value="test",
            )
        ]

    # -- test methods ---------------------------------------------------

    async def test_connect_execute_close_lifecycle(self) -> None:
        """Test full adapter lifecycle: connect -> execute -> close."""
        adapter: Adapter = self.adapter_factory()
        await adapter.connect(self.source_config)
        result = await adapter.execute(
            self._make_intent(), self._make_scope(), self._make_context()
        )
        assert isinstance(result, AdapterResult), (
            f"execute() must return AdapterResult, got {type(result).__name__}"
        )
        await adapter.close()

    async def test_scope_enforcement_valid_operator(self) -> None:
        """Validate that a valid scope operator is accepted."""
        adapter: Adapter = self.adapter_factory()
        await adapter.connect(self.source_config)
        try:
            result = await adapter.execute(
                self._make_intent(), self._make_scope("="), self._make_context()
            )
            assert isinstance(result, AdapterResult)
        finally:
            await adapter.close()

    async def test_scope_enforcement_invalid_operator(self) -> None:
        """Validate ScopeEnforcementError on invalid operator."""
        adapter: Adapter = self.adapter_factory()
        await adapter.connect(self.source_config)
        try:
            raised = False
            try:
                await adapter.execute(
                    self._make_intent(),
                    self._make_scope("INVALID_OP"),
                    self._make_context(),
                )
            except ScopeEnforcementError:
                raised = True
            assert raised, "execute() with invalid operator must raise ScopeEnforcementError"
        finally:
            await adapter.close()

    async def test_idempotent_close(self) -> None:
        """Calling close() twice must not raise."""
        adapter: Adapter = self.adapter_factory()
        await adapter.connect(self.source_config)
        await adapter.close()
        await adapter.close()  # second call must not error

    async def test_error_path_reports_the_failure(self) -> None:
        """A failing query must report, not return junk.

        Two shapes are compliant, because both are what the broker handles:
        an :class:`AdapterResult` whose ``error`` is populated, or a raised
        :class:`AdapterError`, which ``Broker._execute_adapter`` catches and
        turns into exactly that. Every shipped adapter takes the second path,
        which the previous version of this check failed outright.
        """
        adapter: Adapter = self.adapter_factory()
        await adapter.connect(self.source_config)
        try:
            # Use an impossible intent to trigger error path
            bad_intent = IntentAnalysis(
                raw_intent="__compliance_error_trigger__",
                data_types_needed=["nonexistent"],
                entities=[],
            )
            try:
                result = await adapter.execute(bad_intent, self._make_scope(), self._make_context())
            except AdapterError:
                return
            assert isinstance(result, AdapterResult), (
                f"the error path must return an AdapterResult carrying an "
                f"ErrorRecord, or raise AdapterError; got {type(result).__name__}"
            )
        finally:
            await adapter.close()

test_connect_execute_close_lifecycle() async

Test full adapter lifecycle: connect -> execute -> close.

Source code in packages/nautilus-adapter-sdk/src/nautilus_adapter_sdk/testing/compliance.py 
71
72
73
74
75
76
77
78
79
80
81
async def test_connect_execute_close_lifecycle(self) -> None:
    """Test full adapter lifecycle: connect -> execute -> close."""
    adapter: Adapter = self.adapter_factory()
    await adapter.connect(self.source_config)
    result = await adapter.execute(
        self._make_intent(), self._make_scope(), self._make_context()
    )
    assert isinstance(result, AdapterResult), (
        f"execute() must return AdapterResult, got {type(result).__name__}"
    )
    await adapter.close()

test_scope_enforcement_valid_operator() async

Validate that a valid scope operator is accepted.

Source code in packages/nautilus-adapter-sdk/src/nautilus_adapter_sdk/testing/compliance.py 
83
84
85
86
87
88
89
90
91
92
93
async def test_scope_enforcement_valid_operator(self) -> None:
    """Validate that a valid scope operator is accepted."""
    adapter: Adapter = self.adapter_factory()
    await adapter.connect(self.source_config)
    try:
        result = await adapter.execute(
            self._make_intent(), self._make_scope("="), self._make_context()
        )
        assert isinstance(result, AdapterResult)
    finally:
        await adapter.close()

test_scope_enforcement_invalid_operator() async

Validate ScopeEnforcementError on invalid operator.

Source code in packages/nautilus-adapter-sdk/src/nautilus_adapter_sdk/testing/compliance.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
async def test_scope_enforcement_invalid_operator(self) -> None:
    """Validate ScopeEnforcementError on invalid operator."""
    adapter: Adapter = self.adapter_factory()
    await adapter.connect(self.source_config)
    try:
        raised = False
        try:
            await adapter.execute(
                self._make_intent(),
                self._make_scope("INVALID_OP"),
                self._make_context(),
            )
        except ScopeEnforcementError:
            raised = True
        assert raised, "execute() with invalid operator must raise ScopeEnforcementError"
    finally:
        await adapter.close()

test_idempotent_close() async

Calling close() twice must not raise.

Source code in packages/nautilus-adapter-sdk/src/nautilus_adapter_sdk/testing/compliance.py 
113
114
115
116
117
118
async def test_idempotent_close(self) -> None:
    """Calling close() twice must not raise."""
    adapter: Adapter = self.adapter_factory()
    await adapter.connect(self.source_config)
    await adapter.close()
    await adapter.close()  # second call must not error

test_error_path_reports_the_failure() async

A failing query must report, not return junk.

Two shapes are compliant, because both are what the broker handles: an :class:AdapterResult whose error is populated, or a raised :class:AdapterError, which Broker._execute_adapter catches and turns into exactly that. Every shipped adapter takes the second path, which the previous version of this check failed outright.

Source code in packages/nautilus-adapter-sdk/src/nautilus_adapter_sdk/testing/compliance.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def test_error_path_reports_the_failure(self) -> None:
    """A failing query must report, not return junk.

    Two shapes are compliant, because both are what the broker handles:
    an :class:`AdapterResult` whose ``error`` is populated, or a raised
    :class:`AdapterError`, which ``Broker._execute_adapter`` catches and
    turns into exactly that. Every shipped adapter takes the second path,
    which the previous version of this check failed outright.
    """
    adapter: Adapter = self.adapter_factory()
    await adapter.connect(self.source_config)
    try:
        # Use an impossible intent to trigger error path
        bad_intent = IntentAnalysis(
            raw_intent="__compliance_error_trigger__",
            data_types_needed=["nonexistent"],
            entities=[],
        )
        try:
            result = await adapter.execute(bad_intent, self._make_scope(), self._make_context())
        except AdapterError:
            return
        assert isinstance(result, AdapterResult), (
            f"the error path must return an AdapterResult carrying an "
            f"ErrorRecord, or raise AdapterError; got {type(result).__name__}"
        )
    finally:
        await adapter.close()