652 lines
24 KiB
Python
652 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
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 as adapter # noqa: E402
|
|
|
|
|
|
OBJECT_GUID = "11111111-2222-3333-4444-555555555555"
|
|
|
|
|
|
def atom(value: Any) -> dict[str, Any]:
|
|
return {"type": "atom", "value": str(value)}
|
|
|
|
|
|
def string(value: Any) -> dict[str, Any]:
|
|
return {"type": "string", "value": str(value)}
|
|
|
|
|
|
def seq(*items: dict[str, Any]) -> dict[str, Any]:
|
|
return {"type": "list", "items": list(items)}
|
|
|
|
|
|
def identity_tree(*, synonym: str = "Товары", comment: str = "Комментарий") -> dict[str, Any]:
|
|
return seq(
|
|
seq(atom(1), atom(0), atom(OBJECT_GUID)),
|
|
string("Номенклатура"),
|
|
seq(atom(1), string("ru"), string(synonym)),
|
|
string(comment),
|
|
atom(0),
|
|
)
|
|
|
|
|
|
def identity_tree_with_member(*, member_synonym: str = "Артикул товара") -> dict[str, Any]:
|
|
return seq(
|
|
seq(atom(1), atom(0), atom(OBJECT_GUID)),
|
|
string("Номенклатура"),
|
|
seq(atom(1), string("ru"), string("Товары")),
|
|
string("Комментарий"),
|
|
seq(
|
|
seq(atom(1), atom(0), atom("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")),
|
|
string("Артикул"),
|
|
seq(atom(1), string("ru"), string(member_synonym)),
|
|
string("Комментарий реквизита"),
|
|
),
|
|
atom(0),
|
|
)
|
|
|
|
|
|
def declared_attribute_tree(*, include_new: bool = False) -> dict[str, Any]:
|
|
records = [
|
|
seq(
|
|
seq(atom(1), atom(0), atom("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")),
|
|
string("Артикул"),
|
|
seq(atom(1), string("ru"), string("Артикул товара")),
|
|
string("Комментарий реквизита"),
|
|
atom("type-settings"),
|
|
)
|
|
]
|
|
if include_new:
|
|
new_guid = adapter.deterministic_member_guid(
|
|
OBJECT_GUID,
|
|
"Attribute",
|
|
"КодПоставщика",
|
|
scope="Catalog.Номенклатура",
|
|
)
|
|
records.append(
|
|
seq(
|
|
seq(atom(1), atom(0), atom(new_guid)),
|
|
string("КодПоставщика"),
|
|
seq(atom(1), string("ru"), string("Код поставщика")),
|
|
string(""),
|
|
atom("type-settings"),
|
|
)
|
|
)
|
|
return seq(
|
|
seq(atom(1), atom(0), atom(OBJECT_GUID)),
|
|
string("Номенклатура"),
|
|
seq(atom(1), string("ru"), string("Товары")),
|
|
string("Комментарий"),
|
|
seq(atom(0), atom(len(records)), *records),
|
|
atom(0),
|
|
)
|
|
|
|
|
|
def prepare_collision() -> dict[str, Any]:
|
|
return {
|
|
"schema": "onec_saved_state_prepare.v1",
|
|
"status": "blocked_target_collision",
|
|
"base_id": "base",
|
|
"target": {"table": "ConfigSave"},
|
|
"object": {
|
|
"kind": "Catalog",
|
|
"name": "Номенклатура",
|
|
"synonym": "Товары",
|
|
"guid": OBJECT_GUID,
|
|
},
|
|
"file_names": ["object.0"],
|
|
"counts": {"file_names": 1, "target_rows": 1},
|
|
}
|
|
|
|
|
|
def base_payload(**overrides: Any) -> dict[str, Any]:
|
|
return {
|
|
"base_id": "base",
|
|
"ref": "Catalog.Номенклатура",
|
|
"property": "synonym",
|
|
"value": "Номенклатура товаров",
|
|
"allow_saved_state_write": True,
|
|
**overrides,
|
|
}
|
|
|
|
|
|
def install_common_mocks(monkeypatch, *, tree: dict[str, Any] | None = None) -> None:
|
|
monkeypatch.setattr(adapter, "repository_apply_gate", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(adapter, "metadata_saved_state_prepare", lambda _payload: prepare_collision())
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"read_storage_file_bytes",
|
|
lambda *_args, **_kwargs: (b"payload", {"database": "db"}, None),
|
|
)
|
|
monkeypatch.setattr(adapter, "parse_config_tree_from_bytes", lambda _data: tree or identity_tree())
|
|
|
|
|
|
def test_identity_property_target_resolves_existing_synonym_and_comment_paths() -> None:
|
|
tree = identity_tree()
|
|
|
|
synonym = adapter.config_tree_identity_property_target(tree, OBJECT_GUID, "синоним", language="ru")
|
|
comment = adapter.config_tree_identity_property_target(tree, OBJECT_GUID, "комментарий")
|
|
|
|
assert synonym["status"] == "ok"
|
|
assert synonym["path"] == "2.2"
|
|
assert synonym["current"] == "Товары"
|
|
assert synonym["node_type"] == "string"
|
|
assert comment["status"] == "ok"
|
|
assert comment["path"] == "3"
|
|
assert comment["current"] == "Комментарий"
|
|
|
|
|
|
def test_member_identity_resolves_full_public_path_by_semantic_category(monkeypatch) -> None:
|
|
member_guid = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
|
|
tree = identity_tree_with_member()
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"nested_metadata_semantic_index",
|
|
lambda *_args, **_kwargs: {
|
|
member_guid: {
|
|
"category": "Attribute",
|
|
"ref": "Catalog.Номенклатура.Attribute.Артикул",
|
|
}
|
|
},
|
|
)
|
|
|
|
resolved = adapter.config_tree_member_identity_resolve(
|
|
tree,
|
|
parent_guid=OBJECT_GUID,
|
|
parent_kind="Catalog",
|
|
parent_name="Номенклатура",
|
|
member_path=["Реквизит", "Артикул"],
|
|
member_name="Артикул",
|
|
)
|
|
|
|
assert resolved["status"] == "ok"
|
|
assert resolved["member"]["guid"] == member_guid
|
|
assert resolved["member"]["kind"] == "Attribute"
|
|
assert resolved["member"]["ref"] == "Catalog.Номенклатура.Attribute.Артикул"
|
|
|
|
|
|
def test_declared_member_record_resolves_append_container() -> None:
|
|
resolved = adapter.config_tree_declared_record_for_guid(
|
|
declared_attribute_tree(),
|
|
"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
|
|
)
|
|
|
|
assert resolved["status"] == "ok"
|
|
assert resolved["record_path"] == "4.2"
|
|
assert resolved["parent_path"] == "4"
|
|
|
|
|
|
def test_object_property_plan_uses_exact_guid_path_and_compare_guard(monkeypatch) -> None:
|
|
install_common_mocks(monkeypatch)
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_propose(payload: dict[str, Any]) -> dict[str, Any]:
|
|
captured.update(payload)
|
|
return {
|
|
"status": "accepted_for_review",
|
|
"source": payload["source"],
|
|
"original": {"sha1": "old"},
|
|
"encoded": {"sha1": "new"},
|
|
"edits": payload["edits"],
|
|
"validation": {"status": "ok", "mode": "path_preserve_format"},
|
|
}
|
|
|
|
monkeypatch.setattr(adapter, "changes_propose", fake_propose)
|
|
|
|
result = adapter.metadata_object_property_write(base_payload(expected_old="Товары"))
|
|
|
|
assert result["status"] == "planned"
|
|
assert result["object"]["ref"] == "Catalog.Номенклатура"
|
|
assert result["property"] == {
|
|
"name": "synonym",
|
|
"language": "ru",
|
|
"current": "Товары",
|
|
"requested": "Номенклатура товаров",
|
|
}
|
|
assert captured["source"]["table"] == "ConfigSave"
|
|
assert captured["edits"] == [
|
|
{
|
|
"path": "2.2",
|
|
"value": "Номенклатура товаров",
|
|
"node_type": "string",
|
|
"expected_old": "Товары",
|
|
"property": "synonym",
|
|
}
|
|
]
|
|
assert "source" not in result["proposal"]
|
|
|
|
|
|
def test_object_property_plan_returns_name_first_prepare_call(monkeypatch) -> None:
|
|
monkeypatch.setattr(adapter, "repository_apply_gate", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"metadata_saved_state_prepare",
|
|
lambda _payload: {
|
|
"status": "plan_ready",
|
|
"base_id": "base",
|
|
"target": {"table": "ConfigSave"},
|
|
"source": {"table": "Config"},
|
|
"object": {"kind": "Catalog", "name": "Номенклатура", "guid": OBJECT_GUID},
|
|
"file_names": ["object.0"],
|
|
"counts": {"source_rows": 1, "target_rows": 0},
|
|
},
|
|
)
|
|
|
|
result = adapter.metadata_object_property_write(base_payload())
|
|
|
|
assert result["status"] == "needs_prepare"
|
|
assert result["next_call"] == {
|
|
"method": "metadata.saved_state.prepare",
|
|
"params": {
|
|
"base_id": "base",
|
|
"layer": "base_saved_state",
|
|
"ref": "Catalog.Номенклатура",
|
|
"kind": "Catalog",
|
|
"name": "Номенклатура",
|
|
"mode": "plan",
|
|
},
|
|
}
|
|
assert "table" not in result["next_call"]["params"]
|
|
assert "file_name" not in result["next_call"]["params"]
|
|
|
|
|
|
def test_object_property_plan_writes_existing_member_by_name(monkeypatch) -> None:
|
|
install_common_mocks(monkeypatch, tree=identity_tree_with_member())
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_propose(payload: dict[str, Any]) -> dict[str, Any]:
|
|
captured.update(payload)
|
|
return {
|
|
"status": "accepted_for_review",
|
|
"source": payload["source"],
|
|
"original": {"sha1": "old"},
|
|
"encoded": {"sha1": "new"},
|
|
"edits": payload["edits"],
|
|
"validation": {"status": "ok", "mode": "path_preserve_format"},
|
|
}
|
|
|
|
monkeypatch.setattr(adapter, "changes_propose", fake_propose)
|
|
|
|
result = adapter.metadata_object_property_write(
|
|
base_payload(
|
|
member_name="Артикул",
|
|
value="Код товара поставщика",
|
|
expected_old="Артикул товара",
|
|
)
|
|
)
|
|
|
|
assert result["status"] == "planned"
|
|
assert result["member"]["name"] == "Артикул"
|
|
assert result["member"]["guid"] == "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"
|
|
assert captured["edits"][0]["path"] == "4.2.2"
|
|
assert captured["edits"][0]["expected_old"] == "Артикул товара"
|
|
|
|
|
|
def test_object_property_apply_verify_and_rollback(monkeypatch) -> None:
|
|
install_common_mocks(monkeypatch)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"metadata_saved_state_prepare",
|
|
lambda _payload: {
|
|
"status": "verified",
|
|
"applied": True,
|
|
"object": {
|
|
"kind": "Catalog",
|
|
"name": "Номенклатура",
|
|
"guid": OBJECT_GUID,
|
|
},
|
|
"file_names": ["object.0"],
|
|
"apply_result": {
|
|
"prepare_receipt": {
|
|
"receipt_id": "a" * 32,
|
|
}
|
|
},
|
|
},
|
|
)
|
|
reads = iter([(b"old", {"database": "db"}, None), (b"new", {"database": "db"}, None)])
|
|
monkeypatch.setattr(adapter, "read_storage_file_bytes", lambda *_args, **_kwargs: next(reads))
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"parse_config_tree_from_bytes",
|
|
lambda data: identity_tree(synonym="Номенклатура товаров" if data == b"new" else "Товары"),
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"changes_propose",
|
|
lambda payload: {
|
|
"status": "accepted_for_review",
|
|
"source": payload["source"],
|
|
"original": {"sha1": "old"},
|
|
"encoded": {"sha1": "new", "payload_hex": "00"},
|
|
"edits": payload["edits"],
|
|
"validation": {"status": "ok", "mode": "path_preserve_format"},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"storage_saved_state_apply_proposal",
|
|
lambda _payload: {"status": "applied", "applied": True, "backup": {"backup_id": "backup-1"}},
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"storage_saved_state_rollback",
|
|
lambda payload: {"status": "applied", "applied": True, "backup_id": payload["backup_id"]},
|
|
)
|
|
prepare_rollback: dict[str, Any] = {}
|
|
|
|
def fake_prepare_rollback(base_id: str, receipt_id: str, *, timeout_seconds: int) -> dict[str, Any]:
|
|
prepare_rollback.update(
|
|
{
|
|
"base_id": base_id,
|
|
"receipt_id": receipt_id,
|
|
"timeout_seconds": timeout_seconds,
|
|
}
|
|
)
|
|
return {"status": "rolled_back", "applied": True}
|
|
|
|
monkeypatch.setattr(adapter, "rollback_saved_state_prepare_receipt", fake_prepare_rollback)
|
|
|
|
result = adapter.metadata_object_property_write(
|
|
base_payload(
|
|
execution_mode="apply_and_rollback",
|
|
allow_sql_saved_state_apply=True,
|
|
allow_sql_saved_state_rollback=True,
|
|
)
|
|
)
|
|
|
|
assert result["status"] == "verified_and_rolled_back"
|
|
assert result["applied"] is True
|
|
assert result["rolled_back"] is True
|
|
assert result["prepare_rollback_result"]["status"] == "rolled_back"
|
|
assert prepare_rollback["receipt_id"] == "a" * 32
|
|
assert result["semantic_verification"] == {
|
|
"status": "ok",
|
|
"expected": "Номенклатура товаров",
|
|
"actual": "Номенклатура товаров",
|
|
}
|
|
|
|
|
|
def test_object_property_write_rejects_rename_and_expected_old_drift(monkeypatch) -> None:
|
|
install_common_mocks(monkeypatch)
|
|
|
|
rename = adapter.call_method(
|
|
adapter.OBJECT_PROPERTY_WRITE_METHOD,
|
|
base_payload(property="name", value="НовоеИмя"),
|
|
)
|
|
drift = adapter.metadata_object_property_write(base_payload(expected_old="Старое значение"))
|
|
|
|
assert rename["status"] == "invalid_argument"
|
|
assert rename["argument"] == "property"
|
|
assert drift["status"] == "precondition_failed"
|
|
assert drift["error"] == "expected_old_mismatch"
|
|
|
|
|
|
def test_metadata_write_routes_object_property_by_public_selector(monkeypatch) -> None:
|
|
monkeypatch.setattr(adapter, "repository_apply_gate", lambda *_args, **_kwargs: None)
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_write(payload: dict[str, Any]) -> dict[str, Any]:
|
|
captured.update(payload)
|
|
return {"status": "planned", "property": {"name": payload["property"]}}
|
|
|
|
monkeypatch.setattr(adapter, "metadata_object_property_write", fake_write)
|
|
|
|
result = adapter.metadata_write(
|
|
{
|
|
"base_id": "base",
|
|
"target": {
|
|
"area": "object",
|
|
"kind": "Catalog",
|
|
"name": "Номенклатура",
|
|
"property": "comment",
|
|
},
|
|
"value": "Пояснение",
|
|
"execution_mode": "plan",
|
|
}
|
|
)
|
|
|
|
assert result["status"] == "planned"
|
|
assert result["routed_method"] == adapter.OBJECT_PROPERTY_WRITE_METHOD
|
|
assert captured["kind"] == "Catalog"
|
|
assert captured["name"] == "Номенклатура"
|
|
assert captured["property"] == "comment"
|
|
assert captured["value"] == "Пояснение"
|
|
assert captured["allow_saved_state_write"] is True
|
|
|
|
|
|
def test_object_member_add_plan_clones_attribute_and_generates_guid(monkeypatch) -> None:
|
|
install_common_mocks(monkeypatch, tree=declared_attribute_tree())
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"nested_metadata_semantic_index",
|
|
lambda *_args, **_kwargs: {
|
|
"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee": {
|
|
"category": "Attribute",
|
|
"ref": "Catalog.Номенклатура.Attribute.Артикул",
|
|
}
|
|
},
|
|
)
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_propose(payload: dict[str, Any]) -> dict[str, Any]:
|
|
captured.update(payload)
|
|
return {
|
|
"status": "accepted_for_review",
|
|
"source": payload["source"],
|
|
"original": {"sha1": "old"},
|
|
"encoded": {"sha1": "new"},
|
|
"edits": [{"mode": "structural_append_child_preserve_format"}],
|
|
"validation": {"status": "ok", "mode": "path_preserve_format"},
|
|
}
|
|
|
|
monkeypatch.setattr(adapter, "changes_propose", fake_propose)
|
|
|
|
result = adapter.metadata_object_member_add(
|
|
{
|
|
"base_id": "base",
|
|
"template_member_ref": "Catalog.Номенклатура.Attribute.Артикул",
|
|
"new_member_name": "КодПоставщика",
|
|
"new_member_synonym": "Код поставщика",
|
|
"allow_saved_state_write": True,
|
|
"execution_mode": "plan",
|
|
}
|
|
)
|
|
|
|
new_guid = adapter.deterministic_member_guid(
|
|
OBJECT_GUID,
|
|
"Attribute",
|
|
"КодПоставщика",
|
|
scope="Catalog.Номенклатура",
|
|
)
|
|
append = captured["edits"][0]["append_child"]
|
|
cloned_identities = adapter.config_tree_identity_records(append["node"])
|
|
assert result["status"] == "planned"
|
|
assert result["requested_member"]["guid"] == new_guid
|
|
assert result["requested_member"]["ref"] == "Catalog.Номенклатура.Attribute.КодПоставщика"
|
|
assert result["container"] == {"ref": "Catalog.Номенклатура", "scope": "object"}
|
|
assert result["requested_member"]["comment"] == ""
|
|
assert result["clone_validation"]["settings_preserved"] is True
|
|
assert result["clone_validation"]["stale_references"] == []
|
|
assert append["parent_path"] == "4"
|
|
assert cloned_identities[new_guid]["name"] == "КодПоставщика"
|
|
assert cloned_identities[new_guid]["synonyms"]["ru"] == "Код поставщика"
|
|
cloned_comment = adapter.config_tree_identity_property_target(
|
|
append["node"],
|
|
new_guid,
|
|
"comment",
|
|
)
|
|
assert cloned_comment["current"] == ""
|
|
|
|
|
|
def test_object_member_clone_blocks_template_identity_references_outside_identity_fields() -> None:
|
|
template = seq(
|
|
seq(atom(1), atom(0), atom("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee")),
|
|
string("Артикул"),
|
|
seq(atom(1), string("ru"), string("Артикул товара")),
|
|
string("Комментарий реквизита"),
|
|
seq(atom("type-settings"), string("Артикул")),
|
|
)
|
|
|
|
cloned, validation = adapter.clone_metadata_member_record(
|
|
template,
|
|
template_guid="aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee",
|
|
new_guid="bbbbbbbb-cccc-dddd-eeee-ffffffffffff",
|
|
new_name="КодПоставщика",
|
|
new_synonym="Код поставщика",
|
|
new_comment="",
|
|
)
|
|
|
|
assert cloned is None
|
|
assert validation["status"] == "blocked"
|
|
assert validation["error"] == "template_identity_referenced_outside_identity_fields"
|
|
assert validation["stale_references"] == [{"path": "4.1", "value": "Артикул"}]
|
|
|
|
|
|
def test_object_member_add_scopes_tabular_column_guid_and_ref(monkeypatch) -> None:
|
|
install_common_mocks(monkeypatch, tree=declared_attribute_tree())
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"nested_metadata_semantic_index",
|
|
lambda *_args, **_kwargs: {
|
|
"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee": {
|
|
"category": "Attribute",
|
|
"ref": "Catalog.Номенклатура.TabularSection.Единицы.Attribute.Артикул",
|
|
}
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"changes_propose",
|
|
lambda payload: {
|
|
"status": "accepted_for_review",
|
|
"source": payload["source"],
|
|
"original": {"sha1": "old"},
|
|
"encoded": {"sha1": "new"},
|
|
"edits": [{"mode": "structural_append_child_preserve_format"}],
|
|
"validation": {"status": "ok", "mode": "path_preserve_format"},
|
|
},
|
|
)
|
|
|
|
result = adapter.metadata_object_member_add(
|
|
{
|
|
"base_id": "base",
|
|
"template_member_ref": "Catalog.Номенклатура.TabularSection.Единицы.Attribute.Артикул",
|
|
"new_member_name": "КодПоставщика",
|
|
"allow_saved_state_write": True,
|
|
"execution_mode": "plan",
|
|
}
|
|
)
|
|
|
|
expected_guid = adapter.deterministic_member_guid(
|
|
OBJECT_GUID,
|
|
"Attribute",
|
|
"КодПоставщика",
|
|
scope="Catalog.Номенклатура.TabularSection.Единицы",
|
|
)
|
|
assert result["status"] == "planned"
|
|
assert result["container"] == {
|
|
"ref": "Catalog.Номенклатура.TabularSection.Единицы",
|
|
"scope": "tabular_section",
|
|
}
|
|
assert result["requested_member"]["ref"] == (
|
|
"Catalog.Номенклатура.TabularSection.Единицы.Attribute.КодПоставщика"
|
|
)
|
|
assert result["requested_member"]["guid"] == expected_guid
|
|
|
|
|
|
def test_object_member_add_apply_verify_and_rollback(monkeypatch) -> None:
|
|
install_common_mocks(monkeypatch, tree=declared_attribute_tree())
|
|
reads = iter([(b"old", {"database": "db"}, None), (b"new", {"database": "db"}, None)])
|
|
monkeypatch.setattr(adapter, "read_storage_file_bytes", lambda *_args, **_kwargs: next(reads))
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"parse_config_tree_from_bytes",
|
|
lambda data: declared_attribute_tree(include_new=data == b"new"),
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"nested_metadata_semantic_index",
|
|
lambda *_args, **_kwargs: {
|
|
"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee": {
|
|
"category": "Attribute",
|
|
"ref": "Catalog.Номенклатура.Attribute.Артикул",
|
|
}
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"changes_propose",
|
|
lambda payload: {
|
|
"status": "accepted_for_review",
|
|
"source": payload["source"],
|
|
"original": {"sha1": "old"},
|
|
"encoded": {"sha1": "new", "payload_hex": "00"},
|
|
"edits": [{"mode": "structural_append_child_preserve_format"}],
|
|
"validation": {"status": "ok", "mode": "path_preserve_format"},
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"storage_saved_state_apply_proposal",
|
|
lambda _payload: {"status": "applied", "applied": True, "backup": {"backup_id": "backup-add"}},
|
|
)
|
|
monkeypatch.setattr(
|
|
adapter,
|
|
"storage_saved_state_rollback",
|
|
lambda _payload: {"status": "applied", "applied": True},
|
|
)
|
|
|
|
result = adapter.metadata_object_member_add(
|
|
{
|
|
"base_id": "base",
|
|
"template_member_ref": "Catalog.Номенклатура.Attribute.Артикул",
|
|
"new_member_name": "КодПоставщика",
|
|
"new_member_synonym": "Код поставщика",
|
|
"allow_saved_state_write": True,
|
|
"allow_sql_saved_state_apply": True,
|
|
"allow_sql_saved_state_rollback": True,
|
|
"execution_mode": "apply_and_rollback",
|
|
}
|
|
)
|
|
|
|
assert result["status"] == "verified_and_rolled_back"
|
|
assert result["semantic_verification"]["status"] == "ok"
|
|
assert result["rolled_back"] is True
|
|
|
|
|
|
def test_metadata_write_routes_add_attribute(monkeypatch) -> None:
|
|
monkeypatch.setattr(adapter, "repository_apply_gate", lambda *_args, **_kwargs: None)
|
|
captured: dict[str, Any] = {}
|
|
|
|
def fake_add(payload: dict[str, Any]) -> dict[str, Any]:
|
|
captured.update(payload)
|
|
return {"status": "planned"}
|
|
|
|
monkeypatch.setattr(adapter, "metadata_object_member_add", fake_add)
|
|
|
|
result = adapter.metadata_write(
|
|
{
|
|
"base_id": "base",
|
|
"target": {
|
|
"area": "object",
|
|
"operation": "add_attribute",
|
|
"template_member_ref": "Catalog.Номенклатура.Attribute.Артикул",
|
|
"new_member_name": "КодПоставщика",
|
|
},
|
|
"mode": "plan",
|
|
}
|
|
)
|
|
|
|
assert result["status"] == "planned"
|
|
assert result["routed_method"] == adapter.OBJECT_MEMBER_ADD_METHOD
|
|
assert captured["template_member_ref"] == "Catalog.Номенклатура.Attribute.Артикул"
|
|
assert captured["new_member_name"] == "КодПоставщика"
|
|
assert captured["allow_saved_state_write"] is True
|