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
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",
            normalized_intent="test_query",
            data_types=["generic"],
            purpose="testing",
            confidence=1.0,
        )

    def _make_scope(self, operator: str = "eq") -> 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(), {}
        )
        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("eq"), {}
            )
            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"),
                    {},
                )
            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_returns_error_record(self) -> None:
        """Adapter errors should return an ErrorRecord."""
        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__",
                normalized_intent="__compliance_error_trigger__",
                data_types=["nonexistent"],
                purpose="compliance_error_test",
                confidence=0.0,
            )
            result = await adapter.execute(bad_intent, self._make_scope(), {})
            # Adapter may return AdapterResult or ErrorRecord
            assert isinstance(result, (AdapterResult, ErrorRecord)), (
                f"Error path must return AdapterResult or ErrorRecord, "
                f"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 
64
65
66
67
68
69
70
71
72
73
74
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(), {}
    )
    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 
76
77
78
79
80
81
82
83
84
85
86
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("eq"), {}
        )
        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 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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"),
                {},
            )
        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 
108
109
110
111
112
113
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_returns_error_record() async

Adapter errors should return an ErrorRecord.

Source code in packages/nautilus-adapter-sdk/src/nautilus_adapter_sdk/testing/compliance.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def test_error_path_returns_error_record(self) -> None:
    """Adapter errors should return an ErrorRecord."""
    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__",
            normalized_intent="__compliance_error_trigger__",
            data_types=["nonexistent"],
            purpose="compliance_error_test",
            confidence=0.0,
        )
        result = await adapter.execute(bad_intent, self._make_scope(), {})
        # Adapter may return AdapterResult or ErrorRecord
        assert isinstance(result, (AdapterResult, ErrorRecord)), (
            f"Error path must return AdapterResult or ErrorRecord, "
            f"got {type(result).__name__}"
        )
    finally:
        await adapter.close()