130 lines
6.2 KiB
Python
130 lines
6.2 KiB
Python
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 require(condition: bool, message: str, failures: list[str]) -> None:
|
|
if not condition:
|
|
failures.append(message)
|
|
|
|
|
|
def patch_adapter_reads() -> None:
|
|
adapter_server.read_storage_file_bytes = lambda base_id, table, file_name, timeout_seconds=30: (b"fake", {"database": "fake"}, None)
|
|
adapter_server.payload_text_from_bytes = lambda data: {
|
|
"status": "ok",
|
|
"text": "Процедура Проверка()\nКонецПроцедуры",
|
|
}
|
|
adapter_server.extension_module_owner_payload = lambda base_id, module_id, table, timeout_seconds=30: None
|
|
adapter_server.cached_module_owner_payload = lambda base_id, module_id: None
|
|
|
|
|
|
def read_origin(method: str, module_ref: str) -> dict[str, Any]:
|
|
result = adapter_server.call_method(
|
|
method,
|
|
{
|
|
"base_id": "upo_test",
|
|
"module_ref": module_ref,
|
|
"include_text": True,
|
|
"max_chars": 200,
|
|
},
|
|
)
|
|
origin = result.get("origin") if isinstance(result.get("origin"), dict) else {}
|
|
return {"result": result, "origin": origin}
|
|
|
|
|
|
def check_code_search_origin(failures: list[str]) -> None:
|
|
adapter_server.search_modules = lambda payload: {
|
|
"status": "ok",
|
|
"source": {"kind": "live_metadata"},
|
|
"matches": [
|
|
{
|
|
"snippet": {"text": "Процедура Проверка()", "offset": 0},
|
|
"owner": {"status": "unresolved", "kind": None, "name": None},
|
|
"origin": {
|
|
"source": "cas_reference",
|
|
"status": "owner_unresolved",
|
|
"write_surface": "requires_owner_resolution",
|
|
},
|
|
"module": {"name": "Модуль БСЛ"},
|
|
"read_selector": {"base_id": "upo_test", "module_ref": "ConfigCAS:object-module"},
|
|
}
|
|
],
|
|
"counts": {"matches": 1, "complete": True, "scan_limit_hit": False},
|
|
"diagnostics": {},
|
|
}
|
|
result = adapter_server.call_method("code.search", {"base_id": "upo_test", "query": "Проверка"})
|
|
items = result.get("items") if isinstance(result.get("items"), list) else []
|
|
origin = items[0].get("origin") if items and isinstance(items[0], dict) and isinstance(items[0].get("origin"), dict) else {}
|
|
require(result.get("schema") == "onec_code_search.v1", "code.search must return code search schema", failures)
|
|
require(bool(items), "code.search must return patched item", failures)
|
|
require(origin.get("source") == "cas_reference", "code.search item must preserve modules.search origin", failures)
|
|
require(origin.get("write_surface") == "requires_owner_resolution", "code.search item origin must keep write_surface", failures)
|
|
|
|
|
|
def run_checks() -> dict[str, Any]:
|
|
patch_adapter_reads()
|
|
failures: list[str] = []
|
|
|
|
config = read_origin("modules.read", "Config:object-module")
|
|
require(config["result"].get("status") == "ok", "Config module_ref must read in patched contract", failures)
|
|
require(config["origin"].get("source") == "configuration", "Config module_ref must expose configuration origin", failures)
|
|
require(config["origin"].get("write_surface") == "base_saved_state", "Config origin must point writes to base saved-state", failures)
|
|
|
|
save = read_origin("modules.read", "ConfigSave:object-module")
|
|
require(save["result"].get("status") == "ok", "ConfigSave module_ref must read in patched contract", failures)
|
|
require(save["origin"].get("source") == "saved_state", "ConfigSave module_ref must expose saved_state origin", failures)
|
|
require(save["origin"].get("write_surface") == "base_saved_state", "ConfigSave origin must point writes to base saved-state", failures)
|
|
|
|
cas = read_origin("modules.read", "ConfigCAS:object-module")
|
|
require(cas["result"].get("status") == "ok", "ConfigCAS module_ref must read in patched contract", failures)
|
|
require(cas["origin"].get("source") == "cas_reference", "ConfigCAS fallback must not pretend extension/base owner", failures)
|
|
require(cas["origin"].get("status") == "owner_unresolved", "ConfigCAS fallback must require owner resolution", failures)
|
|
require(cas["origin"].get("write_surface") == "requires_owner_resolution", "ConfigCAS fallback must block direct write routing", failures)
|
|
|
|
code_save = read_origin("code.read", "ConfigSave:object-module")
|
|
require(code_save["result"].get("schema") == "onec_code_read.v1", "code.read must wrap modules.read as onec_code_read", failures)
|
|
require(code_save["origin"].get("source") == "saved_state", "code.read must preserve modules.read origin", failures)
|
|
require(code_save["origin"].get("write_surface") == "base_saved_state", "code.read origin must keep write_surface", failures)
|
|
check_code_search_origin(failures)
|
|
|
|
return {
|
|
"schema": "onec_module_origin_contract_check.v1",
|
|
"status": "ok" if not failures else "failed",
|
|
"failures": failures,
|
|
"checks": {
|
|
"config_origin": "Config module_ref exposes configuration origin",
|
|
"configsave_origin": "ConfigSave module_ref exposes saved_state origin",
|
|
"configcas_owner_required": "ConfigCAS module_ref requires owner resolution",
|
|
"code_read_preserves_origin": "code.read preserves origin evidence from modules.read",
|
|
"code_search_preserves_origin": "code.search preserves origin evidence from modules.search",
|
|
},
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Check modules.read origin/provenance 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 module origin contract status: ok")
|
|
return 0 if report["status"] == "ok" else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|