129 lines
5.5 KiB
Python
129 lines
5.5 KiB
Python
"Fourth-batch partition, literal source structure and isolated DB refresh."
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
|
|
from scripts.audit_conversion_prerequisites import cycles, graph_rows, inventory
|
|
from scripts.audit_library_documents import audit, audit_fourth
|
|
|
|
from calliope.comfyui import source_reviews
|
|
from calliope.comfyui.library import import_library
|
|
from calliope.db import get_db, migrate_db
|
|
from tests.test_workflow_library import ROOT
|
|
|
|
|
|
def test_actual_field_partition_preserves_every_prior_review():
|
|
repo = ROOT.parent
|
|
report = audit(repo, repo / "docs/hel-892/offline")
|
|
result = audit_fourth(repo, report, repo / "docs/hel-892/fourth-evidence/field-partition.json")
|
|
assert result["classification_counts_new"] == {
|
|
"runtime_parameter": 590,
|
|
"frontend_state": 6,
|
|
"current_value_only": 41,
|
|
"semantic_unknown": 48,
|
|
}
|
|
assert report["unnamed_widget_nodes"] == 3701
|
|
assert result["verified_units_ranges_effects"] == 0
|
|
|
|
|
|
def test_conversion_inventory_is_complete_scoped_and_never_readiness_evidence():
|
|
repo = ROOT.parent
|
|
report = inventory(repo)
|
|
assert report["total_ui_files"] == report["remaining_reliable_conversions"] == 146
|
|
assert report["exact_golden_candidates"] == 7
|
|
expected = json.loads(
|
|
(repo / ("docs/hel-892/fourth-evidence/conversion-prerequisites.json")).read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
assert report == expected
|
|
for row in report["files"]:
|
|
raw = (repo / row["path"]).read_bytes()
|
|
assert row["source_sha"] == hashlib.sha1(raw).hexdigest()
|
|
source = json.loads(raw)
|
|
graphs = list(graph_rows(source))
|
|
assert sum(len(g["nodes"]) for g, _, _ in graphs) == len(row["findings"]["slots"])
|
|
for (g, pointer, scope), recorded in zip(graphs, row["graphs"], strict=True):
|
|
assert recorded["links"] == g.get("links", [])
|
|
assert recorded["scope"] == scope
|
|
assert recorded["pointer"] == (pointer or "/")
|
|
for records in row["findings"].values():
|
|
for ref in records:
|
|
if "pointer" not in ref:
|
|
continue
|
|
value = source
|
|
for part in ref["pointer"].strip("/").split("/"):
|
|
value = value[int(part)] if isinstance(value, list) else value[part]
|
|
if "node_id" in ref and "/nodes/" in ref["pointer"]:
|
|
assert ref["node_id"].split("/")[-1] == str(value["id"])
|
|
if "link" in ref:
|
|
assert ref["link"] == value
|
|
assert row["converted_reliably"] is row["executable"] is row["execution_verified"] is False
|
|
assert all(r["status"] == "requires_materials_not_implemented" for r in row["rule_review"])
|
|
|
|
|
|
def test_literal_cycles_are_not_assumed_from_node_names():
|
|
nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
|
|
assert cycles(nodes, [[1, 1, 0, 2, 0, "*"], [2, 2, 0, 3, 0, "*"]]) == []
|
|
assert cycles(nodes, [[1, 1, 0, 2, 0, "*"], [2, 2, 0, 1, 0, "*"]]) == [["1", "2"]]
|
|
assert cycles(nodes, [{"origin_id": 3, "target_id": 3}]) == [["3"]]
|
|
|
|
|
|
async def test_all_16_changed_reviews_refresh_and_user_row_stays_exact(tmp_path, monkeypatch):
|
|
current = source_reviews._reviews()
|
|
old = copy.deepcopy(current)
|
|
revised = {path for path, r in current.items() if r.get("fourth_review")}
|
|
for path in revised:
|
|
review = old[path]
|
|
keys = {tuple(k) for k in review.pop("fourth_review")["fields"]}
|
|
review["reviewed_named_fields"] = [
|
|
f for f in review["reviewed_named_fields"] if (f["node_id"], f["field"]) not in keys
|
|
]
|
|
monkeypatch.setattr(source_reviews, "_reviews", lambda: old)
|
|
db = tmp_path / "fourth.db"
|
|
await migrate_db(db)
|
|
conn = get_db(db)
|
|
try:
|
|
assert import_library(conn, ROOT, source_commit="fixture")["counts"] == {"imported": 156}
|
|
prior = {r["source_repo_path"]: dict(r) for r in conn.execute("SELECT * FROM workflows")}
|
|
protected_path = sorted(revised)[0]
|
|
conn.execute(
|
|
("UPDATE workflows SET description=? WHERE source_repo_path=?"),
|
|
("用户修改保留", protected_path),
|
|
)
|
|
conn.commit()
|
|
protected = dict(
|
|
conn.execute(
|
|
"SELECT * FROM workflows WHERE source_repo_path=?", (protected_path,)
|
|
).fetchone()
|
|
)
|
|
monkeypatch.setattr(source_reviews, "_reviews", lambda: current)
|
|
assert import_library(conn, ROOT, source_commit="fixture")["counts"] == {
|
|
"updated": 15,
|
|
"skipped": 140,
|
|
"conflict": 1,
|
|
}
|
|
assert import_library(conn, ROOT, source_commit="fixture")["counts"] == {
|
|
"skipped": 155,
|
|
"conflict": 1,
|
|
}
|
|
for row in conn.execute("SELECT * FROM workflows"):
|
|
path = row["source_repo_path"]
|
|
if path == protected_path:
|
|
assert dict(row) == protected
|
|
elif path not in revised:
|
|
assert dict(row) == prior[path]
|
|
else:
|
|
doc = json.loads(row["doc_json"])
|
|
assert doc["source_review"]["fourth_review"] == current[path]["fourth_review"]
|
|
assert current[path]["fourth_review"]["operation_check"] in doc["steps"]
|
|
assert (
|
|
doc["source_review"]["reviewed_named_fields"]
|
|
== current[path]["reviewed_named_fields"]
|
|
)
|
|
assert row["is_enabled"] == 0
|
|
assert row["doc_completed_at"] is row["exec_verified_at"] is None
|
|
finally:
|
|
conn.close()
|