143 lines
5.6 KiB
Python
143 lines
5.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_core.core.datasources import DatasourceField
|
|
from govoplan_datasources.backend.quality import MAX_DIAGNOSTIC_ROWS, validate_stage
|
|
from govoplan_datasources.backend.schemas import DatasourceStageValidationResponse
|
|
|
|
|
|
class DatasourceQualityTests(unittest.TestCase):
|
|
def test_quality_rules_report_counts_without_exposing_values(self) -> None:
|
|
rows = (
|
|
{"id": 1, "name": "Ada", "amount": 10, "status": "new"},
|
|
{"id": 1, "name": None, "amount": -2, "status": "unknown"},
|
|
)
|
|
schema = (
|
|
DatasourceField("id", "integer", False),
|
|
DatasourceField("name", "string", True),
|
|
DatasourceField("amount", "integer", False),
|
|
DatasourceField("status", "string", False),
|
|
)
|
|
result = validate_stage(
|
|
rows=rows,
|
|
schema=schema,
|
|
quality_policy={
|
|
"version": "case-import-v2",
|
|
"rules": [
|
|
{"id": "rows", "type": "row_count", "minimum": 3},
|
|
{"id": "columns", "type": "required_fields", "fields": ["id", "missing"]},
|
|
{"id": "id-shape", "type": "field", "field": "id", "data_type": "string"},
|
|
{"id": "names", "type": "not_null", "fields": ["name"]},
|
|
{"id": "ids", "type": "unique", "fields": ["id"]},
|
|
{"id": "amount", "type": "range", "field": "amount", "minimum": 0},
|
|
{
|
|
"id": "status",
|
|
"type": "referential",
|
|
"field": "status",
|
|
"allowed_values": ["new", "closed"],
|
|
},
|
|
],
|
|
},
|
|
)
|
|
|
|
self.assertFalse(result["valid"])
|
|
self.assertEqual("case-import-v2", result["policy_version"])
|
|
errors = result["errors"]
|
|
self.assertEqual(
|
|
{
|
|
"quality.row_count",
|
|
"quality.required_fields",
|
|
"quality.field_type",
|
|
"quality.not_null",
|
|
"quality.unique",
|
|
"quality.range",
|
|
"quality.referential",
|
|
},
|
|
{item["code"] for item in errors},
|
|
)
|
|
unique = next(item for item in errors if item["code"] == "quality.unique")
|
|
self.assertEqual(2, unique["affected_rows"])
|
|
self.assertEqual([1, 2], unique["row_numbers"])
|
|
self.assertNotIn("unknown", str(errors))
|
|
contract = DatasourceStageValidationResponse.model_validate(result)
|
|
self.assertEqual("case-import-v2", contract.policy_version)
|
|
self.assertEqual(7, contract.quality.rules_failed)
|
|
|
|
def test_warning_rules_do_not_block_promotion_readiness(self) -> None:
|
|
result = validate_stage(
|
|
rows=({"id": 1},),
|
|
schema=(DatasourceField("id", "integer", False),),
|
|
quality_policy={
|
|
"rules": [
|
|
{
|
|
"id": "large-batch",
|
|
"type": "row_count",
|
|
"minimum": 2,
|
|
"severity": "warning",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
|
|
self.assertTrue(result["valid"])
|
|
self.assertEqual([], result["errors"])
|
|
self.assertEqual("quality.row_count", result["warnings"][0]["code"])
|
|
|
|
def test_schema_changes_are_classified_and_can_be_governed(self) -> None:
|
|
baseline = (
|
|
DatasourceField("id", "integer", False),
|
|
DatasourceField("label", "string", False),
|
|
)
|
|
candidate = (
|
|
DatasourceField("id", "number", False),
|
|
DatasourceField("note", "string", True),
|
|
)
|
|
|
|
blocked = validate_stage(
|
|
rows=(),
|
|
schema=candidate,
|
|
baseline_schema=baseline,
|
|
quality_policy={},
|
|
)
|
|
self.assertFalse(blocked["valid"])
|
|
self.assertEqual("breaking", blocked["schema_change"]["classification"])
|
|
self.assertEqual(
|
|
{"schema.type_widened", "schema.field_removed", "schema.field_added_nullable"},
|
|
{item["code"] for item in blocked["schema_change"]["changes"]},
|
|
)
|
|
|
|
governed = validate_stage(
|
|
rows=(),
|
|
schema=candidate,
|
|
baseline_schema=baseline,
|
|
quality_policy={"schema_policy": {"field_removed": "warning"}},
|
|
)
|
|
self.assertTrue(governed["valid"])
|
|
self.assertEqual("warning", governed["schema_change"]["classification"])
|
|
|
|
def test_malformed_policy_fails_closed_and_hash_is_stable(self) -> None:
|
|
policy = {"rules": [{"id": "bad", "type": "unique", "fields": []}]}
|
|
first = validate_stage(rows=(), schema=(), quality_policy=policy)
|
|
second = validate_stage(rows=(), schema=(), quality_policy=policy)
|
|
|
|
self.assertFalse(first["valid"])
|
|
self.assertEqual("quality.policy_invalid", first["errors"][0]["code"])
|
|
self.assertEqual(first["policy_hash"], second["policy_hash"])
|
|
|
|
def test_diagnostic_row_numbers_are_bounded(self) -> None:
|
|
rows = tuple({"id": None} for _ in range(MAX_DIAGNOSTIC_ROWS + 10))
|
|
result = validate_stage(
|
|
rows=rows,
|
|
schema=(DatasourceField("id", "unknown", True),),
|
|
quality_policy={"rules": [{"id": "id", "type": "not_null", "field": "id"}]},
|
|
)
|
|
error = result["errors"][0]
|
|
self.assertEqual(MAX_DIAGNOSTIC_ROWS + 10, error["affected_rows"])
|
|
self.assertEqual(MAX_DIAGNOSTIC_ROWS, len(error["row_numbers"]))
|
|
self.assertTrue(error["row_numbers_truncated"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|