63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
from dataclasses import FrozenInstanceError
|
|
|
|
import pytest
|
|
|
|
from tht.execute import ExecResult, PlanSummary
|
|
from tht.mschema.models import PhysicalSchema
|
|
from tht.ports.dwh import (
|
|
DwhAdapter,
|
|
DwhCapabilities,
|
|
DwhHealth,
|
|
DistinctValues,
|
|
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) -> 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) -> DistinctValues:
|
|
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)
|
|
with pytest.raises(FrozenInstanceError):
|
|
capabilities.explain = False
|
|
|
|
|
|
def test_all_contract_types_are_exported_from_public_package():
|
|
from tht.ports import DistinctValues as PublicDistinctValues
|
|
|
|
result = PublicDistinctValues(values=["a"], truncated=True)
|
|
assert result.values == ["a"]
|
|
assert result.truncated is True
|