187 lines
6.6 KiB
Python
187 lines
6.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_core.core.datasources import (
|
|
CAPABILITY_DATASOURCE_CATALOGUE,
|
|
CAPABILITY_DATASOURCE_LIFECYCLE,
|
|
CAPABILITY_DATASOURCE_ORIGINS,
|
|
CAPABILITY_DATASOURCE_PUBLICATION,
|
|
DatasourceCatalogueProvider,
|
|
DatasourceDescriptor,
|
|
DatasourceField,
|
|
DatasourceLifecycleProvider,
|
|
DatasourceMaterialization,
|
|
DatasourceOrigin,
|
|
DatasourceOriginProvider,
|
|
DatasourceOriginReadResult,
|
|
DatasourcePublicationProvider,
|
|
DatasourcePublicationResult,
|
|
DatasourceReadResult,
|
|
DatasourceStage,
|
|
datasource_catalogue,
|
|
datasource_lifecycle,
|
|
datasource_origins,
|
|
datasource_publication,
|
|
)
|
|
from govoplan_core.core.modules import ModuleContext, ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
|
|
|
|
class _Provider:
|
|
descriptor = DatasourceDescriptor(
|
|
ref="datasource:source-1",
|
|
source_name="monthly_cases",
|
|
name="Monthly cases",
|
|
kind="upload",
|
|
mode="static",
|
|
shape="tabular",
|
|
schema=(DatasourceField(name="case_id", data_type="string", nullable=False),),
|
|
fingerprint="abc123",
|
|
)
|
|
|
|
def list_datasources(self, session, principal, *, query="", limit=100):
|
|
del session, principal, query
|
|
return (self.descriptor,)[:limit]
|
|
|
|
def get_datasource(self, session, principal, *, datasource_ref):
|
|
del session, principal
|
|
return self.descriptor if datasource_ref == self.descriptor.ref else None
|
|
|
|
def read_datasource(self, session, principal, *, request):
|
|
del session, principal, request
|
|
return DatasourceReadResult(
|
|
datasource=self.descriptor,
|
|
rows=({"case_id": "0012"},),
|
|
total_rows=1,
|
|
truncated=False,
|
|
)
|
|
|
|
def list_materializations(self, session, principal, *, datasource_ref):
|
|
del session, principal, datasource_ref
|
|
return ()
|
|
|
|
def list_stages(self, session, principal, *, limit=100):
|
|
del session, principal, limit
|
|
return ()
|
|
|
|
def create_stage(self, session, principal, *, stage):
|
|
del session, principal, stage
|
|
return DatasourceStage(
|
|
ref="stage:1",
|
|
name="Stage",
|
|
source_name="stage",
|
|
kind="upload",
|
|
mode="static",
|
|
shape="tabular",
|
|
state="ready",
|
|
)
|
|
|
|
def promote_stage(self, session, principal, *, stage_ref, freeze=False, frozen_label=None):
|
|
del session, principal, stage_ref, freeze, frozen_label
|
|
return self.descriptor, DatasourceMaterialization(
|
|
ref="materialization:1",
|
|
datasource_ref=self.descriptor.ref,
|
|
revision=1,
|
|
state="published",
|
|
fingerprint=self.descriptor.fingerprint,
|
|
)
|
|
|
|
def register_origin(self, session, principal, **kwargs):
|
|
del session, principal, kwargs
|
|
return self.descriptor
|
|
|
|
def refresh_datasource(self, session, principal, *, datasource_ref):
|
|
del session, principal, datasource_ref
|
|
return self.promote_stage(object(), object(), stage_ref="stage:1")
|
|
|
|
def freeze_datasource(self, session, principal, *, datasource_ref, label=None):
|
|
del session, principal, datasource_ref, label
|
|
return self.promote_stage(object(), object(), stage_ref="stage:1")[1]
|
|
|
|
def retire_datasource(self, session, principal, *, datasource_ref):
|
|
del session, principal, datasource_ref
|
|
return self.descriptor
|
|
|
|
def publish_rows(self, session, principal, *, request):
|
|
del session, principal, request
|
|
return DatasourcePublicationResult(
|
|
ref="publication:1",
|
|
status="published",
|
|
datasource=self.descriptor,
|
|
materialization=DatasourceMaterialization(
|
|
ref="materialization:1",
|
|
datasource_ref=self.descriptor.ref,
|
|
revision=1,
|
|
state="published",
|
|
fingerprint=self.descriptor.fingerprint,
|
|
),
|
|
)
|
|
|
|
def list_origins(self, session, principal, *, query="", limit=100):
|
|
del session, principal, query
|
|
return (
|
|
DatasourceOrigin(
|
|
ref="origin:1",
|
|
source_name="origin",
|
|
name="Origin",
|
|
kind="database",
|
|
shape="tabular",
|
|
supported_modes=("live", "cached"),
|
|
provider="test",
|
|
),
|
|
)[:limit]
|
|
|
|
def get_origin(self, session, principal, *, origin_ref):
|
|
return self.list_origins(session, principal)[0] if origin_ref == "origin:1" else None
|
|
|
|
def read_origin(self, session, principal, *, request):
|
|
origin = self.get_origin(session, principal, origin_ref=request.origin_ref)
|
|
return DatasourceOriginReadResult(
|
|
origin=origin,
|
|
rows=({"id": 1},),
|
|
total_rows=1,
|
|
truncated=False,
|
|
)
|
|
|
|
|
|
class DatasourceContractTests(unittest.TestCase):
|
|
def test_capabilities_are_runtime_checkable_and_resolved_without_modules(self) -> None:
|
|
provider = _Provider()
|
|
self.assertIsInstance(provider, DatasourceCatalogueProvider)
|
|
self.assertIsInstance(provider, DatasourceLifecycleProvider)
|
|
self.assertIsInstance(provider, DatasourcePublicationProvider)
|
|
self.assertIsInstance(provider, DatasourceOriginProvider)
|
|
registry = PlatformRegistry()
|
|
registry.register(
|
|
ModuleManifest(
|
|
id="datasource_contract_test",
|
|
name="Datasource contract test",
|
|
version="test",
|
|
capability_factories={
|
|
CAPABILITY_DATASOURCE_CATALOGUE: lambda context: provider,
|
|
CAPABILITY_DATASOURCE_LIFECYCLE: lambda context: provider,
|
|
CAPABILITY_DATASOURCE_PUBLICATION: lambda context: provider,
|
|
CAPABILITY_DATASOURCE_ORIGINS: lambda context: provider,
|
|
},
|
|
)
|
|
)
|
|
registry.configure_capability_context(ModuleContext(registry=registry, settings=object()))
|
|
|
|
self.assertIs(provider, datasource_catalogue(registry))
|
|
self.assertIs(provider, datasource_lifecycle(registry))
|
|
self.assertIs(provider, datasource_publication(registry))
|
|
self.assertIs(provider, datasource_origins(registry))
|
|
self.assertIsNone(datasource_catalogue(PlatformRegistry()))
|
|
|
|
def test_descriptor_distinguishes_mode_kind_shape_and_materialization(self) -> None:
|
|
descriptor = _Provider.descriptor
|
|
|
|
self.assertEqual("static", descriptor.mode)
|
|
self.assertEqual("upload", descriptor.kind)
|
|
self.assertEqual("tabular", descriptor.shape)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|