26 lines
1.1 KiB
Python
26 lines
1.1 KiB
Python
from scripts.rollback_run import preview_records
|
|
|
|
|
|
def test_rollback_preview_is_scoped_and_detects_drift():
|
|
records = [{
|
|
"run_id": "run-1", "db": "ACME", "product_code": "P1",
|
|
"fields": {"IPI": {"before": "0", "after": "5"}}, "outcome": "SUCCEEDED",
|
|
}]
|
|
|
|
preview = preview_records(records, run_id="run-1", database="ACME", field="IPI", product="P1",
|
|
current={("ACME", "P1", "IPI"): "4"})
|
|
|
|
assert preview[0]["target"] == "0"
|
|
assert preview[0]["affected_rows"] == 1
|
|
assert preview[0]["status"] == "CONFLICT"
|
|
assert preview[0]["reason"] == "current != expected_after"
|
|
|
|
|
|
def test_rollback_preview_does_not_include_other_runs_or_fields():
|
|
records = [
|
|
{"run_id": "run-1", "db": "ACME", "product_code": "P1", "fields": {"IPI": {"before": "0", "after": "5"}}, "outcome": "SUCCEEDED"},
|
|
{"run_id": "run-2", "db": "ACME", "product_code": "P1", "fields": {"CEST": {"before": "1", "after": "2"}}, "outcome": "SUCCEEDED"},
|
|
]
|
|
|
|
assert len(preview_records(records, run_id="run-1", database="ACME", field="IPI")) == 1
|