601 lines
28 KiB
Python
601 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "scripts"))
|
|
sys.path.insert(0, str(ROOT / "plugins" / "1c" / "connector"))
|
|
sys.path.insert(0, str(ROOT / "plugins" / "1c" / "mcp"))
|
|
|
|
import adapter_1c_mcp as adapter_mcp # noqa: E402
|
|
import adapter_1c_server as adapter_server # noqa: E402
|
|
import smoke_1c_mcp_selector_chain as selector_chain_smoke # noqa: E402
|
|
|
|
|
|
FORBIDDEN_CONCRETE_SELECTOR_VALUES = (
|
|
"ПечатьЭтикеток",
|
|
"УОП_Печать",
|
|
"ЦенаСоСкидкой",
|
|
"fs_Отчеты",
|
|
)
|
|
PRODUCTION_SELECTOR_CONTRACT_ROOTS = (
|
|
ROOT / "plugins" / "1c",
|
|
ROOT / "docs",
|
|
)
|
|
PRODUCTION_SELECTOR_CONTRACT_SUFFIXES = {
|
|
".css",
|
|
".html",
|
|
".js",
|
|
".json",
|
|
".md",
|
|
".ps1",
|
|
".py",
|
|
".toml",
|
|
".txt",
|
|
".yaml",
|
|
".yml",
|
|
}
|
|
SYNTHETIC_SELECTOR_GUID = "00000000-0000-4000-8000-000000000001"
|
|
PRODUCTION_SELECTOR_CONTRACT_EXCLUDED_PARTS = {
|
|
"__pycache__",
|
|
".pytest_cache",
|
|
"reports",
|
|
"node_modules",
|
|
".git",
|
|
}
|
|
|
|
|
|
def find_tool(name: str) -> dict[str, Any] | None:
|
|
for tool in adapter_mcp.TOOLS:
|
|
if tool.get("name") == name:
|
|
return tool
|
|
return None
|
|
|
|
|
|
def iter_production_selector_contract_files() -> list[Path]:
|
|
files: list[Path] = []
|
|
for root in PRODUCTION_SELECTOR_CONTRACT_ROOTS:
|
|
if not root.exists():
|
|
continue
|
|
for path in root.rglob("*"):
|
|
if not path.is_file():
|
|
continue
|
|
try:
|
|
relative_parts = set(path.relative_to(ROOT).parts)
|
|
except ValueError:
|
|
relative_parts = set(path.parts)
|
|
if relative_parts.intersection(PRODUCTION_SELECTOR_CONTRACT_EXCLUDED_PARTS):
|
|
continue
|
|
if path.suffix.lower() not in PRODUCTION_SELECTOR_CONTRACT_SUFFIXES:
|
|
continue
|
|
files.append(path)
|
|
return sorted(files)
|
|
|
|
|
|
def contract_checks(issues: list[dict[str, Any]], *, onec_request_present: bool) -> dict[str, bool]:
|
|
return {
|
|
"generic_onec_request_schema": onec_request_present
|
|
and not any(
|
|
issue["code"].startswith("mcp_onec_request")
|
|
or issue["code"] in {"mcp_payload_not_open", "mcp_payload_selector_guidance_missing"}
|
|
for issue in issues
|
|
),
|
|
"mcp_tool_selector_guidance": not any(issue["code"] in {"mcp_tool_selector_guidance_missing", "mcp_examples_public_ref_placeholder_missing"} for issue in issues),
|
|
"mcp_saved_state_examples_name_first": not any(issue["code"].startswith("mcp_saved_state_example_") for issue in issues),
|
|
"all_adapter_methods_forward_over_rpc": not any(issue["code"].startswith("mcp_method") or issue["code"] == "mcp_rpc_body_method_mismatch" for issue in issues),
|
|
"no_unified_shadowing_adapter_methods": not any(issue["code"] == "mcp_unified_shadows_adapter_methods" for issue in issues),
|
|
"adapter_help_selector_guidance": not any(issue["code"] == "adapter_help_selector_guidance_missing" for issue in issues),
|
|
"adapter_selector_argument_type_validation": not any(issue["code"] == "adapter_selector_argument_type_not_validated" for issue in issues),
|
|
"adapter_selector_normalizer_type_validation": not any(issue["code"] == "adapter_selector_normalizer_type_not_validated" for issue in issues),
|
|
"adapter_selector_required_message": not any(
|
|
issue["code"] in {"adapter_selector_required_message_incomplete", "adapter_selector_message_constant_incomplete"}
|
|
for issue in issues
|
|
),
|
|
"adapter_selector_presence_aliases": not any(issue["code"] == "adapter_selector_presence_alias_mismatch" for issue in issues),
|
|
"adapter_selector_capability_descriptor": not any(
|
|
issue["code"] in {"adapter_selector_capability_descriptor_mismatch", "adapter_help_selector_capabilities_missing"}
|
|
for issue in issues
|
|
),
|
|
"adapter_definition_read_selector_public_ref": not any(issue["code"] == "adapter_definition_read_selector_public_ref_missing" for issue in issues),
|
|
"adapter_module_read_selector_public_ref": not any(issue["code"] == "adapter_module_read_selector_public_ref_missing" for issue in issues),
|
|
"agent_facing_object_methods_name_first": not any(issue["code"] == "agent_facing_object_method_not_name_first" for issue in issues),
|
|
"adapter_parse_ordinal_unpacked": not any(issue["code"] == "adapter_parse_ordinal_not_unpacked" for issue in issues),
|
|
"adapter_contract_version": not any(
|
|
issue["code"] in {"adapter_contract_version_missing", "mcp_contract_version_mismatch", "adapter_help_contract_version_missing"}
|
|
for issue in issues
|
|
),
|
|
"mcp_policy_blocks_missing_base_id": not any(issue["code"] == "mcp_policy_missing_base_id_not_blocked" for issue in issues),
|
|
"mcp_policy_blocks_diagnostic_fallback": not any(issue["code"] == "mcp_policy_diagnostic_fallback_not_blocked" for issue in issues),
|
|
"mcp_policy_allows_explicit_diagnostics": not any(issue["code"] == "mcp_policy_explicit_diagnostic_not_forwarded" for issue in issues),
|
|
"mcp_selector_schema_template": not any(
|
|
issue["code"] in {"mcp_selector_schema_properties_mismatch", "mcp_schema_requires_object_type_name_selector"}
|
|
for issue in issues
|
|
),
|
|
"selector_chain_live_coverage_schema": not any(issue["code"].startswith("selector_chain_coverage_") for issue in issues),
|
|
"selector_chain_strict_composition_gate": not any(issue["code"].startswith("selector_chain_strict_") for issue in issues),
|
|
"selector_chain_working_state": not any(issue["code"].startswith("selector_chain_working_state_") for issue in issues),
|
|
"no_concrete_selector_values_in_production": not any(issue["code"] == "production_concrete_selector_value_present" for issue in issues),
|
|
}
|
|
|
|
|
|
def check_contract() -> dict[str, Any]:
|
|
issues: list[dict[str, Any]] = []
|
|
methods = [str(row.get("name") or "") for row in adapter_server.METHODS if row.get("name")]
|
|
method_set = set(methods)
|
|
selector_guidance_terms = tuple(getattr(adapter_server, "OBJECT_SELECTOR_GUIDANCE_TERMS", ("ref", "kind/name/guid", "object_type/object_name/object_guid")))
|
|
adapter_contract_version = str(getattr(adapter_server, "ADAPTER_CONTRACT_VERSION", ""))
|
|
mcp_contract_version = str(getattr(adapter_mcp, "MCP_CONTRACT_VERSION", ""))
|
|
if not adapter_contract_version:
|
|
issues.append({"code": "adapter_contract_version_missing"})
|
|
if adapter_contract_version != mcp_contract_version:
|
|
issues.append({"code": "mcp_contract_version_mismatch", "adapter": adapter_contract_version, "mcp": mcp_contract_version})
|
|
duplicate_methods = sorted({name for name in methods if methods.count(name) > 1})
|
|
if duplicate_methods:
|
|
issues.append({"code": "adapter_methods_not_unique", "methods": duplicate_methods})
|
|
shadowed_methods = sorted(method_set.intersection(getattr(adapter_mcp, "UNIFIED_METHODS", set())))
|
|
if shadowed_methods:
|
|
issues.append({"code": "mcp_unified_shadows_adapter_methods", "methods": shadowed_methods})
|
|
mcp_selector_schema_args = set(getattr(adapter_mcp, "OBJECT_SELECTOR_SCHEMA_PROPERTIES", {}))
|
|
adapter_selector_args = set(getattr(adapter_server, "OBJECT_SELECTOR_ARGUMENTS", []))
|
|
if mcp_selector_schema_args != adapter_selector_args:
|
|
issues.append(
|
|
{
|
|
"code": "mcp_selector_schema_properties_mismatch",
|
|
"mcp_arguments": sorted(mcp_selector_schema_args),
|
|
"adapter_arguments": sorted(adapter_selector_args),
|
|
}
|
|
)
|
|
selector_capability_methods = set(getattr(adapter_server, "OBJECT_SELECTOR_METHOD_CAPABILITIES", {}))
|
|
selector_alias_methods = set(getattr(adapter_server, "OBJECT_SELECTOR_ALIAS_METHODS", set()))
|
|
if selector_capability_methods != selector_alias_methods:
|
|
issues.append(
|
|
{
|
|
"code": "adapter_selector_capability_descriptor_mismatch",
|
|
"capability_methods": sorted(selector_capability_methods),
|
|
"alias_methods": sorted(selector_alias_methods),
|
|
}
|
|
)
|
|
mcp_source = Path(adapter_mcp.__file__).read_text(encoding="utf-8")
|
|
adapter_source = Path(adapter_server.__file__).read_text(encoding="utf-8")
|
|
for line_number, line in enumerate(adapter_source.splitlines(), start=1):
|
|
if "parse_ordinal(" not in line or "=" not in line:
|
|
continue
|
|
left_side = line.split("=", 1)[0].strip()
|
|
if "," not in left_side and not left_side.startswith(("return ", "raise ")):
|
|
issues.append(
|
|
{
|
|
"code": "adapter_parse_ordinal_not_unpacked",
|
|
"line": line_number,
|
|
"source": line.strip(),
|
|
}
|
|
)
|
|
if '"required": ["base_id", "object_type", "object_name"]' in mcp_source:
|
|
issues.append({"code": "mcp_schema_requires_object_type_name_selector"})
|
|
normalizer_check = adapter_server.normalize_object_selector_aliases({"kind": 123}, "contract.selector")
|
|
if not isinstance(normalizer_check, dict) or normalizer_check.get("status") != "invalid_argument" or normalizer_check.get("argument") != "kind":
|
|
issues.append(
|
|
{
|
|
"code": "adapter_selector_normalizer_type_not_validated",
|
|
"result": normalizer_check,
|
|
}
|
|
)
|
|
required_message = str(getattr(adapter_server, "OBJECT_SELECTOR_REQUIRED_MESSAGE", ""))
|
|
for term in (*selector_guidance_terms, "ordinal"):
|
|
if term not in required_message:
|
|
issues.append(
|
|
{
|
|
"code": "adapter_selector_required_message_incomplete",
|
|
"term": term,
|
|
"message": required_message,
|
|
}
|
|
)
|
|
selector_message_constants = {
|
|
"OBJECT_SELECTOR_REQUIRED_MESSAGE_NO_ORDINAL": selector_guidance_terms,
|
|
"OBJECT_SELECTOR_GLOBAL_REQUIRED_MESSAGE": (*selector_guidance_terms, "areas metadata/extensions"),
|
|
"OBJECT_SELECTOR_OR_MODULE_REQUIRED_MESSAGE": (*selector_guidance_terms, "module_ref", "module_id"),
|
|
"MODULE_READ_SELECTOR_OR_MODULE_ID_MESSAGE": (*selector_guidance_terms, "ordinal", "module_id"),
|
|
}
|
|
for constant_name, required_terms in selector_message_constants.items():
|
|
message = str(getattr(adapter_server, constant_name, ""))
|
|
for term in required_terms:
|
|
if term not in message:
|
|
issues.append(
|
|
{
|
|
"code": "adapter_selector_message_constant_incomplete",
|
|
"constant": constant_name,
|
|
"term": term,
|
|
"message": message,
|
|
}
|
|
)
|
|
selector_presence_cases = [
|
|
({"ref": "Document.ObjectName"}, True),
|
|
({"object_name": "ObjectName"}, True),
|
|
({"object_guid": SYNTHETIC_SELECTOR_GUID}, True),
|
|
({"kind": "Document"}, False),
|
|
({"object_type": "Document"}, False),
|
|
]
|
|
for payload, expected in selector_presence_cases:
|
|
actual = adapter_server.has_object_selector(payload)
|
|
if actual is not expected:
|
|
issues.append(
|
|
{
|
|
"code": "adapter_selector_presence_alias_mismatch",
|
|
"payload": payload,
|
|
"expected": expected,
|
|
"actual": actual,
|
|
}
|
|
)
|
|
read_selector = adapter_server.definition_read_selector(
|
|
"contract_base",
|
|
{
|
|
"kind": "DataProcessor",
|
|
"name": "ObjectName",
|
|
"guid": SYNTHETIC_SELECTOR_GUID,
|
|
},
|
|
method="metadata.object.get",
|
|
)
|
|
if read_selector.get("ref") != "DataProcessor.ObjectName":
|
|
issues.append(
|
|
{
|
|
"code": "adapter_definition_read_selector_public_ref_missing",
|
|
"selector": read_selector,
|
|
}
|
|
)
|
|
module_read_selector = adapter_server.enrich_selector_with_object_ref(
|
|
{
|
|
"base_id": "contract_base",
|
|
"method": "modules.read",
|
|
"kind": "DataProcessor",
|
|
"guid": SYNTHETIC_SELECTOR_GUID,
|
|
"module_ordinal": 1,
|
|
},
|
|
{
|
|
"kind": "DataProcessor",
|
|
"name": "ObjectName",
|
|
"guid": SYNTHETIC_SELECTOR_GUID,
|
|
},
|
|
)
|
|
if module_read_selector.get("ref") != "DataProcessor.ObjectName" or module_read_selector.get("name") != "ObjectName":
|
|
issues.append(
|
|
{
|
|
"code": "adapter_module_read_selector_public_ref_missing",
|
|
"selector": module_read_selector,
|
|
}
|
|
)
|
|
synthetic_live_steps = [
|
|
{
|
|
"name": "metadata.resolve_overrides",
|
|
"status": "not_found",
|
|
"write_plan_evidence": True,
|
|
"next_method": adapter_server.SAVED_STATE_MODULES_SEARCH_METHOD,
|
|
},
|
|
{
|
|
"name": adapter_server.SAVED_STATE_MODULES_SEARCH_METHOD,
|
|
"status": "ok",
|
|
"modules": 0,
|
|
"write_plan_target": False,
|
|
},
|
|
{
|
|
"name": adapter_server.METADATA_WRITE_PLAN_METHOD,
|
|
"status": "skipped_no_saved_state_target",
|
|
"from_write_plan_target": False,
|
|
},
|
|
]
|
|
live_coverage = selector_chain_smoke.live_coverage_from_steps(synthetic_live_steps)
|
|
expected_coverage = {
|
|
"resolve_overrides": {
|
|
"attempted": True,
|
|
"status": "not_found",
|
|
"write_plan_evidence": True,
|
|
"next_method": adapter_server.SAVED_STATE_MODULES_SEARCH_METHOD,
|
|
},
|
|
"saved_state_resolution": {
|
|
"attempted": True,
|
|
"status": "ok",
|
|
"modules": 0,
|
|
"write_plan_target": False,
|
|
},
|
|
"write_plan_composition": {
|
|
"attempted": True,
|
|
"status": "skipped_no_saved_state_target",
|
|
"composed": False,
|
|
"from_write_plan_target": False,
|
|
},
|
|
}
|
|
for section, expected in expected_coverage.items():
|
|
actual = live_coverage.get(section)
|
|
if actual != expected:
|
|
issues.append(
|
|
{
|
|
"code": "selector_chain_coverage_section_mismatch",
|
|
"section": section,
|
|
"expected": expected,
|
|
"actual": actual,
|
|
}
|
|
)
|
|
if live_coverage.get("skips") != [{"step": adapter_server.METADATA_WRITE_PLAN_METHOD, "status": "skipped_no_saved_state_target"}]:
|
|
issues.append({"code": "selector_chain_coverage_skips_mismatch", "actual": live_coverage.get("skips")})
|
|
strict_skip_issue = selector_chain_smoke.live_write_plan_composition_required_issue(live_coverage)
|
|
if not isinstance(strict_skip_issue, dict) or strict_skip_issue.get("code") != "live_write_plan_composition_required":
|
|
issues.append({"code": "selector_chain_strict_skip_not_blocked", "actual": strict_skip_issue})
|
|
composed_coverage = selector_chain_smoke.live_coverage_from_steps(
|
|
[
|
|
{
|
|
"name": "metadata.resolve_overrides",
|
|
"status": "ok",
|
|
"write_plan_evidence": True,
|
|
"next_method": adapter_server.SAVED_STATE_MODULES_SEARCH_METHOD,
|
|
},
|
|
{
|
|
"name": adapter_server.SAVED_STATE_MODULES_SEARCH_METHOD,
|
|
"status": "ok",
|
|
"modules": 1,
|
|
"write_plan_target": True,
|
|
},
|
|
{
|
|
"name": adapter_server.METADATA_WRITE_PLAN_METHOD,
|
|
"status": "planned",
|
|
"allowed": True,
|
|
"from_write_plan_target": True,
|
|
},
|
|
]
|
|
)
|
|
strict_composed_issue = selector_chain_smoke.live_write_plan_composition_required_issue(composed_coverage)
|
|
if strict_composed_issue is not None:
|
|
issues.append({"code": "selector_chain_strict_composed_blocked", "actual": strict_composed_issue})
|
|
selector_chain_report = selector_chain_smoke.build_report()
|
|
if selector_chain_report.get("passed") is not True:
|
|
issues.append({"code": "selector_chain_working_state_smoke_failed", "report": selector_chain_report})
|
|
for chain in selector_chain_report.get("chains") or []:
|
|
if not isinstance(chain, dict):
|
|
continue
|
|
for step in chain.get("steps") or []:
|
|
if not isinstance(step, dict) or step.get("method") not in selector_chain_smoke.WORKING_STATE_METHODS:
|
|
continue
|
|
payload = step.get("payload") if isinstance(step.get("payload"), dict) else {}
|
|
if payload.get("source_state") != "working" and payload.get("state") != "working":
|
|
issues.append(
|
|
{
|
|
"code": "selector_chain_working_state_missing",
|
|
"chain": chain.get("name"),
|
|
"method": step.get("method"),
|
|
}
|
|
)
|
|
production_selector_files = iter_production_selector_contract_files()
|
|
for path in production_selector_files:
|
|
if not path.exists():
|
|
continue
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
for value in FORBIDDEN_CONCRETE_SELECTOR_VALUES:
|
|
if value in text:
|
|
issues.append(
|
|
{
|
|
"code": "production_concrete_selector_value_present",
|
|
"path": str(path.relative_to(ROOT)),
|
|
"value": value,
|
|
}
|
|
)
|
|
public_methods = {
|
|
str(row.get("name") or ""): row
|
|
for row in adapter_server.call_method("help.methods", {}).get("methods", [])
|
|
if isinstance(row, dict) and row.get("name")
|
|
}
|
|
help_result = adapter_server.call_method("help.methods", {})
|
|
if help_result.get("contract_version") != adapter_contract_version:
|
|
issues.append(
|
|
{
|
|
"code": "adapter_help_contract_version_missing",
|
|
"expected": adapter_contract_version,
|
|
"actual": help_result.get("contract_version"),
|
|
}
|
|
)
|
|
agent_facing_object_terms = ("object", "module", "form", "template", "code")
|
|
for row in adapter_server.METHODS:
|
|
method = str(row.get("name") or "")
|
|
description = str(row.get("description") or "")
|
|
normalized_description = description.casefold()
|
|
if "agent-facing" not in normalized_description:
|
|
continue
|
|
if not any(term in normalized_description for term in agent_facing_object_terms):
|
|
continue
|
|
if method not in selector_alias_methods:
|
|
issues.append(
|
|
{
|
|
"code": "agent_facing_object_method_not_name_first",
|
|
"method": method,
|
|
"description": description,
|
|
}
|
|
)
|
|
for method in sorted(getattr(adapter_server, "OBJECT_SELECTOR_ALIAS_METHODS", set())):
|
|
capabilities = (public_methods.get(method) or {}).get("selector_capabilities")
|
|
if not isinstance(capabilities, dict) or not capabilities.get("accepts_ref") or not capabilities.get("accepts_object_aliases"):
|
|
issues.append(
|
|
{
|
|
"code": "adapter_help_selector_capabilities_missing",
|
|
"method": method,
|
|
"selector_capabilities": capabilities,
|
|
}
|
|
)
|
|
description = str((public_methods.get(method) or {}).get("description") or "")
|
|
for term in selector_guidance_terms:
|
|
if term not in description:
|
|
issues.append(
|
|
{
|
|
"code": "adapter_help_selector_guidance_missing",
|
|
"method": method,
|
|
"term": term,
|
|
"description": description,
|
|
}
|
|
)
|
|
for argument in getattr(adapter_server, "OBJECT_SELECTOR_ARGUMENTS", ["ref"]):
|
|
selector_check = adapter_server.validate_adapter_job_payload(method, {"base_id": "contract_base", argument: 123})
|
|
if (
|
|
not isinstance(selector_check, dict)
|
|
or selector_check.get("status") != "invalid_argument"
|
|
or selector_check.get("argument") != argument
|
|
):
|
|
issues.append(
|
|
{
|
|
"code": "adapter_selector_argument_type_not_validated",
|
|
"method": method,
|
|
"argument": argument,
|
|
"result": selector_check,
|
|
}
|
|
)
|
|
|
|
onec_request = find_tool("onec_request")
|
|
if not onec_request:
|
|
issues.append({"code": "mcp_onec_request_missing"})
|
|
else:
|
|
schema = onec_request.get("inputSchema") or {}
|
|
tool_description = str(onec_request.get("description") or "")
|
|
for term in selector_guidance_terms:
|
|
if term not in tool_description:
|
|
issues.append(
|
|
{
|
|
"code": "mcp_tool_selector_guidance_missing",
|
|
"term": term,
|
|
"description": tool_description,
|
|
}
|
|
)
|
|
examples_json = json.dumps(schema.get("examples") or [], ensure_ascii=False)
|
|
if "<metadata-kind>.<metadata-object-name>" not in examples_json:
|
|
issues.append({"code": "mcp_examples_public_ref_placeholder_missing"})
|
|
examples_by_method = {
|
|
str(example.get("method") or ""): example.get("payload")
|
|
for example in schema.get("examples") or []
|
|
if isinstance(example, dict) and isinstance(example.get("payload"), dict)
|
|
}
|
|
saved_state_example_contract = {
|
|
"metadata.saved_state.prepare": {"required": {"layer", "ref"}, "forbidden": {"target_table", "table", "file_name", "module_ref"}},
|
|
"metadata.saved_state.diff": {"required": {"ref", "module_ordinal"}, "forbidden": {"target_table", "table", "file_name", "module_ref"}},
|
|
"metadata.saved_state.status": {"required": {"layer"}, "forbidden": {"target_table", "table"}},
|
|
"metadata.saved_state.changes.list": {"required": {"layer"}, "forbidden": {"target_table", "table"}},
|
|
}
|
|
for method, requirements in saved_state_example_contract.items():
|
|
example_payload = examples_by_method.get(method)
|
|
if not isinstance(example_payload, dict):
|
|
issues.append({"code": "mcp_saved_state_example_missing", "method": method})
|
|
continue
|
|
missing = sorted(requirements["required"] - set(example_payload))
|
|
forbidden = sorted(requirements["forbidden"].intersection(example_payload))
|
|
if missing:
|
|
issues.append({"code": "mcp_saved_state_example_public_selector_missing", "method": method, "fields": missing})
|
|
if forbidden:
|
|
issues.append({"code": "mcp_saved_state_example_storage_selector_present", "method": method, "fields": forbidden})
|
|
if schema.get("required") != ["method"]:
|
|
issues.append({"code": "mcp_onec_request_required_not_generic", "required": schema.get("required")})
|
|
if "oneOf" in schema:
|
|
issues.append({"code": "mcp_onec_request_enumerates_adapter_methods"})
|
|
payload_schema = (schema.get("properties") or {}).get("payload") or {}
|
|
if payload_schema.get("additionalProperties") is not True:
|
|
issues.append({"code": "mcp_payload_not_open", "payload_schema": payload_schema})
|
|
payload_description = str(payload_schema.get("description") or "")
|
|
for term in selector_guidance_terms:
|
|
if term not in payload_description:
|
|
issues.append(
|
|
{
|
|
"code": "mcp_payload_selector_guidance_missing",
|
|
"term": term,
|
|
"description": payload_description,
|
|
}
|
|
)
|
|
|
|
calls: list[tuple[str, str, Any]] = []
|
|
|
|
def fake_http_json(method: str, path: str, body: Any | None = None, *, timeout: float | None = None, request_id: str | None = None) -> dict[str, Any]:
|
|
calls.append((method, path, body))
|
|
return {"status": "ok", "method": body.get("method") if isinstance(body, dict) else "health"}
|
|
|
|
original_http_json = adapter_mcp.http_json
|
|
previous_diagnostic_mode = os.environ.get("ONEC_MCP_ALLOW_DIAGNOSTIC")
|
|
os.environ["ONEC_MCP_ALLOW_DIAGNOSTIC"] = "true"
|
|
adapter_mcp.http_json = fake_http_json
|
|
try:
|
|
calls.clear()
|
|
missing_base_id_result = adapter_mcp.run_or_enqueue_adapter_method("metadata.write.plan", {})
|
|
if calls or not isinstance(missing_base_id_result, dict) or missing_base_id_result.get("schema") != "adapter_1c_mcp_policy.v1" or missing_base_id_result.get("reason") != "base_id_required":
|
|
issues.append(
|
|
{
|
|
"code": "mcp_policy_missing_base_id_not_blocked",
|
|
"result": missing_base_id_result,
|
|
"calls": calls,
|
|
}
|
|
)
|
|
|
|
calls.clear()
|
|
diagnostic_fallback_result = adapter_mcp.run_or_enqueue_adapter_method("storage.files.list", {"base_id": "contract_base"})
|
|
if calls or not isinstance(diagnostic_fallback_result, dict) or diagnostic_fallback_result.get("schema") != "adapter_1c_mcp_policy.v1" or diagnostic_fallback_result.get("reason") != "diagnostic_method":
|
|
issues.append(
|
|
{
|
|
"code": "mcp_policy_diagnostic_fallback_not_blocked",
|
|
"result": diagnostic_fallback_result,
|
|
"calls": calls,
|
|
}
|
|
)
|
|
|
|
calls.clear()
|
|
adapter_mcp.run_or_enqueue_adapter_method("storage.files.list", {"base_id": "contract_base", "diagnostic": True})
|
|
if not calls:
|
|
issues.append({"code": "mcp_policy_explicit_diagnostic_not_forwarded"})
|
|
|
|
for method in methods:
|
|
calls.clear()
|
|
adapter_mcp.call_adapter_method(method, {"base_id": "contract_base"})
|
|
if method == "health":
|
|
expected = ("GET", "/health?base_id=contract_base")
|
|
if not calls or calls[0][0:2] != expected:
|
|
issues.append({"code": "mcp_health_not_get_health", "method": method, "calls": calls})
|
|
continue
|
|
if not calls:
|
|
issues.append({"code": "mcp_method_not_forwarded", "method": method})
|
|
continue
|
|
http_method, path, body = calls[0]
|
|
if http_method != "POST" or path != "/rpc":
|
|
issues.append({"code": "mcp_method_not_rpc", "method": method, "calls": calls})
|
|
continue
|
|
if not isinstance(body, dict) or body.get("method") != method:
|
|
issues.append({"code": "mcp_rpc_body_method_mismatch", "method": method, "body": body})
|
|
finally:
|
|
adapter_mcp.http_json = original_http_json
|
|
if previous_diagnostic_mode is None:
|
|
os.environ.pop("ONEC_MCP_ALLOW_DIAGNOSTIC", None)
|
|
else:
|
|
os.environ["ONEC_MCP_ALLOW_DIAGNOSTIC"] = previous_diagnostic_mode
|
|
|
|
return {
|
|
"schema": "onec_mcp_adapter_contract_check.v1",
|
|
"passed": not issues,
|
|
"adapter_methods": len(methods),
|
|
"production_selector_files": len(production_selector_files),
|
|
"mcp_tools": [tool.get("name") for tool in adapter_mcp.TOOLS],
|
|
"checks": contract_checks(issues, onec_request_present=onec_request is not None),
|
|
"issues": issues,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Check that adapter-1c MCP stays a generic proxy for REST adapter methods.")
|
|
parser.add_argument("--json", action="store_true", help="Print JSON report.")
|
|
args = parser.parse_args()
|
|
|
|
report = check_contract()
|
|
if args.json:
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
elif report["passed"]:
|
|
print(f"OK: MCP generic proxy contract passed for {report['adapter_methods']} adapter methods.")
|
|
else:
|
|
print(json.dumps(report, ensure_ascii=False, indent=2), file=sys.stderr)
|
|
return 0 if report["passed"] else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|