43 lines
1.2 KiB
Python
43 lines
1.2 KiB
Python
import pytest
|
|
|
|
from tht.config import DatabaseConfig, RestConfig
|
|
from tht.execute import ExecutionError
|
|
from tht.ports import DwhAdapter
|
|
|
|
|
|
def postgres_factory():
|
|
from tht.adapters.dwh import PostgresDwhAdapter
|
|
|
|
return PostgresDwhAdapter(
|
|
DatabaseConfig(database="analytics", schema="dw", user="reader", password="secret")
|
|
)
|
|
|
|
|
|
def rest_factory():
|
|
from tht.adapters.dwh import ThothRestDwhAdapter
|
|
|
|
database = DatabaseConfig(
|
|
database="analytics",
|
|
schema="dw",
|
|
user="unused",
|
|
password="unused",
|
|
transport="rest",
|
|
)
|
|
return ThothRestDwhAdapter(
|
|
database,
|
|
RestConfig(base_url="https://dwh.example.test", api_key="secret"),
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("factory", [postgres_factory, rest_factory])
|
|
def test_adapter_rejects_write_sql_without_using_transport(factory):
|
|
with pytest.raises(ExecutionError, match="read-only enforcement"):
|
|
factory().run_query("delete from fact_sales")
|
|
|
|
|
|
@pytest.mark.parametrize("factory", [postgres_factory, rest_factory])
|
|
def test_adapter_satisfies_dwh_protocol(factory):
|
|
adapter = factory()
|
|
assert isinstance(adapter, DwhAdapter)
|
|
assert adapter.capabilities.introspection is True
|