43 lines
1.7 KiB
Python
43 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
CONNECTOR = Path(__file__).resolve().parents[2] / "plugins" / "1c" / "connector"
|
|
if str(CONNECTOR) not in sys.path:
|
|
sys.path.insert(0, str(CONNECTOR))
|
|
|
|
from write.registry import registered_handlers, select_write_handler # noqa: E402
|
|
from write.context import AdapterWriteContext # noqa: E402
|
|
from write.handlers.scheduled_job import execute as execute_scheduled_job # noqa: E402
|
|
|
|
|
|
def test_registry_selects_one_supported_handler() -> None:
|
|
assert select_write_handler(target_kind="module").key == "module"
|
|
assert select_write_handler(target_kind="форма").key == "form"
|
|
assert select_write_handler(target_kind="object", operation="add_attribute").key == "object_member"
|
|
assert select_write_handler(target_kind="object", operation="synonym").key == "object_property"
|
|
assert select_write_handler(target_kind="anything", is_schedule=True).key == "scheduled_job_schedule"
|
|
|
|
|
|
def test_registry_denies_unknown_target_by_default() -> None:
|
|
assert select_write_handler(target_kind="template") is None
|
|
|
|
|
|
def test_registry_exposes_each_typed_handler_once() -> None:
|
|
handlers = registered_handlers()
|
|
assert [item["key"] for item in handlers] == [
|
|
"module", "form", "object_property", "object_member", "scheduled_job_schedule"
|
|
]
|
|
|
|
|
|
def test_scheduled_job_handler_uses_explicit_context() -> None:
|
|
seen: list[dict] = []
|
|
result = execute_scheduled_job(
|
|
{"base_id": "upo_test"},
|
|
AdapterWriteContext(legacy_scheduled_job_writer=lambda payload: seen.append(payload) or {"status": "planned"}),
|
|
)
|
|
assert result == {"status": "planned"}
|
|
assert seen == [{"base_id": "upo_test"}]
|