48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
from tht.execute import ExecResult, PlanSummary
|
|
from tht.mschema.models import PhysicalSchema
|
|
from tht.ports.dwh import (
|
|
DwhAdapter,
|
|
DwhCapabilities,
|
|
DwhHealth,
|
|
UnsupportedCapability,
|
|
)
|
|
|
|
|
|
class FakeDwhAdapter:
|
|
capabilities = DwhCapabilities()
|
|
|
|
def health(self) -> DwhHealth:
|
|
return DwhHealth(ok=True)
|
|
|
|
def introspect(self) -> PhysicalSchema:
|
|
raise NotImplementedError
|
|
|
|
def run_query(self, sql: str, *, limit: int | None = None) -> ExecResult:
|
|
raise NotImplementedError
|
|
|
|
def explain(self, sql: str) -> PlanSummary:
|
|
raise NotImplementedError
|
|
|
|
def sample_column(self, table: str, column: str, *, limit: int) -> list[object]:
|
|
raise NotImplementedError
|
|
|
|
def distinct_values(self, table: str, column: str) -> list[object]:
|
|
raise NotImplementedError
|
|
|
|
|
|
def test_fake_adapter_satisfies_runtime_protocol():
|
|
adapter = FakeDwhAdapter()
|
|
|
|
assert isinstance(adapter, DwhAdapter)
|
|
assert adapter.capabilities.explain is True
|
|
assert adapter.health().ok is True
|
|
|
|
|
|
def test_contract_types_are_public_and_capabilities_are_immutable():
|
|
capabilities = DwhCapabilities()
|
|
|
|
assert capabilities.introspection is True
|
|
assert capabilities.sampling is True
|
|
assert capabilities.distinct_values is True
|
|
assert issubclass(UnsupportedCapability, Exception)
|