feat(core): add datasource lifecycle governance contracts
Module Package Release / publish-packages (push) Successful in 13s

This commit is contained in:
2026-08-22 19:37:44 +02:00
parent 1c3ee9e8c7
commit c6f6faf64f
7 changed files with 66 additions and 7 deletions
+50
View File
@@ -13,6 +13,7 @@ from govoplan_core.core.datasources import (
DatasourceArtifactBackendProvider,
DatasourceDescriptor,
DatasourceField,
DatasourceGovernance,
DatasourceLifecycleProvider,
DatasourceMaterialization,
DatasourceOrigin,
@@ -212,6 +213,55 @@ class DatasourceContractTests(unittest.TestCase):
self.assertEqual("upload", descriptor.kind)
self.assertEqual("tabular", descriptor.shape)
def test_lifecycle_governance_round_trips_without_provider_specific_types(self) -> None:
governance = DatasourceGovernance.from_mapping(
{
"approval_policy": {
"version": "approval-v2",
"required": True,
"required_approvals": 2,
},
"retention_policy": {
"version": "retention-v3",
"enabled": True,
"stage_days": 30,
},
}
)
self.assertEqual("approval-v2", governance.approval_policy["version"])
self.assertEqual(30, governance.retention_policy["stage_days"])
self.assertEqual(
governance.approval_policy,
governance.to_dict()["approval_policy"],
)
self.assertEqual(
governance.retention_policy,
governance.to_dict()["retention_policy"],
)
stage = DatasourceStage(
ref="stage:governed",
name="Governed stage",
source_name="governed",
kind="upload",
mode="static",
shape="tabular",
state="awaiting_approval",
approval={"status": "pending", "policy_version": "approval-v2"},
)
materialization = DatasourceMaterialization(
ref="materialization:disposed",
datasource_ref="datasource:governed",
revision=1,
state="disposed",
fingerprint="abc123",
disposition={"reason": "retention_policy", "policy_version": "retention-v3"},
)
self.assertEqual("pending", stage.approval["status"])
self.assertEqual("retention_policy", materialization.disposition["reason"])
if __name__ == "__main__":
unittest.main()