from __future__ import annotations import argparse import json import sys from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "plugins" / "1c")) sys.path.insert(0, str(ROOT / "plugins" / "1c" / "connector")) import adapter_1c_server as adapter_server # noqa: E402 def problem_codes(result: dict[str, Any]) -> set[str]: return {str(problem.get("code") or "") for problem in result.get("problems") or [] if isinstance(problem, dict)} def require(condition: bool, message: str, failures: list[str]) -> None: if not condition: failures.append(message) def check_blocked_effective_form_path(failures: list[str]) -> None: result = adapter_server.metadata_write( { "base_id": "upo_test", "target": {"canonical_path": "Документ.Заказ.Форма.ФормаДокумента.КнопкаЗаписать"}, "mode": "plan", "edits": [{"property": "Заголовок", "value": "Записать"}], } ) require(result.get("status") == "blocked", "effective form path must be blocked without concrete route", failures) require(result.get("error") == "write_plan_required", "effective form path must return write_plan_required", failures) require((result.get("next_resolution") or {}).get("method") == adapter_server.FORM_WRITE_TARGET_RESOLVE_METHOD, "form path must expose form write target resolver", failures) hint = result.get("apply_payload_hint") if isinstance(result.get("apply_payload_hint"), dict) else {} require(hint.get("ready_for_apply_method") is False, "selector-only form hint must not be ready for apply", failures) def check_module_control_guard(failures: list[str]) -> None: result = adapter_server.metadata_write( { "base_id": "upo_test", "target": { "canonical_path": "ОбщийМодуль.Интеграция.Отправить", "module_ref": "ConfigCASSave:common_module.0#stream:0", }, "intent": {"operation": "replace_with_control", "new": "Сообщить(\"new\");"}, } ) require(result.get("status") == "blocked", "replace_with_control without control fragment must be blocked", failures) require(result.get("error") == "write_plan_blocked", "blocked control guard must return write_plan_blocked", failures) require("missing_control_fragment" in problem_codes(result), "blocked control guard must expose missing_control_fragment", failures) allowed = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": { "kind": "module", "module_ref": "ConfigCASSave:common_module.0#stream:0", }, "intent": {"operation": "replace_with_control", "expected_old_contains": "old", "new": "new", "current_text": "prefix old suffix"}, } ) require(allowed.get("allowed") is True, "replace_with_control with expected_old_contains must be allowed on concrete route", failures) drift = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": { "kind": "module", "module_ref": "ConfigCASSave:common_module.0#stream:0", }, "intent": {"operation": "replace_with_control", "control_fragment": "old", "new": "new", "current_text": "prefix changed suffix"}, } ) require(drift.get("allowed") is False, "replace_with_control drift must be blocked when current_text is provided", failures) require("control_fragment_drift" in problem_codes(drift), "replace_with_control drift must expose control_fragment_drift", failures) def check_reference_identity(failures: list[str]) -> None: module_plan = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": {"kind": "module", "file_name": "object-guid__module-guid.0", "stream_index": 4}, "intent": {"operation": "replace", "old": "a", "new": "b"}, } ) module_hint = ((module_plan.get("route") or {}).get("apply_payload_hint") or {}).get("payload") or {} require(module_plan.get("allowed") is True, "module file_name concrete route must be allowed", failures) require((module_plan.get("target") or {}).get("concrete_reference_field") == "file_name", "module file_name route must preserve concrete_reference_field", failures) require(module_hint.get("file_name") == "object-guid__module-guid.0", "module hint must preserve file_name", failures) require("module_ref" not in module_hint, "module file_name hint must not invent module_ref", failures) form_plan = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": {"kind": "form", "form_guid": "form-guid"}, "intent": {"operation": "property_change", "property": "Заголовок", "value": "Новый"}, } ) form_hint = ((form_plan.get("route") or {}).get("apply_payload_hint") or {}).get("payload") or {} require(form_plan.get("allowed") is True, "form_guid concrete route must be allowed", failures) require((form_plan.get("target") or {}).get("concrete_reference_field") == "form_guid", "form_guid route must preserve concrete_reference_field", failures) require(form_hint.get("form_guid") == "form-guid", "form hint must preserve form_guid", failures) require("file_name" not in form_hint, "form_guid hint must not invent file_name", failures) def check_reference_mismatch(failures: list[str]) -> None: plan = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": {"kind": "module", "form_guid": "form-guid"}, "intent": {"operation": "replace", "old": "a", "new": "b"}, } ) require(plan.get("allowed") is False, "module target with form_guid must be blocked", failures) require("concrete_reference_kind_mismatch" in problem_codes(plan), "module target with form_guid must expose concrete_reference_kind_mismatch", failures) def check_provided_origin_evidence(failures: list[str]) -> None: plan = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": { "canonical_path": "ОбщийМодуль.Интеграция.Отправить", "origin": {"source": "configuration", "presentation": "Конфигурация", "status": "ok"}, }, "intent": {"operation": "replace", "old": "a", "new": "b"}, } ) recommended = (plan.get("route") or {}).get("recommended_write") or {} require((plan.get("origin_lookup") or {}).get("method") == "provided_origin_evidence", "planner must accept provided origin evidence", failures) require(recommended.get("write_surface") == "base_saved_state", "provided configuration origin must recommend base_saved_state", failures) require("write_route_required" in problem_codes(plan), "provided origin must still require concrete write route", failures) cas_plan = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": { "canonical_path": "ОбщийМодуль.Интеграция.Отправить", "origin": {"source": "cas_reference", "status": "owner_unresolved", "write_surface": "requires_owner_resolution"}, }, "intent": {"operation": "replace", "old": "a", "new": "b"}, } ) cas_recommended = (cas_plan.get("route") or {}).get("recommended_write") or {} require(cas_recommended.get("write_surface") == "blocked_unknown", "unresolved CAS origin must be blocked_unknown", failures) require("blocked_unknown" in problem_codes(cas_plan), "unresolved CAS origin must expose blocked_unknown", failures) def check_origin_ambiguity(failures: list[str]) -> None: original_definition_find = adapter_server.metadata_definition_find def fake_definition_find(payload: dict[str, Any]) -> dict[str, Any]: return { "schema": "onec_metadata_definition_find.v1", "status": "ok", "matches": [ { "area": "object", "kind": "Document", "name": "Заказ", "match_by": "synthetic_contract", "location": {"presentation": "Документ.Заказ.Реквизит.КнопкаЗаписать"}, "origin": {"source": "configuration", "status": "ok"}, }, { "area": "form", "kind": "Document", "name": "Заказ", "match_by": "synthetic_contract", "location": {"presentation": "Документ.Заказ.Форма.ФормаДокумента.КнопкаЗаписать"}, "origin": {"source": "configuration", "status": "ok"}, }, ], "counts": {"matches": 2}, } adapter_server.metadata_definition_find = fake_definition_find try: plan = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": {"canonical_path": "Документ.Заказ.Форма.ФормаДокумента.КнопкаЗаписать"}, "intent": {"operation": "property_change", "property": "Заголовок", "value": "Записать"}, } ) finally: adapter_server.metadata_definition_find = original_definition_find require(plan.get("allowed") is False, "ambiguous origin plan must not be allowed", failures) require("ambiguous_origin_matches" in problem_codes(plan), "ambiguous origin plan must expose ambiguous_origin_matches", failures) ambiguity = next((problem for problem in plan.get("problems") or [] if isinstance(problem, dict) and problem.get("code") == "ambiguous_origin_matches"), {}) require(ambiguity.get("match_count") == 2, "ambiguous origin plan must expose match_count", failures) def check_extension_action_evidence(failures: list[str]) -> None: inferred = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": { "kind": "module", "module_ref": "ConfigCASSave:common_module.0#stream:0", "extension_action": {"status": "ok", "operation_class": "replace_with_control"}, }, "intent": {"control_fragment": "old", "new": "new"}, } ) require(inferred.get("allowed") is True, "known extension action with guards must be allowed on concrete route", failures) require((inferred.get("route") or {}).get("operation_class") == "replace_with_control", "planner must infer operation from extension_action", failures) require((inferred.get("route") or {}).get("operation_inferred_from") == "extension_action", "planner must mark operation_inferred_from", failures) unknown = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": { "kind": "module", "module_ref": "ConfigCASSave:common_module.0#stream:0", "extension_action": {"status": "unknown", "operation_class": "unknown_extension_action"}, }, "intent": {"new": "new"}, } ) require(unknown.get("allowed") is False, "unknown extension action must block module write planning", failures) require("extension_action_unknown" in problem_codes(unknown), "unknown extension action must expose extension_action_unknown", failures) mismatch = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": { "kind": "module", "module_ref": "ConfigCASSave:common_module.0#stream:0", "extension_action": {"status": "ok", "operation_class": "insert_after"}, }, "intent": {"operation": "replace", "old": "old", "new": "new"}, } ) require(mismatch.get("allowed") is False, "operation mismatch with extension action must be blocked", failures) require("extension_action_operation_mismatch" in problem_codes(mismatch), "operation mismatch must expose extension_action_operation_mismatch", failures) ambiguous = adapter_server.metadata_write_plan( { "base_id": "upo_test", "target": {"kind": "module", "module_ref": "ConfigCASSave:common_module.0#stream:0"}, "extension_actions": [ {"status": "ok", "operation_class": "insert_before"}, {"status": "ok", "operation_class": "replace"}, ], "intent": {"operation": "replace", "old": "old", "new": "new"}, } ) require(ambiguous.get("allowed") is False, "multiple extension actions must block module write planning", failures) require("extension_action_ambiguous" in problem_codes(ambiguous), "multiple extension actions must expose extension_action_ambiguous", failures) def check_apply_methods_use_write_plan_gate(failures: list[str]) -> None: original_changes_propose = adapter_server.changes_propose original_apply = adapter_server.storage_saved_state_apply_proposal apply_calls: list[dict[str, Any]] = [] def fake_changes_propose(payload: dict[str, Any]) -> dict[str, Any]: source = payload.get("source") if isinstance(payload.get("source"), dict) else {} return { "schema": "onec_change_proposal.v1", "status": "accepted_for_review", "source": { "table": source.get("table") or "ConfigCASSave", "file_name": source.get("file_name") or "common_module.0", }, "original": {"sha1": "0" * 40, "bytes": 10}, "encoded": {"sha1": "1" * 40, "bytes": 11, "payload_hex": "00"}, "validation": {"status": "ok"}, } def fake_apply(payload: dict[str, Any]) -> dict[str, Any]: apply_calls.append(payload) return {"schema": "onec_storage_saved_state_apply.v1", "status": "applied", "applied": True} adapter_server.changes_propose = fake_changes_propose adapter_server.storage_saved_state_apply_proposal = fake_apply try: blocked = adapter_server.metadata_module_write_apply( { "base_id": "upo_test", "module_ref": "ConfigCASSave:common_module.0#stream:0", "mode": "apply", "allow_saved_state_write": True, "allow_sql_saved_state_apply": True, "operation": "replace_with_control", "old": "old", "new": "Сообщить(\"new\");", } ) require(blocked.get("status") == "blocked", "module apply must be blocked when write_plan rejects guards", failures) require(blocked.get("error") == "write_plan_blocked", "module apply block must report write_plan_blocked", failures) require("missing_control_fragment" in problem_codes(blocked), "module apply block must expose missing_control_fragment", failures) require(not apply_calls, "blocked module apply must not call storage_saved_state_apply_proposal", failures) drift_blocked = adapter_server.metadata_module_write_apply( { "base_id": "upo_test", "module_ref": "ConfigCASSave:common_module.0#stream:0", "mode": "apply", "allow_saved_state_write": True, "allow_sql_saved_state_apply": True, "operation": "replace_with_control", "control_fragment": "old", "old": "old", "current_text": "changed", "new": "Сообщить(\"new\");", } ) require(drift_blocked.get("status") == "blocked", "module apply must be blocked when write_plan detects control drift", failures) require(drift_blocked.get("error") == "write_plan_blocked", "module apply drift block must report write_plan_blocked", failures) require("control_fragment_drift" in problem_codes(drift_blocked), "module apply drift block must expose control_fragment_drift", failures) require(not apply_calls, "drift-blocked module apply must not call storage_saved_state_apply_proposal", failures) planned = adapter_server.metadata_module_write_apply( { "base_id": "upo_test", "module_ref": "ConfigCASSave:common_module.0#stream:0", "mode": "plan", "allow_saved_state_write": True, "old": "old", "new": "new", } ) require(planned.get("status") == "planned", "module plan with concrete route must remain planned", failures) require((planned.get("write_plan") or {}).get("allowed") is True, "module plan must include an allowed write_plan", failures) finally: adapter_server.changes_propose = original_changes_propose adapter_server.storage_saved_state_apply_proposal = original_apply def run_checks() -> dict[str, Any]: failures: list[str] = [] check_blocked_effective_form_path(failures) check_module_control_guard(failures) check_reference_identity(failures) check_reference_mismatch(failures) check_provided_origin_evidence(failures) check_origin_ambiguity(failures) check_extension_action_evidence(failures) check_apply_methods_use_write_plan_gate(failures) return { "schema": "onec_write_plan_contract_check.v1", "status": "ok" if not failures else "failed", "failures": failures, "checks": { "blocked_effective_form_path": True, "module_control_guard": True, "reference_identity": True, "reference_mismatch": True, "provided_origin_evidence": True, "origin_ambiguity": True, "extension_action_evidence": True, "apply_methods_use_write_plan_gate": True, }, } def main() -> int: parser = argparse.ArgumentParser(description="Check local 1C metadata write-plan contract invariants.") parser.add_argument("--print", action="store_true") args = parser.parse_args() report = run_checks() if args.print or report["status"] != "ok": print(json.dumps(report, ensure_ascii=False, indent=2)) else: print("1C write-plan contract status: ok") return 0 if report["status"] == "ok" else 1 if __name__ == "__main__": raise SystemExit(main())