241 lines
8.9 KiB
Python
241 lines
8.9 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT / "plugins" / "1c"))
|
|
sys.path.insert(0, str(ROOT / "plugins" / "1c" / "connector"))
|
|
|
|
import adapter_1c_server # noqa: E402
|
|
from parser.support_rules import parse_parent_configurations_bytes
|
|
|
|
|
|
GUID = "8cec71dd-5046-4b16-ab13-cca39c826ca2"
|
|
CONFIG_GUID = "8cbe2d2d-295e-4b77-813b-63588e84f87a"
|
|
OTHER_GUID = "1fbe6b79-1ea3-43f9-bf72-f6cd4da9743b"
|
|
|
|
|
|
def payload(general_mode: int = 0, rule: int = 1) -> bytes:
|
|
text = (
|
|
f'{{6,0,1,{CONFIG_GUID},{general_mode},{OTHER_GUID},"1.2.3",'
|
|
f'"Поставщик","Конфигурация",1,{rule},0,{GUID},{GUID},0,0}}'
|
|
)
|
|
return text.encode("utf-8-sig")
|
|
|
|
|
|
def test_decodes_editable_support_rule() -> None:
|
|
result = parse_parent_configurations_bytes(payload())
|
|
supplier = result["suppliers"][0]
|
|
assert result["supplier_count"] == 1
|
|
assert supplier.general_mode == "editable"
|
|
assert supplier.name == "Конфигурация"
|
|
assert supplier.rules[0].object_guid == GUID
|
|
assert supplier.rules[0].rule == "editable_support_preserved"
|
|
|
|
|
|
def test_locked_general_mode_overrides_object_rule() -> None:
|
|
result = parse_parent_configurations_bytes(payload(general_mode=1, rule=2))
|
|
supplier = result["suppliers"][0]
|
|
assert supplier.general_mode == "locked"
|
|
assert supplier.rules[0].rule_code == 0
|
|
assert supplier.rules[0].rule == "not_editable"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("data", "message"),
|
|
[
|
|
(b'{5,0,0}', "expected format marker 6"),
|
|
(b'{6,0,1,not-a-guid,0,x,v,p,n,0,0,0}', "configuration GUID is invalid"),
|
|
(payload(rule=9), "unsupported object rule code 9"),
|
|
],
|
|
)
|
|
def test_rejects_invalid_support_payload(data: bytes, message: str) -> None:
|
|
with pytest.raises(ValueError, match=message):
|
|
parse_parent_configurations_bytes(data)
|
|
|
|
|
|
def test_sql_method_decodes_only_explicit_source_and_filters_object(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"read_storage_file_bytes",
|
|
lambda base_id, table, file_name, timeout_seconds=30: (payload(), {"database": "db"}, None),
|
|
)
|
|
result = adapter_1c_server.metadata_support_decode(
|
|
{"base_id": "base", "table": "Params", "file_name": "support.si", "object_guid": GUID}
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert result["source"] == {"kind": "live_sql", "database": "db", "table": "Params", "file_name": "support.si"}
|
|
assert result["matches"][0]["rule"] == "editable_support_preserved"
|
|
assert result["object_support"]["status"] == "editable_support_preserved"
|
|
assert result["object_support"]["edit_allowed_by_support"] is True
|
|
|
|
|
|
def test_sql_method_resolves_public_1c_name_before_filtering_support(monkeypatch) -> None:
|
|
captured: dict[str, str] = {}
|
|
|
|
def get_object(kind, name, **kwargs):
|
|
captured["kind"] = kind
|
|
captured["name"] = name
|
|
return {
|
|
"status": "ok",
|
|
"object": {
|
|
"kind": "CommonModule",
|
|
"name": "ОбменДанными",
|
|
"guid": GUID,
|
|
"ref": "CommonModule.ОбменДанными",
|
|
},
|
|
}
|
|
|
|
monkeypatch.setattr(adapter_1c_server, "get_object", get_object)
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"read_storage_file_bytes",
|
|
lambda base_id, table, file_name, timeout_seconds=30: (payload(), {"database": "db"}, None),
|
|
)
|
|
|
|
result = adapter_1c_server.call_method(
|
|
"metadata.support.decode",
|
|
{
|
|
"base_id": "base",
|
|
"ref": "ОбщиеМодули.ОбменДанными",
|
|
"table": "Params",
|
|
"file_name": "support.si",
|
|
},
|
|
)
|
|
|
|
assert result["status"] == "ok"
|
|
assert captured == {"kind": "CommonModule", "name": "ОбменДанными"}
|
|
assert result["query"]["object_guid"] == GUID
|
|
assert result["query"]["object"]["ref"] == "CommonModule.ОбменДанными"
|
|
assert result["object_support"]["status"] == "editable_support_preserved"
|
|
|
|
|
|
def test_support_decode_help_publishes_optional_name_selector() -> None:
|
|
methods = {
|
|
row["name"]: row
|
|
for row in adapter_1c_server.call_method("help.methods", {})["methods"]
|
|
}
|
|
method = methods["metadata.support.decode"]
|
|
|
|
assert method["selector_capabilities"]["selector_purpose"] == "optional_object_support_filter"
|
|
assert method["selector_capabilities"]["accepts_ordinal"] is False
|
|
assert method["input_schema"]["properties"]["object_name"]["type"] == "string"
|
|
|
|
|
|
def test_sql_method_treats_absent_guid_as_not_supported_after_complete_decode(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"read_storage_file_bytes",
|
|
lambda base_id, table, file_name, timeout_seconds=30: (payload(), {"database": "db"}, None),
|
|
)
|
|
result = adapter_1c_server.metadata_support_decode(
|
|
{"base_id": "base", "table": "Config", "file_name": "support.bin", "object_guid": OTHER_GUID}
|
|
)
|
|
assert result["status"] == "ok"
|
|
assert result["matches"] == []
|
|
assert result["object_support"] == {
|
|
"object_guid": OTHER_GUID,
|
|
"status": "not_supported",
|
|
"edit_allowed_by_support": True,
|
|
"evidence": "absent_from_complete_parent_configurations",
|
|
}
|
|
assert result["diagnostics"]["repository_lock_evaluated"] is False
|
|
|
|
|
|
def test_live_sql_support_gate_resolves_public_object_name(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
adapter_1c_server.repository_control,
|
|
"support_gate",
|
|
lambda request: {"required": True, "allowed": False, "status": "blocked_support_unknown"},
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"get_object",
|
|
lambda kind, name, **kwargs: {"status": "ok", "object": {"guid": GUID}},
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"metadata_support_decode",
|
|
lambda request: {
|
|
"status": "ok",
|
|
"object_support": {
|
|
"status": "not_supported",
|
|
"edit_allowed_by_support": True,
|
|
"evidence": "absent_from_complete_parent_configurations",
|
|
},
|
|
"diagnostics": {"source_discovery": "base_root_descriptor"},
|
|
},
|
|
)
|
|
gate = adapter_1c_server.live_sql_support_gate(
|
|
{"base_id": "base", "layer_id": "base", "ref": "ОбщийМодуль.Модуль"}
|
|
)
|
|
assert gate["allowed"] is True
|
|
assert gate["required"] is False
|
|
assert gate["mode"] == "live_sql"
|
|
assert gate["object_guid"] == GUID
|
|
|
|
|
|
def test_sql_method_discovers_base_source_and_rejects_invalid_payload(monkeypatch) -> None:
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"discover_base_support_source",
|
|
lambda base_id, timeout_seconds=30: (
|
|
{
|
|
"kind": "live_sql",
|
|
"database": "db",
|
|
"table": "Config",
|
|
"file_name": f"{CONFIG_GUID}.4",
|
|
"root_file": OTHER_GUID,
|
|
"configuration_id": CONFIG_GUID,
|
|
"discovery": "base_root_descriptor",
|
|
},
|
|
None,
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"read_storage_file_bytes",
|
|
lambda base_id, table, file_name, timeout_seconds=30: (payload(), {"database": "db"}, None),
|
|
)
|
|
discovered = adapter_1c_server.metadata_support_decode({"base_id": "base", "object_guid": GUID})
|
|
assert discovered["status"] == "ok"
|
|
assert discovered["source"]["file_name"] == f"{CONFIG_GUID}.4"
|
|
assert discovered["diagnostics"]["source_discovery"] == "base_root_descriptor"
|
|
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"discover_extension_support_source",
|
|
lambda base_id, extension_guid, timeout_seconds=30: (
|
|
{
|
|
"kind": "live_sql",
|
|
"table": "ConfigCAS",
|
|
"configuration_id": OTHER_GUID,
|
|
"support_payload_present": False,
|
|
"discovery": "extension_cas_manifest",
|
|
},
|
|
None,
|
|
),
|
|
)
|
|
extension = adapter_1c_server.metadata_support_decode(
|
|
{"base_id": "base", "layer_id": f"extension:{CONFIG_GUID}", "object_guid": GUID}
|
|
)
|
|
assert extension["status"] == "ok"
|
|
assert extension["object_support"]["status"] == "not_supported"
|
|
assert extension["object_support"]["edit_allowed_by_support"] is True
|
|
assert extension["diagnostics"]["repository_lock_evaluated"] is False
|
|
|
|
monkeypatch.setattr(
|
|
adapter_1c_server,
|
|
"read_storage_file_bytes",
|
|
lambda base_id, table, file_name, timeout_seconds=30: (b"{0}", {"database": "db"}, None),
|
|
)
|
|
invalid = adapter_1c_server.metadata_support_decode(
|
|
{"base_id": "base", "table": "Params", "file_name": "not-support.si"}
|
|
)
|
|
assert invalid["status"] == "invalid_format"
|
|
assert invalid["diagnostics"]["permission_inference"] == "forbidden"
|