from pathlib import Path from uuid import uuid4 import time import zipfile from fastapi.testclient import TestClient from api_server import main from api_server.main import app from one_c_normalizer import ConfigurationRoot, MetadataGroup, MetadataObject, Module, NormalizedProject def create_authoring_session(client: TestClient, project_id: str, task_id: str, session_id: str, user_id: str = "dev.ivan") -> None: user = client.post("/collaboration/users", json={"user_id": user_id, "display_name": user_id}) assert user.status_code == 200 grant = client.post(f"/security/users/{user_id}/roles/developer") assert grant.status_code == 200 task = client.post( "/collaboration/tasks", json={"task_id": task_id, "project_id": project_id, "title": f"Authoring {task_id}", "assignee_user_id": user_id}, ) assert task.status_code == 200 session = client.post( "/collaboration/sessions", json={"session": {"session_id": session_id, "task_id": task_id, "user_id": user_id}}, ) assert session.status_code == 200 def test_cors_allows_lan_panel_origin(): client = TestClient(app) response = client.options( "/health", headers={ "Origin": "http://192.168.200.60:3000", "Access-Control-Request-Method": "GET", }, ) assert response.status_code == 200 assert response.headers["access-control-allow-origin"] == "http://192.168.200.60:3000" def test_metadata_catalog_exposes_1c_tree_structure(): client = TestClient(app) response = client.get("/metadata/catalog") assert response.status_code == 200 payload = response.json() assert "8.3" in payload["platform_family"] assert "Общие модули" in payload["common_branch_children"] document = next(item for item in payload["types"] if item["code"] == "DOCUMENT") common_module = next(item for item in payload["types"] if item["code"] == "COMMON_MODULE") accumulation_register = next(item for item in payload["types"] if item["code"] == "ACCUMULATION_REGISTER") http_service = next(item for item in payload["types"] if item["code"] == "HTTP_SERVICE") report = next(item for item in payload["types"] if item["code"] == "REPORT") catalog = next(item for item in payload["types"] if item["code"] == "CATALOG") url_template = next(item for item in payload["child_object_types"] if item["code"] == "URL_TEMPLATE") data_composition_schema = next(item for item in payload["child_object_types"] if item["code"] == "DATA_COMPOSITION_SCHEMA") assert document["tree_branch"] == "Документы" assert "Реквизиты" in document["child_groups"] assert "Табличные части" in document["child_groups"] assert "Движения" in document["child_groups"] assert "Проведение" in document["properties"] assert "Открыть модуль объекта" in document["context_actions"] assert common_module["tree_branch"] == "Общие модули" assert "Экспортные методы" in common_module["child_groups"] assert "Клиент" in common_module["properties"] assert "Найти вызовы" in common_module["context_actions"] assert accumulation_register["module_kinds"] == ["Модуль набора записей", "Модуль менеджера"] assert "Ресурсы" in accumulation_register["child_groups"] assert "Показать чтение/запись" in accumulation_register["context_actions"] assert http_service["child_groups"] == ["Шаблоны URL", "Методы", "Модуль"] assert "HTTP-методы" in http_service["properties"] assert "Показать URL-шаблоны" in http_service["context_actions"] assert "СКД" in report["child_groups"] assert "Табличные документы" in report["child_groups"] assert "Варианты отчета" in report["child_groups"] assert "Настройки" in report["child_groups"] assert "Справочник" in catalog["description"] assert "HTTP-сервис" in http_service["description"] assert "Шаблоны URL" in url_template["parent_groups"] assert data_composition_schema["parent_groups"] == ["СКД"] def test_html5_server_rendered_project_editor(tmp_path: Path): client = TestClient(app) project_id = f"html5-editor-{uuid4()}" module = tmp_path / "demo_module.bsl" module.write_text( "Процедура Проверить()\n" " Сообщить(\"HTML5\");\n" "КонецПроцедуры\n", encoding="utf-8", ) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 index = client.get("/html5") assert index.status_code == 200 assert "text/html" in index.headers["content-type"] assert 'data-html5-page="projects"' in index.text assert project_id in index.text assert "__next" not in index.text editor = client.get(f"/html5/projects/{project_id}/editor", params={"q": "Проверить"}) assert editor.status_code == 200 assert 'data-html5-page="editor"' in editor.text assert "data-html5-editor" in editor.text assert "data-html5-symbol-results" in editor.text assert "data-html5-symbol-detail" in editor.text assert "data-html5-flowchart" in editor.text assert f'hx-get="/html5/projects/{project_id}/flowchart?depth=1"' in editor.text assert "data-html5-project-report" in editor.text assert f'hx-get="/html5/projects/{project_id}/report"' in editor.text assert "data-html5-review" in editor.text assert f'hx-get="/html5/projects/{project_id}/review"' in editor.text assert "data-html5-authoring-preview" in editor.text assert "data-html5-authoring-preview-form" in editor.text assert f'hx-post="/html5/projects/{project_id}/authoring/completion-preview"' in editor.text assert "data-html5-authoring-diff-form" in editor.text assert f'hx-post="/html5/projects/{project_id}/authoring/semantic-diff-preview"' in editor.text assert "data-html5-metadata-authoring" in editor.text assert "data-html5-metadata-preview-form" in editor.text assert f'hx-post="/html5/projects/{project_id}/authoring/metadata-object-preview"' in editor.text assert "data-html5-authoring-changes" in editor.text assert f'hx-get="/html5/projects/{project_id}/authoring/changes"' in editor.text assert 'hx-get="/html5/projects/' in editor.text assert 'hx-target="[data-html5-symbol-results]"' in editor.text assert 'hx-target="[data-html5-symbol-detail]"' in editor.text assert 'hx-target="[data-html5-source]"' in editor.text assert 'hx-swap="outerHTML"' in editor.text assert "hx-ext=\"sse\"" in editor.text assert f"sse-connect=\"/html5/projects/{project_id}/events\"" in editor.text assert "htmx.org" in editor.text assert "htmx-ext-sse" in editor.text assert "client-js: htmx+sse only" in editor.text assert ".object-actions .button[data-html5-object-action-active" in editor.text assert ".inline-actions" in editor.text assert ".object-breadcrumb" in editor.text assert ".object-summary" in editor.text assert "Проверить" in editor.text assert "__next" not in editor.text with client.stream("GET", f"/html5/projects/{project_id}/events?once=1") as events: first_chunk = next(events.iter_text()) assert "event: status" in first_chunk assert "data:" in first_chunk assert project_id in first_chunk symbols = client.get(f"/html5/projects/{project_id}/symbols", params={"q": "Проверить"}) assert symbols.status_code == 200 assert "text/html" in symbols.headers["content-type"] assert 'data-html5-symbol' in symbols.text assert 'data-html5-lineage-id' in symbols.text assert 'hx-target="[data-html5-symbol-detail]"' in symbols.text assert "Проверить" in symbols.text assert "
""", encoding="utf-8", ) module = tmp_path / "Documents" / "ЗаказПокупателя" / "Ext" / "ObjectModule.bsl" module.parent.mkdir(parents=True) module.write_text( """ Процедура ПровестиКоманда() ПроверитьКонтрагента(); Соединение = Новый HTTPСоединение("api.example.local"); Адрес = "https://api.example.local/orders"; КонецПроцедуры Процедура ПроверитьКонтрагента() КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 snapshot = client.get(f"/projects/{project_id}/snapshot/export").json() handler = next(node for node in snapshot["nodes"] if node["name"] == "ПровестиКоманда") attribute = next(node for node in snapshot["nodes"] if node["name"] == "Контрагент") signal = client.post( f"/projects/{project_id}/runtime/signals", json={ "signal": { "signal_id": "html5-runtime.1", "lineage_id": handler["lineage_id"], "kind": "ERROR", "duration_ms": 125.0, } }, ) assert signal.status_code == 200 knowledge = client.post( "/knowledge", json={ "record_id": f"knowledge.html5.object.{uuid4()}", "scope": "PROJECT", "title": "Правила проведения HTML5", "body": "Контекст проведения заказа для HTML5 inspector.", "related_lineages": [handler["lineage_id"]], }, ) assert knowledge.status_code == 200 marker = client.post( f"/projects/{project_id}/privacy/markers", json={ "target_id": attribute["lineage_id"], "classification": "PERSONAL_DATA", "reason": "Контрагент содержит персональные данные", }, ) assert marker.status_code == 200 editor = client.get(f"/html5/projects/{project_id}/editor") assert editor.status_code == 200 assert "data-html5-object-context" in editor.text assert f'hx-get="/html5/projects/{project_id}/objects/context/%D0%94%D0%BE%D0%BA%D1%83%D0%BC%D0%B5%D0%BD%D1%82.%D0%97%D0%B0%D0%BA%D0%B0%D0%B7%D0%9F%D0%BE%D0%BA%D1%83%D0%BF%D0%B0%D1%82%D0%B5%D0%BB%D1%8F"' in editor.text assert 'hx-target="[data-html5-object-context]"' in editor.text context = client.get(f"/html5/projects/{project_id}/objects/context/Документ.ЗаказПокупателя") assert context.status_code == 200 assert "text/html" in context.headers["content-type"] assert "data-html5-object-context" in context.text assert "Документ.ЗаказПокупателя" in context.text assert "data-html5-object-actions" in context.text assert f"/html5/projects/{project_id}/objects/context/%D0%94%D0%BE%D0%BA%D1%83%D0%BC%D0%B5%D0%BD%D1%82.%D0%97%D0%B0%D0%BA%D0%B0%D0%B7%D0%9F%D0%BE%D0%BA%D1%83%D0%BF%D0%B0%D1%82%D0%B5%D0%BB%D1%8F" in context.text assert f"/projects/{project_id}/objects/schema/%D0%94%D0%BE%D0%BA%D1%83%D0%BC%D0%B5%D0%BD%D1%82.%D0%97%D0%B0%D0%BA%D0%B0%D0%B7%D0%9F%D0%BE%D0%BA%D1%83%D0%BF%D0%B0%D1%82%D0%B5%D0%BB%D1%8F" in context.text assert f"/projects/{project_id}/objects/impact/%D0%94%D0%BE%D0%BA%D1%83%D0%BC%D0%B5%D0%BD%D1%82.%D0%97%D0%B0%D0%BA%D0%B0%D0%B7%D0%9F%D0%BE%D0%BA%D1%83%D0%BF%D0%B0%D1%82%D0%B5%D0%BB%D1%8F" in context.text assert "mode=schema" in context.text assert "mode=impact" in context.text assert "mode=privacy" in context.text assert 'data-html5-object-mode="overview"' in context.text assert "data-html5-object-breadcrumb" in context.text assert "Документ" in context.text assert "ЗаказПокупателя" in context.text assert "data-html5-object-summary" in context.text assert "1 attrs" in context.text assert "1 tables" in context.text assert "1 commands" in context.text assert "1 access rules" in context.text assert "1 runtime signals" in context.text assert "1 privacy markers" in context.text assert "Object context · overview" in context.text assert 'data-html5-object-action-active="true"' in context.text assert 'aria-current="page"' in context.text assert 'hx-target="[data-html5-object-context]"' in context.text assert 'hx-target="[data-html5-flowchart]"' in context.text assert f"/html5/projects/{project_id}/flowchart?focus=%D0%94%D0%BE%D0%BA%D1%83%D0%BC%D0%B5%D0%BD%D1%82.%D0%97%D0%B0%D0%BA%D0%B0%D0%B7%D0%9F%D0%BE%D0%BA%D1%83%D0%BF%D0%B0%D1%82%D0%B5%D0%BB%D1%8F&depth=1" in context.text assert 'hx-target="[data-html5-source]"' in context.text assert 'hx-target="[data-html5-symbol-detail]"' in context.text assert "Контрагент" in context.text assert "Товары" in context.text assert "ФормаДокумента" in context.text assert "Провести" in context.text assert "ПровестиКоманда" in context.text assert "ПроверитьКонтрагента" in context.text assert "HTTPConnection" in context.text assert "https://api.example.local/orders" in context.text assert "OUTBOUND" in context.text assert "data-html5-object-context-item=\"flow-edge\"" in context.text assert "data-html5-flowchart-focus" in context.text assert "data-html5-flowchart-context" in context.text assert "data-html5-flowchart" in context.text assert 'hx-swap-oob="outerHTML"' in context.text assert "Карта связей · focus" in context.text assert "data-html5-source" in context.text assert "data-html5-source-summary" in context.text assert "ObjectModule.bsl" in context.text assert "Соединение = Новый HTTPСоединение" in context.text assert "data-html5-symbol-detail" in context.text assert "data-html5-symbol-summary" in context.text assert "data-html5-symbol-source" in context.text assert "Символ · DOCUMENT" in context.text assert "HAS_ATTRIBUTE" in context.text assert "data-html5-project-report" in context.text assert "Отчет объекта" in context.text assert "server focused summary" in context.text assert "data-html5-object-report-summary" in context.text assert "data links" in context.text assert "data-html5-review" in context.text assert "data-html5-review-summary" in context.text assert "data-html5-review-source" in context.text assert "Review объекта" in context.text assert "External integration endpoint" in context.text assert "1 signals" in context.text assert "1 errors" in context.text assert "125.0 ms" in context.text assert "Правила проведения HTML5" in context.text assert "Контекст проведения заказа" in context.text assert "PERSONAL_DATA" in context.text assert "Контрагент содержит персональные данные" in context.text assert "Роль.Менеджер" in context.text assert "read, write, post" in context.text or "post, read, write" in context.text assert "= 5 assert payload["form_count"] >= 2 assert payload["role_count"] >= 1 setup = client.get(f"/projects/{project_id}/setup") assert setup.status_code == 200 assert setup.json()["status"] == "INDEXED" assert setup.json()["last_import"]["source_path"] assert setup.json()["import_history"][0]["status"] == "mock_indexed" history = client.get(f"/projects/{project_id}/imports/history") assert history.status_code == 200 assert history.json()[0]["source"] == "XML_DUMP" tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 assert tree.json()["root"]["kind"] == "PROJECT" reindexed = client.post(f"/projects/{project_id}/reindex") assert reindexed.status_code == 200 assert reindexed.json()["status"] == "reindexed" def test_windows_agent_import_job_protocol_applies_server_result(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) project_id = f"agent-import-{uuid4()}" agent_id = f"win-agent-{uuid4()}" client.post("/agent/heartbeat", json={"agent_id": agent_id, "hostname": "test-host"}) created = client.post( f"/projects/{project_id}/imports/EDT_PROJECT/agent-jobs", json={ "agent_id": agent_id, "source": "EDT_PROJECT", "local_path": r"D:\edt\upo", "bin_path": r"C:\Program Files\1cv8\8.3.24.0000\bin\1cv8.exe", "mode": "FULL_REPLACE", }, ) assert created.status_code == 200 job_id = created.json()["job_id"] assert created.json()["status"] == "QUEUED" claimed = client.get("/agent/jobs/next", params={"agent_id": agent_id}) assert claimed.status_code == 200 assert claimed.json()["job_id"] == job_id assert claimed.json()["status"] == "RUNNING" assert claimed.json()["local_path"] == r"D:\edt\upo" completed = client.post( f"/agent/jobs/{job_id}/result", json={ "status": "SUCCEEDED", "server_path": str(tmp_path), "logs": ["EDT exported and uploaded."], }, ) assert completed.status_code == 200 deadline = time.monotonic() + 10 payload = completed.json() while time.monotonic() < deadline and payload["status"] == "RUNNING": time.sleep(0.05) jobs = client.get(f"/projects/{project_id}/imports/agent-jobs") assert jobs.status_code == 200 payload = next(item for item in jobs.json() if item["job_id"] == job_id) assert payload["status"] == "SUCCEEDED" assert payload["import_summary"]["status"] == "structure_indexed" assert payload["import_summary"]["object_count"] == 1 setup = client.get(f"/projects/{project_id}/setup") assert setup.status_code == 200 assert setup.json()["last_import"]["source"] == "EDT_PROJECT" def test_windows_agent_import_job_accepts_uploaded_zip(tmp_path: Path): payload_root = tmp_path / "payload" payload_root.mkdir() (payload_root / "metadata.xml").write_text( """ """, encoding="utf-8", ) archive_path = tmp_path / "edt.zip" with zipfile.ZipFile(archive_path, "w") as archive: archive.write(payload_root / "metadata.xml", "metadata.xml") client = TestClient(app) project_id = f"agent-upload-{uuid4()}" agent_id = f"win-agent-{uuid4()}" client.post("/agent/heartbeat", json={"agent_id": agent_id, "hostname": "test-host"}) created = client.post( f"/projects/{project_id}/imports/EDT_PROJECT/agent-jobs", json={"agent_id": agent_id, "source": "EDT_PROJECT", "local_path": r"D:\edt\upo", "mode": "FULL_REPLACE"}, ) assert created.status_code == 200 job_id = created.json()["job_id"] uploaded = client.post( f"/agent/jobs/{job_id}/upload", params={"filename": "edt.zip"}, content=archive_path.read_bytes(), headers={"Content-Type": "application/octet-stream"}, ) assert uploaded.status_code == 200 assert uploaded.json()["server_path"] deadline = time.monotonic() + 10 payload = uploaded.json() while time.monotonic() < deadline and payload["status"] == "RUNNING": time.sleep(0.05) jobs = client.get(f"/projects/{project_id}/imports/agent-jobs") assert jobs.status_code == 200 payload = next(item for item in jobs.json() if item["job_id"] == job_id) assert payload["status"] == "SUCCEEDED" assert payload["import_summary"]["object_count"] == 1 def test_windows_agent_cf_export_job_accepts_infobase_settings_without_local_path(): client = TestClient(app) project_id = f"agent-cf-{uuid4()}" agent_id = f"win-agent-{uuid4()}" client.post("/agent/heartbeat", json={"agent_id": agent_id, "hostname": "test-host"}) created = client.post( f"/projects/{project_id}/imports/CF_FILE/agent-jobs", json={ "agent_id": agent_id, "source": "CF_FILE", "mode": "FULL_REPLACE", "metadata": { "one_c_server": "192.168.200.95", "one_c_infobase": "upo", "one_c_user": "import-user", "one_c_password": "secret", }, }, ) assert created.status_code == 200 payload = created.json() assert payload["source"] == "CF_FILE" assert payload["local_path"] is None assert payload["metadata"]["one_c_infobase"] == "upo" def test_windows_agent_browse_request_protocol(): client = TestClient(app) agent_id = f"win-agent-{uuid4()}" created = client.post("/agent/browse-requests", json={"agent_id": agent_id, "path": r"D:\EDT"}) assert created.status_code == 200 request_id = created.json()["request_id"] assert created.json()["status"] == "QUEUED" claimed = client.get("/agent/browse/next", params={"agent_id": agent_id}) assert claimed.status_code == 200 assert claimed.json()["request_id"] == request_id assert claimed.json()["status"] == "RUNNING" completed = client.post( f"/agent/browse/{request_id}/result", json={ "status": "SUCCEEDED", "parent_path": r"D:\\", "entries": [{"name": "UPO", "path": r"D:\EDT\UPO", "is_directory": True}], }, ) assert completed.status_code == 200 assert completed.json()["status"] == "SUCCEEDED" assert completed.json()["entries"][0]["path"] == r"D:\EDT\UPO" def test_windows_agent_heartbeat_status(): client = TestClient(app) agent_id = f"win-agent-{uuid4()}" offline = client.get(f"/agent/status/{agent_id}") assert offline.status_code == 200 assert offline.json()["status"] == "offline" heartbeat = client.post( "/agent/heartbeat", json={ "agent_id": agent_id, "host": "workstation-1", "user": "svc-sfera", "version": "dev", "network_roots": [r"\\server\share"], }, ) assert heartbeat.status_code == 200 assert heartbeat.json()["status"] == "online" online = client.get(f"/agent/status/{agent_id}") assert online.status_code == 200 assert online.json()["status"] == "online" assert online.json()["host"] == "workstation-1" assert online.json()["network_roots"] == [r"\\server\share"] def test_server_browse_lists_directories(tmp_path: Path): (tmp_path / "edt").mkdir() (tmp_path / "file.txt").write_text("not a dir", encoding="utf-8") client = TestClient(app) response = client.get("/server/browse", params={"path": str(tmp_path)}) assert response.status_code == 200 payload = response.json() assert payload["path"] assert payload["parent_path"] assert payload["entries"] == [{"name": "edt", "path": (tmp_path / "edt").as_posix(), "is_directory": True}] def test_server_smb_browse_validates_unc_path(): client = TestClient(app) response = client.post("/server/smb/browse", json={"path": "192.168.200.200", "username": "user", "password": "secret"}) assert response.status_code == 200 assert "UNC путь" in response.json()["error"] def test_project_setup_recovers_indexed_status_from_stored_snapshot(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) project_id = f"setup-recover-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/XML_DUMP", json={"source": "XML_DUMP", "path": str(tmp_path)}, ) assert imported.status_code == 200 main._snapshots.pop(project_id, None) main._project_setup.pop(project_id, None) recovered = client.get(f"/projects/{project_id}/setup") assert recovered.status_code == 200 payload = recovered.json() assert payload["status"] == "INDEXED" assert payload["last_import"]["source"] == "XML_DUMP" def test_project_create_and_delete_lifecycle(): client = TestClient(app) project_id = f"lifecycle-{uuid4()}" created = client.post("/projects", json={"project_id": project_id, "name": "Lifecycle"}) assert created.status_code == 200 assert created.json()["project_id"] == project_id assert created.json()["settings"]["name"] == "Lifecycle" duplicate = client.post("/projects", json={"project_id": project_id}) assert duplicate.status_code == 409 wrong_confirmation = client.request("DELETE", f"/projects/{project_id}", json={"confirmation": "wrong"}) assert wrong_confirmation.status_code == 400 deleted = client.request("DELETE", f"/projects/{project_id}", json={"confirmation": project_id}) assert deleted.status_code == 200 assert deleted.json()["project_id"] == project_id def test_project_settings_persist_wizard_sections(): client = TestClient(app) project_id = f"setup-sections-{uuid4()}" settings = client.post( f"/projects/{project_id}/settings", json={ "name": "Wizard Sections", "configuration_source": "EDT local project", "structure_source": "EDT_PROJECT", "platform_version": "8.5.1.1150", "compatibility_mode": "8.3", "extensions": ["UPO_Ext"], "environments": { "dev": { "url": "http://192.168.200.95/upo/ru_RU/", "infobase": 'Srvr="192.168.200.95";Ref="upo";', "runtime": "mock", }, "test": {"runtime": "remote_worker"}, }, "agent": {"mode": "mock", "endpoint": "https://agent.local/snapshot", "allow_snapshots": True}, "privacy_mode": "LOCAL_ONLY", "knowledge_sources": ["wiki", "incidents"], "task_session_policy": { "mode": "required", "default_task_id": "task.default", "session_ttl": "8h", "audit_privacy": True, }, }, ) assert settings.status_code == 200 payload = settings.json()["settings"] assert payload["environments"]["dev"]["runtime"] == "mock" assert payload["agent"]["allow_snapshots"] is True assert payload["knowledge_sources"] == ["wiki", "incidents"] assert payload["task_session_policy"]["default_task_id"] == "task.default" def test_import_source_check_reports_preflight_requirements(tmp_path: Path): client = TestClient(app) project_id = f"source-check-{uuid4()}" xml_file = tmp_path / "metadata.xml" xml_file.write_text("", encoding="utf-8") xml_check = client.post( f"/projects/{project_id}/imports/XML_DUMP/check", json={"source": "XML_DUMP", "path": str(xml_file)}, ) assert xml_check.status_code == 200 assert xml_check.json()["ready"] is True assert any(check["code"] == "path" and check["status"] == "OK" for check in xml_check.json()["checks"]) live_check = client.post( f"/projects/{project_id}/imports/LIVE_INFOBASE/check", json={"source": "LIVE_INFOBASE"}, ) assert live_check.status_code == 200 assert live_check.json()["ready"] is False assert any(check["code"] == "credentials" and check["status"] == "BLOCKED" for check in live_check.json()["checks"]) def test_runtime_required_import_uses_api_mock_when_adapter_unavailable(monkeypatch): monkeypatch.setenv("RUNTIME_ADAPTER_MODE", "mock") monkeypatch.setenv("RUNTIME_ADAPTER_URL", "http://127.0.0.1:1") client = TestClient(app) project_id = f"runtime-import-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/CF_FILE", json={"source": "CF_FILE", "metadata": {"platform_version": "8.3.24"}}, ) assert imported.status_code == 200 payload = imported.json() assert payload["status"] == "mock_indexed" assert payload["runtime_mode"] == "mock" assert payload["runtime_diagnostics"] assert payload["snapshot"]["project_id"] == project_id platform = client.get("/runtime/platform") assert platform.status_code == 200 assert platform.json()["mode"] == "mock" assert platform.json()["platform_found"] is False def test_runtime_required_import_does_not_index_cf_file_directly(monkeypatch, tmp_path: Path): monkeypatch.setenv("RUNTIME_ADAPTER_MODE", "mock") monkeypatch.setenv("RUNTIME_ADAPTER_URL", "http://127.0.0.1:1") cf_file = tmp_path / "demo.cf" cf_file.write_text("not xml", encoding="utf-8") client = TestClient(app) project_id = f"cf-import-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/CF_FILE", json={"source": "CF_FILE", "path": str(cf_file), "metadata": {"platform_version": "8.3.24"}}, ) assert imported.status_code == 200 payload = imported.json() assert payload["status"] == "mock_indexed" assert payload["source_path"] != cf_file.as_posix() assert payload["normalized_summary"]["object_count"] >= 1 def test_import_supports_structure_only_indexing(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) project_id = f"structure-only-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(tmp_path), "structure_only": True}, ) assert imported.status_code == 200 payload = imported.json() assert payload["status"] == "structure_indexed" assert payload["snapshot"]["project_id"] == project_id assert payload["object_count"] >= 2 assert payload["normalized_summary"]["group_count"] >= 3 assert payload["normalized_summary"]["rights_count"] == 1 setup = client.get(f"/projects/{project_id}/setup") assert setup.status_code == 200 assert setup.json()["status"] == "STRUCTURE_INDEXED" assert "Запустите индексацию" in setup.json()["message"] summary = client.get(f"/projects/{project_id}/normalized/summary") assert summary.status_code == 200 assert summary.json()["rights_count"] == 1 quality = client.get(f"/projects/{project_id}/imports/quality") assert quality.status_code == 200 quality_payload = quality.json() assert quality_payload["project_id"] == project_id assert quality_payload["score"] > 0 assert any(check["code"] == "rights" and check["passed"] for check in quality_payload["checks"]) normalized = client.get(f"/projects/{project_id}/normalized") assert normalized.status_code == 200 groups = {group["name"] for group in normalized.json()["configuration"]["groups"]} assert {"HTTP-сервисы", "Подсистемы", "Роли"}.issubset(groups) detail = client.get( f"/projects/{project_id}/normalized/object", params={"qualified_name": "Роль.Менеджер"}, ) assert detail.status_code == 200 assert detail.json()["group_name"] == "Роли" assert detail.json()["object"]["rights"][0]["target"] == "HTTPСервис.ПубличныйAPI" tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 root = tree.json()["root"] assert root["label"] == "Проект" assert [item["label"] for item in root["children"][:4]] == [ "Основная конфигурация", "Расширение: <Имя>", "SFERA", "Среды", ] main_configuration = root["children"][0] common = next(item for item in main_configuration["children"] if item["label"] == "Общие") assert any(item["label"] == "HTTP-сервисы" for item in common["children"]) def test_import_full_replace_replaces_current_normalized_project(tmp_path: Path): first = tmp_path / "first" second = tmp_path / "second" first.mkdir() second.mkdir() (first / "Контрагенты.mdo").write_text( """ Контрагенты Контрагенты """, encoding="utf-8", ) (second / "Номенклатура.mdo").write_text( """ Номенклатура Номенклатура """, encoding="utf-8", ) client = TestClient(app) project_id = f"full-replace-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(first), "mode": "FULL_REPLACE"}, ) assert imported.status_code == 200 assert imported.json()["mode"] == "FULL_REPLACE" assert imported.json()["applied"] is True replaced = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(second), "mode": "FULL_REPLACE"}, ) assert replaced.status_code == 200 assert replaced.json()["mode"] == "FULL_REPLACE" assert replaced.json()["normalized_summary"]["object_count"] == 1 normalized = client.get(f"/projects/{project_id}/normalized") assert normalized.status_code == 200 payload = normalized.json() objects = [ item["qualified_name"] for group in payload["configuration"]["groups"] for item in group["objects"] ] assert objects == ["Справочник.Номенклатура"] def test_import_sync_preview_does_not_replace_current_project(tmp_path: Path): current = tmp_path / "current" incoming = tmp_path / "incoming" current.mkdir() incoming.mkdir() (current / "Контрагенты.mdo").write_text( """ Контрагенты """, encoding="utf-8", ) (incoming / "Номенклатура.mdo").write_text( """ Номенклатура """, encoding="utf-8", ) client = TestClient(app) project_id = f"sync-preview-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(current)}, ) assert imported.status_code == 200 preview = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(incoming), "mode": "SYNC_PREVIEW"}, ) assert preview.status_code == 200 payload = preview.json() assert payload["status"] == "sync_preview" assert payload["mode"] == "SYNC_PREVIEW" assert payload["applied"] is False assert payload["sync_preview"]["added_count"] == 1 assert payload["sync_preview"]["removed_count"] == 1 assert {item["change_kind"] for item in payload["sync_preview"]["items"]} == {"ADD", "REMOVE"} normalized = client.get(f"/projects/{project_id}/normalized") assert normalized.status_code == 200 objects = [ item["qualified_name"] for group in normalized.json()["configuration"]["groups"] for item in group["objects"] ] assert objects == ["Справочник.Контрагенты"] def test_import_full_replace_keeps_valid_edt_objects_when_one_xml_is_broken(tmp_path: Path): (tmp_path / "valid.mdo").write_text( """ Контрагенты Контрагенты """, encoding="utf-8", ) (tmp_path / "broken.mdo").write_text(" Контрагенты """, encoding="utf-8", ) (module_dir / "ObjectModule.bsl").write_text( """ Процедура ПроверитьКонтрагента() Экспорт КонецПроцедуры """, encoding="utf-8", ) (module_dir / "ManagerModule.bsl").write_text( """ Процедура Создать() Экспорт КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) project_id = f"edt-modules-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(tmp_path), "mode": "FULL_REPLACE"}, ) assert imported.status_code == 200 assert imported.json()["normalized_summary"]["module_count"] == 2 detail = client.get( f"/projects/{project_id}/normalized/object", params={"qualified_name": "Справочник.Контрагенты"}, ) assert detail.status_code == 200 modules = detail.json()["object"]["modules"] assert [module["module_kind"] for module in modules] == ["MANAGER_MODULE", "OBJECT_MODULE"] assert all(module["attributes"]["source_hash"] for module in modules) def test_import_edt_full_replace_keeps_tabular_section_columns_nested(tmp_path: Path): (tmp_path / "ЗаказПокупателя.mdo").write_text( """ ЗаказПокупателя Товары Номенклатура Количество """, encoding="utf-8", ) client = TestClient(app) project_id = f"edt-tabular-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(tmp_path), "mode": "FULL_REPLACE"}, ) assert imported.status_code == 200 assert imported.json()["normalized_summary"]["attribute_count"] == 2 detail = client.get( f"/projects/{project_id}/normalized/object", params={"qualified_name": "Документ.ЗаказПокупателя"}, ) assert detail.status_code == 200 document = detail.json()["object"] assert document["attributes"] == [] assert [child["name"] for child in document["tabular_sections"][0]["children"]] == ["Номенклатура", "Количество"] def test_import_full_replace_preserves_configuration_root_metadata(tmp_path: Path): (tmp_path / "configuration.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) project_id = f"configuration-root-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/XML_DUMP", json={"source": "XML_DUMP", "path": str(tmp_path), "mode": "FULL_REPLACE"}, ) assert imported.status_code == 200 assert imported.json()["normalized_summary"]["object_count"] == 1 normalized = client.get(f"/projects/{project_id}/normalized") assert normalized.status_code == 200 configuration = normalized.json()["configuration"] assert configuration["name"] == "УправлениеТорговлей" assert configuration["metadata"]["synonym"] == "Управление торговлей" assert configuration["metadata"]["platformVersion"] == "8.3.24" def test_project_metadata_tree_uses_catalog_structure(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ ОстаткиТоваров РегистрНакопления.ОстаткиТоваров """, encoding="utf-8", ) client = TestClient(app) project_id = f"metadata-tree-{uuid4()}" indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 response = client.get(f"/projects/{project_id}/metadata/tree") assert response.status_code == 200 root = response.json()["root"] main_configuration = next(item for item in root["children"] if item["label"] == "Основная конфигурация") assert any(item["label"].startswith("Расширение:") for item in root["children"]) assert any(item["label"] == "SFERA" for item in root["children"]) assert any(item["label"] == "Среды" for item in root["children"]) common = next(item for item in main_configuration["children"] if item["label"] == "Общие") catalogs = next(item for item in main_configuration["children"] if item["label"] == "Справочники") documents = next(item for item in main_configuration["children"] if item["label"] == "Документы") accumulation_registers = next(item for item in main_configuration["children"] if item["label"] == "Регистры накопления") assert any(item["label"] == "Общие модули" for item in common["children"]) assert any(item["label"] == "Общие формы" for item in common["children"]) assert not any(item["label"] == "Общие формы" for item in main_configuration["children"]) assert catalogs["children"][0]["qualified_name"] == "Справочник.Контрагенты" assert any(item["label"] == "Реквизиты" for item in catalogs["children"][0]["children"]) assert documents["children"][0]["qualified_name"] == "Документ.ЗаказПокупателя" assert accumulation_registers["children"][0]["qualified_name"] == "РегистрНакопления.ОстаткиТоваров" lazy = client.get(f"/projects/{project_id}/metadata/tree", params={"object_limit_per_branch": 0}) assert lazy.status_code == 200 lazy_main_configuration = next(item for item in lazy.json()["root"]["children"] if item["label"] == "Основная конфигурация") lazy_catalogs = next(item for item in lazy_main_configuration["children"] if item["label"] == "Справочники") assert lazy_catalogs["children"] == [] assert lazy_catalogs["has_more"] is True assert lazy_catalogs["count"] == 1 children = client.get( f"/projects/{project_id}/metadata/tree/children", params={"node_id": "branch.CATALOG", "offset": 0, "limit": 1}, ) assert children.status_code == 200 assert children.json()["total"] == 1 assert children.json()["children"][0]["qualified_name"] == "Справочник.Контрагенты" search = client.get(f"/projects/{project_id}/metadata/tree/search", params={"q": "Контраг", "limit": 5}) assert search.status_code == 200 assert search.json()["total"] >= 1 assert search.json()["results"][0]["qualified_name"] == "Справочник.Контрагенты" path = client.get( f"/projects/{project_id}/metadata/tree/path", params={"node_id": search.json()["results"][0]["id"]}, ) assert path.status_code == 200 assert path.json()["path"] == ["main-configuration", "branch.CATALOG", search.json()["results"][0]["id"]] assert path.json()["steps"][-1] == { "parent_id": "branch.CATALOG", "child_id": search.json()["results"][0]["id"], "offset": 0, } def test_project_metadata_tree_loads_normalized_lazy_children(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """
""", encoding="utf-8", ) client = TestClient(app) project_id = f"normalized-tree-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/XML_DUMP", json={"source": "XML_DUMP", "path": str(tmp_path), "mode": "FULL_REPLACE"}, ) assert imported.status_code == 200 lazy = client.get(f"/projects/{project_id}/metadata/tree", params={"object_limit_per_branch": 0}) assert lazy.status_code == 200 main_configuration = next(item for item in lazy.json()["root"]["children"] if item["label"] == "Основная конфигурация") catalogs = next(item for item in main_configuration["children"] if item["label"] == "Справочники") assert catalogs["id"].startswith("normalized.branch.") assert catalogs["children"] == [] assert catalogs["has_more"] is True objects = client.get( f"/projects/{project_id}/metadata/tree/children", params={"node_id": catalogs["id"], "offset": 0, "limit": 80}, ) assert objects.status_code == 200 catalog = objects.json()["children"][0] assert catalog["qualified_name"] == "Справочник.Контрагенты" attributes = next(item for item in catalog["children"] if item["label"] == "Реквизиты") parts = client.get( f"/projects/{project_id}/metadata/tree/children", params={"node_id": attributes["id"], "offset": 0, "limit": 80}, ) assert parts.status_code == 200 assert parts.json()["children"][0]["qualified_name"] == "Справочник.Контрагенты.ИНН" def test_project_metadata_tree_shows_http_service_documented_structure(tmp_path: Path): (tmp_path / "ПубличныйAPI.mdo").write_text( """ ПубличныйAPI api Orders GET ПолучитьЗаказ """, encoding="utf-8", ) client = TestClient(app) project_id = f"http-service-tree-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(tmp_path), "mode": "FULL_REPLACE"}, ) assert imported.status_code == 200 tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 main_configuration = next(item for item in tree.json()["root"]["children"] if item["label"] == "Основная конфигурация") common = next(item for item in main_configuration["children"] if item["label"] == "Общие") http_services = next(item for item in common["children"] if item["label"] == "HTTP-сервисы") service = http_services["children"][0] assert service["qualified_name"] == "HTTPСервис.ПубличныйAPI" assert [item["label"] for item in service["children"]] == ["Шаблоны URL"] templates = client.get( f"/projects/{project_id}/metadata/tree/children", params={"node_id": service["children"][0]["id"], "offset": 0, "limit": 80}, ) assert templates.status_code == 200 assert templates.json()["children"][0]["label"] == "Orders" def test_project_metadata_tree_shows_report_documented_structure(tmp_path: Path): (tmp_path / "АнализПродаж.mdo").write_text( """ АнализПродаж Период Показатели ФормаОтчета ПечатнаяФорма ТабличныйДокумент ОсновнаяСхемаКомпоновкиДанных Основной НастройкиПоУмолчанию """, encoding="utf-8", ) client = TestClient(app) project_id = f"report-tree-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(tmp_path), "mode": "FULL_REPLACE"}, ) assert imported.status_code == 200 tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 main_configuration = next(item for item in tree.json()["root"]["children"] if item["label"] == "Основная конфигурация") reports = next(item for item in main_configuration["children"] if item["label"] == "Отчеты") report = reports["children"][0] assert report["qualified_name"] == "Отчет.АнализПродаж" assert [item["label"] for item in report["children"]] == [ "Реквизиты", "Табличные части", "Формы", "Макеты", "Табличные документы", "СКД", "Варианты отчета", "Настройки", ] dcs = next(item for item in report["children"] if item["label"] == "СКД") dcs_children = client.get( f"/projects/{project_id}/metadata/tree/children", params={"node_id": dcs["id"], "offset": 0, "limit": 80}, ) assert dcs_children.status_code == 200 assert dcs_children.json()["children"][0]["label"] == "ОсновнаяСхемаКомпоновкиДанных" def test_project_metadata_tree_shows_extension_as_configuration_structure(tmp_path: Path): (tmp_path / "РасширениеCRM.mdo").write_text( """ CRM 1.0 КонтрагентыCRM ВнешнийКод CRMСервер """, encoding="utf-8", ) client = TestClient(app) project_id = f"extension-tree-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/EDT_PROJECT", json={"source": "EDT_PROJECT", "path": str(tmp_path), "mode": "FULL_REPLACE"}, ) assert imported.status_code == 200 assert imported.json()["normalized_summary"]["extension_count"] == 1 tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 root_children = tree.json()["root"]["children"] extension = next(item for item in root_children if item["label"] == "Расширение: CRM") assert extension["kind"] == "EXTENSION" assert [item["label"] for item in extension["children"][:2]] == ["Сведения (1.0)", "Общие"] common = next(item for item in extension["children"] if item["label"] == "Общие") common_modules = next(item for item in common["children"] if item["label"] == "Общие модули") assert common_modules["children"][0]["label"] == "CRMСервер" catalogs = next(item for item in extension["children"] if item["label"] == "Справочники") catalog = catalogs["children"][0] assert catalog["qualified_name"].endswith("Справочник.КонтрагентыCRM") attributes = next(item for item in catalog["children"] if item["label"] == "Реквизиты") parts = client.get( f"/projects/{project_id}/metadata/tree/children", params={"node_id": attributes["id"], "offset": 0, "limit": 80}, ) assert parts.status_code == 200 assert parts.json()["children"][0]["label"] == "ВнешнийКод" lazy_tree = client.get(f"/projects/{project_id}/metadata/tree", params={"object_limit_per_branch": 0}) lazy_extension = next(item for item in lazy_tree.json()["root"]["children"] if item["label"] == "Расширение: CRM") lazy_catalogs = next(item for item in lazy_extension["children"] if item["label"] == "Справочники") lazy_catalog_children = client.get( f"/projects/{project_id}/metadata/tree/children", params={"node_id": lazy_catalogs["id"], "offset": 0, "limit": 80}, ) assert lazy_catalog_children.status_code == 200 assert lazy_catalog_children.json()["children"][0]["label"] == "КонтрагентыCRM" def test_project_metadata_tree_adds_reference_configuration_root(): client = TestClient(app) project_id = f"reference-tree-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/REFERENCE_CONFIGURATION", json={ "source": "REFERENCE_CONFIGURATION", "metadata": {"context_description": "Reference baseline for comparison"}, }, ) assert imported.status_code == 200 tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 root = tree.json()["root"] reference = next(item for item in root["children"] if item["kind"] == "REFERENCE_CONFIGURATION") assert reference["label"] == "Reference configuration" assert [item["label"] for item in reference["children"][:3]] == ["Сведения", "Общие", "Константы"] lazy_tree = client.get(f"/projects/{project_id}/metadata/tree", params={"object_limit_per_branch": 0}) assert lazy_tree.status_code == 200 lazy_root = lazy_tree.json()["root"] lazy_reference = next(item for item in lazy_root["children"] if item["kind"] == "REFERENCE_CONFIGURATION") assert lazy_reference["label"] == "Reference configuration" def test_project_metadata_tree_adds_context_configuration_root(): client = TestClient(app) project_id = f"context-tree-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/CONTEXT_ONLY", json={ "source": "CONTEXT_ONLY", "metadata": {"context_description": "Context-only outline"}, }, ) assert imported.status_code == 200 tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 root = tree.json()["root"] context_root = next(item for item in root["children"] if item["kind"] == "CONTEXT_CONFIGURATION") assert context_root["label"] == "Context-only configuration" assert [item["label"] for item in context_root["children"][:3]] == ["Сведения", "Общие", "Константы"] lazy_tree = client.get(f"/projects/{project_id}/metadata/tree", params={"object_limit_per_branch": 0}) assert lazy_tree.status_code == 200 lazy_root = lazy_tree.json()["root"] lazy_context_root = next(item for item in lazy_root["children"] if item["kind"] == "CONTEXT_CONFIGURATION") assert lazy_context_root["label"] == "Context-only configuration" def test_project_metadata_tree_does_not_add_context_or_reference_roots_for_regular_sources(): client = TestClient(app) project_id = f"regular-tree-{uuid4()}" imported = client.post( f"/projects/{project_id}/imports/XML_DUMP", json={ "source": "XML_DUMP", "metadata": {"platform_version": "8.3.24", "compatibility_mode": "8.3.20"}, }, ) assert imported.status_code == 200 tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 root = tree.json()["root"] kinds = {item["kind"] for item in root["children"]} assert "CONTEXT_CONFIGURATION" not in kinds assert "REFERENCE_CONFIGURATION" not in kinds lazy_tree = client.get(f"/projects/{project_id}/metadata/tree", params={"object_limit_per_branch": 0}) assert lazy_tree.status_code == 200 lazy_root = lazy_tree.json()["root"] lazy_kinds = {item["kind"] for item in lazy_root["children"]} assert "CONTEXT_CONFIGURATION" not in lazy_kinds assert "REFERENCE_CONFIGURATION" not in lazy_kinds def test_project_metadata_tree_places_conditional_configuration_roots_before_sfera(): client = TestClient(app) context_project = f"context-order-{uuid4()}" context_import = client.post( f"/projects/{context_project}/imports/CONTEXT_ONLY", json={"source": "CONTEXT_ONLY", "metadata": {"context_description": "Context order check"}}, ) assert context_import.status_code == 200 context_tree = client.get(f"/projects/{context_project}/metadata/tree") assert context_tree.status_code == 200 context_children = context_tree.json()["root"]["children"] context_index = next(i for i, item in enumerate(context_children) if item["kind"] == "CONTEXT_CONFIGURATION") context_sfera_index = next(i for i, item in enumerate(context_children) if item["label"] == "SFERA") assert context_index < context_sfera_index context_lazy_tree = client.get(f"/projects/{context_project}/metadata/tree", params={"object_limit_per_branch": 0}) assert context_lazy_tree.status_code == 200 context_lazy_children = context_lazy_tree.json()["root"]["children"] context_lazy_index = next(i for i, item in enumerate(context_lazy_children) if item["kind"] == "CONTEXT_CONFIGURATION") context_lazy_sfera_index = next(i for i, item in enumerate(context_lazy_children) if item["label"] == "SFERA") assert context_lazy_index < context_lazy_sfera_index reference_project = f"reference-order-{uuid4()}" reference_import = client.post( f"/projects/{reference_project}/imports/REFERENCE_CONFIGURATION", json={"source": "REFERENCE_CONFIGURATION", "metadata": {"context_description": "Reference order check"}}, ) assert reference_import.status_code == 200 reference_tree = client.get(f"/projects/{reference_project}/metadata/tree") assert reference_tree.status_code == 200 reference_children = reference_tree.json()["root"]["children"] reference_index = next(i for i, item in enumerate(reference_children) if item["kind"] == "REFERENCE_CONFIGURATION") reference_sfera_index = next(i for i, item in enumerate(reference_children) if item["label"] == "SFERA") assert reference_index < reference_sfera_index reference_lazy_tree = client.get(f"/projects/{reference_project}/metadata/tree", params={"object_limit_per_branch": 0}) assert reference_lazy_tree.status_code == 200 reference_lazy_children = reference_lazy_tree.json()["root"]["children"] reference_lazy_index = next(i for i, item in enumerate(reference_lazy_children) if item["kind"] == "REFERENCE_CONFIGURATION") reference_lazy_sfera_index = next(i for i, item in enumerate(reference_lazy_children) if item["label"] == "SFERA") assert reference_lazy_index < reference_lazy_sfera_index def test_normalized_object_modules_returns_linked_bsl_source(tmp_path: Path): catalog_dir = tmp_path / "Catalogs" / "Контрагенты" catalog_dir.mkdir(parents=True) (catalog_dir / "Контрагенты.mdo").write_text( """ Контрагенты """, encoding="utf-8", ) (catalog_dir / "ObjectModule.bsl").write_text( """ Процедура ПередЗаписью(Отказ) ПроверитьКонтрагента(); Отказ = Ложь; КонецПроцедуры Процедура ПроверитьКонтрагента() Запрос = Новый Запрос("ВЫБРАТЬ Контрагенты.Ссылка ИЗ Справочник.Контрагенты КАК Контрагенты"); КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) project_id = f"object-modules-{uuid4()}" indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 modules = client.get( f"/projects/{project_id}/normalized/object/modules", params={"qualified_name": "Справочник.Контрагенты"}, ) assert modules.status_code == 200 payload = modules.json() assert payload[0]["module_role"] == "OBJECT_MODULE" assert "ПередЗаписью" in payload[0]["source_text"] assert payload[0]["routines_count"] == 2 assert payload[0]["routines"][0]["name"] == "ПередЗаписью" assert payload[0]["routines"][0]["kind"] == "PROCEDURE" assert payload[0]["routines"][0]["calls"] == ["ObjectModule.ПроверитьКонтрагента"] assert payload[0]["routines"][1]["queries_count"] == 1 assert "Справочник.Контрагенты" in payload[0]["routines"][1]["queries"][0] assert payload[0]["routines"][1]["impact_level"] == "MEDIUM" assert "reads query tables" in payload[0]["routines"][1]["impact_reasons"] tree = client.get(f"/projects/{project_id}/metadata/tree") assert tree.status_code == 200 main_configuration = next(item for item in tree.json()["root"]["children"] if item["label"] == "Основная конфигурация") catalogs = next(item for item in main_configuration["children"] if item["label"] == "Справочники") catalog = catalogs["children"][0] object_module_group = next(item for item in catalog["children"] if item["label"] == "Модуль объекта") module_children = client.get( f"/projects/{project_id}/metadata/tree/children", params={"node_id": object_module_group["id"], "offset": 0, "limit": 80}, ) assert module_children.status_code == 200 assert module_children.json()["children"][0]["kind"] == "MODULE" module_qname = module_children.json()["children"][0]["qualified_name"] selected_module = client.get( f"/projects/{project_id}/normalized/object/modules", params={"qualified_name": module_qname}, ) assert selected_module.status_code == 200 assert selected_module.json()[0]["source_text"] == payload[0]["source_text"] selected_normalized_module = client.get( f"/projects/{project_id}/normalized/object/modules", params={"qualified_name": "Справочник.Контрагенты.МодульОбъекта"}, ) assert selected_normalized_module.status_code == 200 assert selected_normalized_module.json()[0]["source_text"] == payload[0]["source_text"] selected_routine = client.get( f"/projects/{project_id}/normalized/object/modules", params={"qualified_name": "ПередЗаписью"}, ) assert selected_routine.status_code == 200 assert selected_routine.json()[0]["source_text"] == payload[0]["source_text"] def test_normalized_object_modules_resolves_common_module_code(tmp_path: Path): common_module_dir = tmp_path / "CommonModules" / "CRMСервер" common_module_dir.mkdir(parents=True) (common_module_dir / "CRMСервер.mdo").write_text( """ CRMСервер """, encoding="utf-8", ) (common_module_dir / "Module.bsl").write_text( """ Процедура ЗагрузитьКонтрагентов() Экспорт Сообщить("Готово"); КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) project_id = f"common-module-code-{uuid4()}" indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 common_module = client.get( f"/projects/{project_id}/normalized/object/modules", params={"qualified_name": "ОбщийМодуль.CRMСервер"}, ) assert common_module.status_code == 200 payload = common_module.json() assert payload[0]["module_role"] == "MODULE" assert "ЗагрузитьКонтрагентов" in payload[0]["source_text"] normalized_module_node = client.get( f"/projects/{project_id}/normalized/object/modules", params={"qualified_name": "ОбщийМодуль.CRMСервер.Модуль"}, ) assert normalized_module_node.status_code == 200 assert normalized_module_node.json()[0]["source_text"] == payload[0]["source_text"] def test_normalized_object_modules_reads_canonical_project_before_snapshot(): client = TestClient(app) project_id = f"normalized-source-{uuid4()}" source_text = "Процедура Выполнить() Экспорт\n Сообщить(\"ok\");\nКонецПроцедуры\n" normalized = NormalizedProject( project_id=project_id, configuration=ConfigurationRoot( groups=[ MetadataGroup( name="Общие модули", object_kinds=["COMMON_MODULE"], objects=[ MetadataObject( name="CRMСервер", qualified_name="ОбщийМодуль.CRMСервер", object_kind="COMMON_MODULE", modules=[ Module( name="Module", qualified_name="ОбщийМодуль.CRMСервер.Модуль", source_path="/edt/CommonModules/CRMСервер/Module.bsl", module_kind="MODULE", attributes={"source_text": source_text, "module_role": "MODULE"}, ) ], ) ], ) ] ), source_path="/edt", ) main._save_normalized_project(project_id, normalized) by_object = client.get( f"/projects/{project_id}/normalized/object/modules", params={"qualified_name": "ОбщийМодуль.CRMСервер"}, ) assert by_object.status_code == 200 assert by_object.json()[0]["source_text"] == source_text assert by_object.json()[0]["routines"][0]["name"] == "Выполнить" assert by_object.json()[0]["routines"][0]["export"] is True by_module = client.get( f"/projects/{project_id}/normalized/object/modules", params={"qualified_name": "ОбщийМодуль.CRMСервер.Модуль"}, ) assert by_module.status_code == 200 assert by_module.json()[0]["source_text"] == source_text def test_bsl_completions_returns_only_exported_common_module_routines(): client = TestClient(app) project_id = f"bsl-completion-{uuid4()}" source_text = "\n".join([ "Процедура Выполнить() Экспорт", "КонецПроцедуры", "", "Функция ВнутреннийРасчет()", " Возврат 1;", "КонецФункции", "", ]) normalized = NormalizedProject( project_id=project_id, configuration=ConfigurationRoot( groups=[ MetadataGroup( name="Общие модули", object_kinds=["COMMON_MODULE"], objects=[ MetadataObject( name="CRMСервер", qualified_name="ОбщийМодуль.CRMСервер", object_kind="COMMON_MODULE", modules=[ Module( name="Module", qualified_name="ОбщийМодуль.CRMСервер.Модуль", source_path="/edt/CommonModules/CRMСервер/Module.bsl", module_kind="MODULE", attributes={"source_text": source_text}, ) ], ) ], ) ] ), ) main._save_normalized_project(project_id, normalized) response = client.get( f"/projects/{project_id}/bsl/completions", params={"receiver": "CRMСервер"}, ) assert response.status_code == 200 labels = {item["label"] for item in response.json()} assert "Выполнить" in labels assert "ВнутреннийРасчет" not in labels def test_project_flowchart_returns_overview_and_focus(tmp_path: Path): catalog_dir = tmp_path / "Catalogs" / "Clients" catalog_dir.mkdir(parents=True) (catalog_dir / "Clients.mdo").write_text( """ Clients """, encoding="utf-8", ) (catalog_dir / "ObjectModule.bsl").write_text( "Процедура ПередЗаписью(Отказ)\n Отказ = Ложь;\nКонецПроцедуры\n", encoding="utf-8", ) client = TestClient(app) project_id = f"flowchart-{uuid4()}" indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 overview = client.get(f"/projects/{project_id}/flowchart") assert overview.status_code == 200 overview_payload = overview.json() assert overview_payload["mode"] == "overview" assert any(node["kind"] == "CATALOG" and node["count"] == 1 for node in overview_payload["nodes"]) assert all(node["kind"] not in {"FORM", "ATTRIBUTE", "TABULAR_SECTION"} for node in overview_payload["nodes"]) focus = client.get( f"/projects/{project_id}/flowchart", params={"focus": "Справочник.Clients", "depth": 2, "limit": 50}, ) assert focus.status_code == 200 focus_payload = focus.json() assert focus_payload["mode"] == "focus" assert any(node["qualified_name"] == "Справочник.Clients" for node in focus_payload["nodes"]) assert all(node["kind"] not in {"FORM", "ATTRIBUTE", "TABULAR_SECTION"} for node in focus_payload["nodes"]) assert focus_payload["edges"] == [] def test_project_flowchart_collapses_module_logic_to_object_links(tmp_path: Path): catalog_dir = tmp_path / "Catalogs" / "Clients" document_dir = tmp_path / "Documents" / "Order" catalog_dir.mkdir(parents=True) document_dir.mkdir(parents=True) (catalog_dir / "Clients.mdo").write_text( """ Clients """, encoding="utf-8", ) (document_dir / "Order.mdo").write_text( """ Order """, encoding="utf-8", ) (document_dir / "ObjectModule.bsl").write_text( """ Процедура Проведение(Отказ) Запрос = Новый Запрос("ВЫБРАТЬ Clients.Ссылка ИЗ Справочник.Clients КАК Clients"); КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) project_id = f"flowchart-logic-{uuid4()}" indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 focus = client.get( f"/projects/{project_id}/flowchart", params={"focus": "Документ.Order", "depth": 2, "limit": 50}, ) assert focus.status_code == 200 payload = focus.json() assert any(node["qualified_name"] == "Документ.Order" for node in payload["nodes"]) assert any(node["qualified_name"] == "Справочник.Clients" for node in payload["nodes"]) assert any(edge["kind"] == "READS_TABLE" for edge in payload["edges"]) def test_index_project_and_query_impact(tmp_path: Path): module = tmp_path / "demo_module.bsl" module.write_text( """ Процедура Проведение() ПроверитьОстатки(); Движения.ОстаткиТоваров.Записать(); КонецПроцедуры Процедура ПроверитьОстатки() КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api"}, ) assert indexed.status_code == 200 assert indexed.json()["snapshot"]["project_id"] == "demo-api" assert client.get("/storage").json()["status"] == "configured" assert any(item["project_id"] == "demo-api" for item in client.get("/storage/snapshots").json()) loaded = client.post("/projects/demo-api/load") assert loaded.status_code == 200 assert loaded.json()["snapshot"]["project_id"] == "demo-api" snapshot = client.get("/projects/demo-api/snapshot") assert snapshot.status_code == 200 assert snapshot.json()["node_count"] >= 3 exported = client.get("/projects/demo-api/snapshot/export") assert exported.status_code == 200 assert exported.json()["project_id"] == "demo-api" impact = client.get("/projects/demo-api/impact/Проведение") assert impact.status_code == 200 payload = impact.json() assert [node["name"] for node in payload["callees"]] == ["ПроверитьОстатки"] assert [node["name"] for node in payload["writes"]] == ["ОстаткиТоваров"] search = client.get("/projects/demo-api/search", params={"q": "Пров", "kind": "PROCEDURE"}) assert search.status_code == 200 assert search.json()["results"][0]["name"] == "Проведение" routine_lineage = search.json()["results"][0]["lineage_id"] versions = client.get("/projects/demo-api/versions") assert versions.status_code == 200 assert any(item["lineage_id"] == routine_lineage for item in versions.json()) lineage_history = client.get(f"/versions/{routine_lineage}") assert lineage_history.status_code == 200 assert lineage_history.json()[0]["lineage_id"] == routine_lineage usage = client.get("/projects/demo-api/tables/usage", params={"table": "ОстаткиТоваров"}) assert usage.status_code == 200 assert usage.json()[0]["writers"][0]["name"] == "Проведение" writes = client.get( "/projects/demo-api/transactions/writes", params={"target": "ОстаткиТоваров"}, ) assert writes.status_code == 200 assert writes.json()[0]["routine"]["name"] == "Проведение" module.write_text( """ Процедура Проведение() ПроверитьОстатки(); КонецПроцедуры Процедура ПроверитьОстатки() КонецПроцедуры """, encoding="utf-8", ) incremental = client.post( "/projects/demo-api/incremental/file", json={"path": str(module)}, ) assert incremental.status_code == 200 assert incremental.json()["removed_edges"] >= 1 assert client.get( "/projects/demo-api/transactions/writes", params={"target": "ОстаткиТоваров"}, ).json() == [] signal = client.post( "/projects/demo-api/runtime/signals", json={ "signal": { "signal_id": "signal.1", "lineage_id": routine_lineage, "kind": "ERROR", "duration_ms": 50.0, } }, ) assert signal.status_code == 200 runtime = client.get("/projects/demo-api/runtime/summary") assert runtime.status_code == 200 assert runtime.json()[0]["node"]["name"] == "Проведение" assert runtime.json()[0]["error_count"] == 1 knowledge = client.post( "/knowledge", json={ "record_id": "knowledge.demo", "scope": "PROJECT", "title": "Проведение документов", "body": "Правила проведения заказа.", "related_lineages": [routine_lineage], }, ) assert knowledge.status_code == 200 search_knowledge = client.get("/knowledge/search", params={"q": "заказа"}) assert search_knowledge.status_code == 200 assert any(item["record_id"] == "knowledge.demo" for item in search_knowledge.json()["results"]) coverage = client.get("/projects/demo-api/knowledge/coverage") assert coverage.status_code == 200 assert any(item["record_count"] == 1 for item in coverage.json()) def test_project_symbol_navigation_endpoints(tmp_path: Path): module = tmp_path / "demo_module.bsl" module.write_text( """ Процедура Проведение() ПроверитьОстатки(); КонецПроцедуры Процедура ПроверитьОстатки() КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) project_id = f"symbols-api-{uuid4()}" indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 symbols = client.get(f"/projects/{project_id}/symbols", params={"q": "Проверить", "kind": "PROCEDURE"}) assert symbols.status_code == 200 symbol_payload = symbols.json() assert symbol_payload[0]["node"]["name"] == "ПроверитьОстатки" assert symbol_payload[0]["source"]["source_path"].endswith("demo_module.bsl") lineage_id = symbol_payload[0]["node"]["lineage_id"] definition = client.get(f"/projects/{project_id}/symbols/definition", params={"lineage_id": lineage_id}) assert definition.status_code == 200 assert definition.json()["node"]["qualified_name"] == "demo_module.ПроверитьОстатки" assert definition.json()["source"]["line_start"] is not None references = client.get(f"/projects/{project_id}/symbols/references", params={"lineage_id": lineage_id}) assert references.status_code == 200 reference_payload = references.json() assert reference_payload["symbol"]["node"]["name"] == "ПроверитьОстатки" assert any( reference["kind"] == "CALLS" and reference["source"]["name"] == "Проведение" and reference["direction"] == "incoming" for reference in reference_payload["references"] ) def test_persisted_project_endpoints_load_snapshot_after_memory_clear(tmp_path: Path): project_id = f"persisted-api-{uuid4()}" module = tmp_path / "persisted_module.bsl" module.write_text( """ Процедура Проведение() ПроверитьОстатки(); Движения.ОстаткиТоваров.Записать(); КонецПроцедуры Процедура ПроверитьОстатки() КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 main._snapshots.pop(project_id, None) main._graphs.pop(project_id, None) assert client.get(f"/projects/{project_id}/snapshot").status_code == 200 search = client.get(f"/projects/{project_id}/search", params={"q": "Пров", "kind": "PROCEDURE"}) assert search.status_code == 200 assert search.json()["results"][0]["name"] == "Проведение" assert client.get(f"/projects/{project_id}/review").status_code == 200 report = client.get(f"/projects/{project_id}/report") assert report.status_code == 200 assert report.json()["project_id"] == project_id def test_xml_ui_forms_endpoint(tmp_path: Path): (tmp_path / "form.xml").write_text( """ """, encoding="utf-8", ) (tmp_path / "form_module.bsl").write_text( "Процедура ПровестиКоманда()\nКонецПроцедуры\n", encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-ui"}, ) assert indexed.status_code == 200 forms = client.get("/projects/demo-api-ui/ui/forms") assert forms.status_code == 200 assert forms.json()[0]["commands"][0]["name"] == "Провести" handlers = forms.json()[0]["command_handlers"] assert next(iter(handlers.values()))["name"] == "ПровестиКоманда" def test_object_ui_endpoint_filters_forms_by_1c_object(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """
""", encoding="utf-8", ) module = tmp_path / "Documents" / "ЗаказПокупателя" / "Ext" / "ObjectModule.bsl" module.parent.mkdir(parents=True) module.write_text("Процедура ПровестиКоманда()\nКонецПроцедуры\n", encoding="utf-8") client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-object-ui"}, ) assert indexed.status_code == 200 response = client.get("/projects/demo-api-object-ui/objects/ui/Документ.ЗаказПокупателя") assert response.status_code == 200 payload = response.json() assert payload["object"]["name"] == "ЗаказПокупателя" assert [form["form"]["name"] for form in payload["forms"]] == ["ФормаДокумента"] assert payload["forms"][0]["commands"][0]["name"] == "Провести" assert next(iter(payload["forms"][0]["command_handlers"].values()))["name"] == "ПровестиКоманда" def test_object_impact_endpoint(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """
""", encoding="utf-8", ) module = tmp_path / "Documents" / "ЗаказПокупателя" / "Ext" / "ObjectModule.bsl" module.parent.mkdir(parents=True) module.write_text( """ Процедура Проведение() Движения.ОстаткиТоваров.Записать(); КонецПроцедуры Процедура ПровестиКоманда() КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-object-impact"}, ) assert indexed.status_code == 200 response = client.get( "/projects/demo-object-impact/objects/impact/Документ.ЗаказПокупателя" ) assert response.status_code == 200 payload = response.json() assert payload["object"]["name"] == "ЗаказПокупателя" assert payload["modules"][0]["name"] == "ObjectModule" assert [routine["name"] for routine in payload["routines"]] == ["ПровестиКоманда", "Проведение"] assert payload["forms"][0]["name"] == "ФормаДокумента" assert payload["commands"][0]["name"] == "Провести" assert payload["attributes"][0]["name"] == "Контрагент" assert payload["tabular_sections"][0]["name"] == "Товары" assert next(iter(payload["tabular_section_columns"].values()))[0]["name"] == "Номенклатура" assert payload["roles"][0]["name"] == "Менеджер" assert payload["role_access"][0]["permissions"]["post"] == "true" assert payload["writes"][0]["name"] == "ОстаткиТоваров" object_access = client.get( "/projects/demo-object-impact/access/objects/Документ.ЗаказПокупателя/roles" ) assert object_access.status_code == 200 assert object_access.json()["grants"][0]["role"]["name"] == "Менеджер" assert object_access.json()["grants"][0]["permissions"]["write"] == "true" role_access = client.get("/projects/demo-object-impact/access/roles/Роль.Менеджер/objects") assert role_access.status_code == 200 assert role_access.json()["objects"][0]["qualified_name"] == "Документ.ЗаказПокупателя" assert role_access.json()["grants"][0]["permissions"]["post"] == "true" def test_object_attributes_endpoint(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-object-attributes"}, ) assert indexed.status_code == 200 response = client.get("/projects/demo-object-attributes/objects/attributes/Документ.ЗаказПокупателя") assert response.status_code == 200 assert [row["name"] for row in response.json()["results"]] == ["Контрагент", "СуммаДокумента"] def test_authoring_context_and_completion_preview(tmp_path: Path): project_id = f"authoring-api-{uuid4()}" (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) module = tmp_path / "Documents" / "ЗаказПокупателя" / "Ext" / "ObjectModule.bsl" module.parent.mkdir(parents=True) source_text = """ Процедура Проведение(Отказ, РежимПроведения) Сумма = 0; Движения.ОстаткиТоваров.Записать(); КонецПроцедуры """ module.write_text(source_text, encoding="utf-8") client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 create_authoring_session(client, project_id, "task.authoring", "session.authoring") create_authoring_session(client, project_id, "task.rollback", "session.rollback") context = client.post( f"/projects/{project_id}/authoring/context", json={ "object_name": "Документ.ЗаказПокупателя", "routine_name": "Проведение", "cursor_line": 3, "source_text": source_text, }, ) assert context.status_code == 200 payload = context.json() assert payload["object"]["qualified_name"] == "Документ.ЗаказПокупателя" assert payload["routine"]["name"] == "Проведение" assert "Сумма" in payload["local_variables"] assert "Отказ" in payload["parameters"] assert payload["object_attributes"][0]["name"] == "Контрагент" assert payload["tabular_sections"][0]["name"] == "Товары" assert "ЗначениеЗаполнено" in payload["available_methods"] preview = client.post( f"/projects/{project_id}/authoring/completion-preview", json={ "object_name": "Документ.ЗаказПокупателя", "routine_name": "Проведение", "cursor_line": 3, "source_text": source_text, "intent": "fill-check", }, ) assert preview.status_code == 200 preview_payload = preview.json() assert preview_payload["allowed"] is False assert "ЗначениеЗаполнено(Контрагент)" in preview_payload["insert_text"] assert any(check["name"] == "apply" and check["status"] == "BLOCKED" for check in preview_payload["checks"]) assert preview_payload["semantic_diff"][0]["kind"] == "ADD" html5_preview = client.post( f"/html5/projects/{project_id}/authoring/completion-preview", data={ "object_name": "Документ.ЗаказПокупателя", "routine_name": "Проведение", "cursor_line": "3", "source_text": source_text, "intent": "fill-check", }, ) assert html5_preview.status_code == 200 assert "text/html" in html5_preview.headers["content-type"] assert "data-html5-authoring-preview-result" in html5_preview.text assert "ЗначениеЗаполнено(Контрагент)" in html5_preview.text assert "BLOCKED" in html5_preview.text assert "ADD" in html5_preview.text assert " """, encoding="utf-8", ) module = tmp_path / "Documents" / "ЗаказПокупателя" / "Ext" / "ObjectModule.bsl" module.parent.mkdir(parents=True) source_text = "Процедура Проверить()\n Сообщить(Телефон);\nКонецПроцедуры\n" module.write_text(source_text, encoding="utf-8") client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 create_authoring_session(client, project_id, "task.privacy", "session.privacy") schema = client.get(f"/projects/{project_id}/objects/schema/Документ.ЗаказПокупателя") phone_lineage = schema.json()["attributes"][0]["lineage_id"] marker = client.post( f"/projects/{project_id}/privacy/markers", json={"target_id": phone_lineage, "classification": "PERSONAL_DATA", "reason": "phone number"}, ) assert marker.status_code == 200 preview = client.post( f"/projects/{project_id}/authoring/semantic-diff-preview", json={ "object_name": "Документ.ЗаказПокупателя", "routine_name": "Проверить", "source_path": str(module), "original_text": source_text, "proposed_text": source_text.replace("Сообщить(Телефон);", "Сообщить(Телефон);\n Возврат;"), "task_id": "task.privacy", "session_id": "session.privacy", "user_id": "dev.ivan", }, ) assert preview.status_code == 200 preview_payload = preview.json() assert any(check["name"] == "privacy" and check["status"] == "BLOCKED" for check in preview_payload["checks"]) apply_response = client.post( f"/projects/{project_id}/authoring/apply-change-set", json={ "object_name": "Документ.ЗаказПокупателя", "routine_name": "Проверить", "source_path": str(module), "original_text": source_text, "proposed_text": source_text.replace("Сообщить(Телефон);", "Сообщить(Телефон);\n Возврат;"), "task_id": "task.privacy", "session_id": "session.privacy", "user_id": "dev.ivan", "expected_next_version_id": preview_payload["version_preview"]["next_version_id"], "approved_by": "dev.ivan", }, ) assert apply_response.status_code == 409 assert any(check["name"] == "privacy" for check in apply_response.json()["detail"]["blocked_checks"]) def test_authoring_metadata_object_preview_and_apply(tmp_path: Path): project_id = f"metadata-authoring-api-{uuid4()}" (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 create_authoring_session(client, project_id, "task.metadata", "session.metadata") create_authoring_session(client, project_id, "task.metadata.rollback", "session.metadata.rollback") draft = { "object_kind": "DOCUMENT", "name": "ЗаявкаНаЗакупку", "synonym": "Заявка на закупку", "attributes": [ {"name": "Контрагент", "type": "СправочникСсылка.Контрагенты", "required": True} ], "tabular_sections": [ { "name": "Товары", "attributes": [ {"name": "Номенклатура", "type": "СправочникСсылка.Номенклатура"}, {"name": "Количество", "type": "Число"}, ], } ], "forms": ["ФормаДокумента"], "commands": [{"name": "Заполнить", "handler": "ЗаполнитьКоманда"}], "task_id": "task.metadata", "session_id": "session.metadata", "user_id": "dev.ivan", } preview = client.post(f"/projects/{project_id}/authoring/metadata-object-preview", json=draft) assert preview.status_code == 200 preview_payload = preview.json() assert preview_payload["target"]["qualified_name"] == "Документ.ЗаявкаНаЗакупку" assert preview_payload["changed"] is True assert preview_payload["version_preview"]["apply_available"] is True assert any(check["name"] == "task-session" and check["status"] == "OK" for check in preview_payload["checks"]) assert any(check["name"] == "rbac" and check["status"] == "OK" for check in preview_payload["checks"]) assert any("Реквизит.Контрагент" in row["text"] for row in preview_payload["semantic_diff"]) assert any("ТабличнаяЧасть.Товары" in row["text"] for row in preview_payload["semantic_diff"]) assert any("Команда.Заполнить" in row["text"] for row in preview_payload["semantic_diff"]) html5_preview = client.post( f"/html5/projects/{project_id}/authoring/metadata-object-preview", data={ "object_kind": "DOCUMENT", "name": "ЗаявкаНаЗакупкуHtml5", "synonym": "Заявка на закупку HTML5", "attributes": "Контрагент:СправочникСсылка.Контрагенты", "tabular_sections": "Товары[Номенклатура:СправочникСсылка.Номенклатура;Количество:Число]", "forms": "ФормаДокумента", "commands": "Заполнить:ЗаполнитьКоманда", "task_id": "task.metadata", "session_id": "session.metadata", "user_id": "dev.ivan", }, ) assert html5_preview.status_code == 200 assert "text/html" in html5_preview.headers["content-type"] assert "data-html5-metadata-preview-result" in html5_preview.text assert "data-html5-metadata-apply-form" in html5_preview.text assert f'hx-post="/html5/projects/{project_id}/authoring/apply-metadata-object"' in html5_preview.text assert "Документ.ЗаявкаНаЗакупкуHtml5" in html5_preview.text assert "Реквизит.Контрагент" in html5_preview.text assert "", encoding="utf-8") client = TestClient(app) project_id = f"metadata-kind-catalog-{uuid4()}" indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 expected_targets = { "COMMON_MODULE": "ОбщийМодуль.ИнтеграцияСервер", "REPORT": "Отчет.АнализПродаж", "DATA_PROCESSOR": "Обработка.ЗагрузкаДанных", "INFORMATION_REGISTER": "РегистрСведений.НастройкиОбмена", "ACCUMULATION_REGISTER": "РегистрНакопления.ОстаткиТоваров", "ACCOUNTING_REGISTER": "РегистрБухгалтерии.Хозрасчетный", "CALCULATION_REGISTER": "РегистрРасчета.Начисления", "BUSINESS_PROCESS": "БизнесПроцесс.СогласованиеЗаявки", "TASK": "Задача.ЗадачаИсполнителя", "CHART_OF_ACCOUNTS": "ПланСчетов.Управленческий", } for object_kind, qualified_name in expected_targets.items(): response = client.post( f"/projects/{project_id}/authoring/metadata-object-preview", json={ "object_kind": object_kind, "name": qualified_name.split(".", 1)[1], "synonym": qualified_name.split(".", 1)[1], "attributes": [{"name": "Комментарий", "type": "Строка"}], "forms": ["ФормаОбъекта"], "commands": [{"name": "Обновить", "handler": "ОбновитьКоманда"}], }, ) assert response.status_code == 200 payload = response.json() assert payload["target"]["qualified_name"] == qualified_name assert payload["changed"] is True def test_authoring_context_includes_query_tables(tmp_path: Path): project_id = f"authoring-query-context-{uuid4()}" module = tmp_path / "demo_module.bsl" source_text = """ Процедура ПроверитьОстатки() Запрос = Новый Запрос; Запрос.Текст = "ВЫБРАТЬ Остатки.Номенклатура ИЗ РегистрНакопления.ОстаткиТоваров КАК Остатки"; КонецПроцедуры """ module.write_text(source_text, encoding="utf-8") client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 context = client.post( f"/projects/{project_id}/authoring/context", json={ "routine_name": "ПроверитьОстатки", "cursor_line": 3, "source_text": source_text, }, ) assert context.status_code == 200 assert [row["qualified_name"] for row in context.json()["query_tables"]] == [ "РегистрНакопления.ОстаткиТоваров" ] def test_object_schema_endpoint(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-object-schema"}, ) assert indexed.status_code == 200 response = client.get("/projects/demo-object-schema/objects/schema/Документ.ЗаказПокупателя") assert response.status_code == 200 payload = response.json() assert payload["object"]["name"] == "ЗаказПокупателя" assert payload["attributes"][0]["name"] == "Контрагент" assert payload["tabular_sections"][0]["tabular_section"]["name"] == "Товары" assert payload["tabular_sections"][0]["columns"][0]["name"] == "Номенклатура" def test_knowledge_schema_coverage_endpoint(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-schema-coverage"}, ) assert indexed.status_code == 200 snapshot = client.get("/projects/demo-api-schema-coverage/snapshot/export").json() document_lineage = next( node["lineage_id"] for node in snapshot["nodes"] if node["kind"] == "DOCUMENT" ) knowledge = client.post( "/knowledge", json={ "record_id": "knowledge.schema.document", "scope": "PROJECT", "title": "Заказ покупателя", "body": "Описание структуры документа.", "related_lineages": [document_lineage], }, ) assert knowledge.status_code == 200 response = client.get("/projects/demo-api-schema-coverage/knowledge/schema-coverage") assert response.status_code == 200 payload = response.json() assert payload["covered_count"] == 1 assert payload["uncovered_count"] == 3 assert any(item["node"]["name"] == "ЗаказПокупателя" and item["record_count"] == 1 for item in payload["items"]) assert {node["name"] for node in payload["uncovered"]} == {"Контрагент", "Номенклатура", "Товары"} def test_knowledge_pack_import_endpoint(): client = TestClient(app) pack_id = f"bsp.core.{uuid4()}" record_id = f"knowledge.bsp.{uuid4()}" response = client.post( "/knowledge/packs", json={ "pack_id": pack_id, "name": "БСП базовые правила", "vendor": "1C", "version": "3.1", "records": [ { "record_id": record_id, "scope": "GLOBAL", "title": "БСП роли", "body": "Рекомендации по настройке ролей БСП.", } ], }, ) assert response.status_code == 200 packs = client.get("/knowledge/packs") assert packs.status_code == 200 assert any(pack["pack_id"] == pack_id for pack in packs.json()) search = client.get("/knowledge/search", params={"q": "БСП"}) assert search.status_code == 200 record = next(item for item in search.json()["results"] if item["record_id"] == record_id) assert f"pack:{pack_id}" in record["tags"] assert "vendor:1C" in record["tags"] assert record["attributes"]["pack_version"] == "3.1" summary = client.get("/admin/summary") assert summary.status_code == 200 assert summary.json()["knowledge_packs"] >= 1 def test_object_ownership_endpoint(tmp_path: Path): project_id = f"demo-api-ownership-{uuid4()}" (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": project_id}, ) assert indexed.status_code == 200 snapshot = client.get(f"/projects/{project_id}/snapshot/export").json() document_lineage = next( node["lineage_id"] for node in snapshot["nodes"] if node["kind"] == "DOCUMENT" ) user_id = f"user.{uuid4()}" user = client.post( "/collaboration/users", json={"user_id": user_id, "display_name": "Owner"}, ) assert user.status_code == 200 ownership = client.post( f"/projects/{project_id}/ownership", json={ "target_id": document_lineage, "owner_user_id": user_id, "role": "RESPONSIBLE", "assigned_by": user_id, }, ) assert ownership.status_code == 200 project_ownership = client.get(f"/projects/{project_id}/ownership") assert project_ownership.status_code == 200 assert project_ownership.json()[0]["target_id"] == document_lineage response = client.get(f"/projects/{project_id}/objects/ownership/Документ.ЗаказПокупателя") assert response.status_code == 200 payload = response.json() assert payload["object"]["name"] == "ЗаказПокупателя" assert payload["owners"][0]["owner_user_id"] == user_id assert payload["owners"][0]["role"] == "RESPONSIBLE" report = client.get(f"/projects/{project_id}/report") assert report.status_code == 200 assert report.json()["ownership_count"] == 1 assert report.json()["unowned_object_count"] == 0 review = client.get(f"/projects/{project_id}/review") assert review.status_code == 200 assert not any(finding["title"] == "Missing 1C object owner" for finding in review.json()) def test_project_comments_endpoint(tmp_path: Path): project_id = f"demo-api-comments-{uuid4()}" (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": project_id}, ) assert indexed.status_code == 200 snapshot = client.get(f"/projects/{project_id}/snapshot/export").json() document_lineage = next( node["lineage_id"] for node in snapshot["nodes"] if node["kind"] == "DOCUMENT" ) user_id = f"user.{uuid4()}" user = client.post( "/collaboration/users", json={"user_id": user_id, "display_name": "Reviewer"}, ) assert user.status_code == 200 comment = client.post( f"/projects/{project_id}/comments", json={ "comment_id": f"comment.{uuid4()}", "target_id": document_lineage, "user_id": user_id, "body": "Проверить правила проведения.", }, ) assert comment.status_code == 200 comments = client.get(f"/projects/{project_id}/comments") assert comments.status_code == 200 assert comments.json()[0]["target_id"] == document_lineage target_comments = client.get(f"/projects/{project_id}/comments/{document_lineage}") assert target_comments.status_code == 200 assert target_comments.json()[0]["body"] == "Проверить правила проведения." activity = client.get(f"/projects/{project_id}/activity") assert activity.status_code == 200 assert activity.json()[0]["verb"] == "ADD_COMMENT" def test_review_endpoint_reports_missing_1c_object_owner(tmp_path: Path): project_id = f"demo-api-missing-owner-{uuid4()}" (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": project_id}, ) assert indexed.status_code == 200 response = client.get(f"/projects/{project_id}/review") assert response.status_code == 200 assert any( finding["title"] == "Missing 1C object owner" and "Документ.ЗаказПокупателя" in finding["message"] for finding in response.json() ) def test_review_endpoint_reports_missing_1c_schema_knowledge(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-schema-knowledge-review"}, ) assert indexed.status_code == 200 snapshot = client.get("/projects/demo-api-schema-knowledge-review/snapshot/export").json() document_lineage = next( node["lineage_id"] for node in snapshot["nodes"] if node["kind"] == "DOCUMENT" ) knowledge = client.post( "/knowledge", json={ "record_id": "knowledge.schema.review.document", "scope": "PROJECT", "title": "Документ заказа", "body": "Описание документа.", "related_lineages": [document_lineage], }, ) assert knowledge.status_code == 200 response = client.get("/projects/demo-api-schema-knowledge-review/review") assert response.status_code == 200 knowledge_findings = [ finding for finding in response.json() if finding["title"] == "Missing 1C schema knowledge" ] assert any("Документ.ЗаказПокупателя.Контрагент" in finding["message"] for finding in knowledge_findings) assert not any("Schema node Документ.ЗаказПокупателя has no" in finding["message"] for finding in knowledge_findings) def test_privacy_marker_endpoint_classifies_1c_attribute(tmp_path: Path): project_id = f"demo-api-privacy-{uuid4()}" (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": project_id}, ) assert indexed.status_code == 200 snapshot = client.get(f"/projects/{project_id}/snapshot/export").json() phone_lineage = next( node["lineage_id"] for node in snapshot["nodes"] if node["kind"] == "ATTRIBUTE" and node["name"] == "Телефон" ) marker = client.post( f"/projects/{project_id}/privacy/markers", json={ "target_id": phone_lineage, "classification": "PERSONAL_DATA", "reason": "Контактный телефон клиента", }, ) assert marker.status_code == 200 markers = client.get(f"/projects/{project_id}/privacy/markers") assert markers.status_code == 200 assert markers.json()[0]["classification"] == "PERSONAL_DATA" object_privacy = client.get(f"/projects/{project_id}/objects/privacy/Документ.ЗаказПокупателя") assert object_privacy.status_code == 200 assert object_privacy.json()["markers"][0]["target_id"] == phone_lineage report = client.get(f"/projects/{project_id}/report") assert report.status_code == 200 assert report.json()["privacy_marker_count"] == 1 assert report.json()["sensitive_candidate_count"] == 1 assert report.json()["unclassified_sensitive_count"] == 0 review = client.get(f"/projects/{project_id}/review") assert review.status_code == 200 assert not any(finding["title"] == "Unclassified sensitive 1C field" for finding in review.json()) def test_review_endpoint_reports_unclassified_sensitive_1c_field(tmp_path: Path): project_id = f"demo-api-privacy-review-{uuid4()}" (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": project_id}, ) assert indexed.status_code == 200 response = client.get(f"/projects/{project_id}/review") assert response.status_code == 200 assert any( finding["title"] == "Unclassified sensitive 1C field" and "Справочник.Контрагенты.ИНН" in finding["message"] for finding in response.json() ) def test_project_report_includes_1c_schema_summary(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-schema-report"}, ) assert indexed.status_code == 200 response = client.get("/projects/demo-api-schema-report/report") assert response.status_code == 200 payload = response.json() assert payload["attribute_count"] == 2 assert payload["object_attribute_count"] == 1 assert payload["tabular_section_count"] == 2 assert payload["tabular_section_column_count"] == 1 assert payload["empty_tabular_sections"] == ["Документ.ЗаказПокупателя.Услуги"] def test_review_endpoint_reports_empty_1c_tabular_section(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-empty-tabular-section"}, ) assert indexed.status_code == 200 response = client.get("/projects/demo-api-empty-tabular-section/review") assert response.status_code == 200 assert any( finding["title"] == "Empty 1C tabular section" and "Документ.ЗаказПокупателя.Услуги" in finding["message"] for finding in response.json() ) def test_review_endpoint_reports_tabular_section_without_subject_column(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-no-subject-column"}, ) assert indexed.status_code == 200 response = client.get("/projects/demo-api-no-subject-column/review") assert response.status_code == 200 assert any( finding["title"] == "No subject column in 1C tabular section" and finding["severity"] == "INFO" for finding in response.json() ) def test_object_tabular_sections_endpoint(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-object-tabular-sections"}, ) assert indexed.status_code == 200 response = client.get( "/projects/demo-object-tabular-sections/objects/tabular-sections/Документ.ЗаказПокупателя" ) assert response.status_code == 200 assert response.json()["results"][0]["name"] == "Товары" columns = client.get( "/projects/demo-object-tabular-sections/objects/tabular-sections/Документ.ЗаказПокупателя/columns" ) assert columns.status_code == 200 assert columns.json()[0]["tabular_section"]["name"] == "Товары" assert columns.json()[0]["columns"][0]["name"] == "Номенклатура" def test_incremental_xml_updates_metadata_semantics(tmp_path: Path): xml = tmp_path / "metadata.xml" xml.write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-xml-incremental"}, ) assert indexed.status_code == 200 xml.write_text( """
""", encoding="utf-8", ) incremental = client.post( "/projects/demo-api-xml-incremental/incremental/file", json={"path": str(xml)}, ) assert incremental.status_code == 200 assert incremental.json()["added_nodes"] >= 2 forms = client.get("/projects/demo-api-xml-incremental/ui/forms") assert forms.status_code == 200 assert forms.json()[0]["commands"][0]["name"] == "Провести" def test_scheduled_jobs_endpoint(tmp_path: Path): (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) module = tmp_path / "CommonModules" / "РегламентныеОперации" / "Module.bsl" module.parent.mkdir(parents=True) module.write_text("Процедура ОбновитьЦены()\nКонецПроцедуры\n", encoding="utf-8") client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-scheduled-jobs"}, ) assert indexed.status_code == 200 response = client.get("/projects/demo-api-scheduled-jobs/jobs/scheduled") assert response.status_code == 200 payload = response.json() assert payload[0]["name"] == "ОбновлениеЦен" assert payload[0]["routine"]["name"] == "ОбновитьЦены" assert payload[0]["schedule"] == "КаждыйДень" impact = client.get( "/projects/demo-api-scheduled-jobs/objects/impact/РегламентноеЗадание.ОбновлениеЦен" ) assert impact.status_code == 200 assert impact.json()["jobs"][0]["name"] == "ОбновлениеЦен" assert impact.json()["routines"][0]["name"] == "ОбновитьЦены" def test_integrations_endpoint(tmp_path: Path): (tmp_path / "integration.bsl").write_text( """ Процедура Отправить() Соединение = Новый HTTPСоединение("api.example.local"); Адрес = "https://api.example.local/orders"; КонецПроцедуры """, encoding="utf-8", ) (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": "demo-api-integrations"}, ) assert indexed.status_code == 200 response = client.get("/projects/demo-api-integrations/integrations") filtered = client.get( "/projects/demo-api-integrations/integrations", params={"kind": "EXCHANGE_PLAN"}, ) assert response.status_code == 200 assert any(item["kind"] == "HTTP_SERVICE" for item in response.json()) assert filtered.status_code == 200 assert filtered.json()[0]["name"] == "ОбменСКассой" def test_patterns_endpoint_finds_repeated_writes(tmp_path: Path): (tmp_path / "module.bsl").write_text( """ Процедура ПровестиЗаказ() Движения.ОстаткиТоваров.Записать(); КонецПроцедуры Процедура ОтменитьЗаказ() Движения.ОстаткиТоваров.Записать(); КонецПроцедуры """, encoding="utf-8", ) client = TestClient(app) project_id = f"demo-api-patterns-{uuid4()}" indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 response = client.get(f"/projects/{project_id}/patterns") assert response.status_code == 200 assert any( pattern["kind"] == "REPEATED_TABLE_WRITE" and pattern["support"] == 2 and pattern["targets"][0]["name"] == "ОстаткиТоваров" for pattern in response.json() ) def test_incremental_reprojects_neo4j_when_project_was_projected(monkeypatch, tmp_path: Path): module = tmp_path / "demo_module.bsl" module.write_text("Процедура Проведение()\nКонецПроцедуры\n", encoding="utf-8") client = TestClient(app) project_id = f"demo-api-neo4j-incremental-{uuid4().hex}" projected: list[tuple[str, int]] = [] applied_deltas: list[tuple[str, int, int]] = [] async def fake_project_snapshot(project_id: str, snapshot): projected.append((project_id, len(snapshot.nodes))) return {"nodes": len(snapshot.nodes), "edges": len(snapshot.edges)} async def fake_apply_delta(project_id: str, delta): applied_deltas.append((project_id, len(delta.added_nodes), len(delta.added_edges))) return {"nodes": len(delta.added_nodes), "edges": len(delta.added_edges)} monkeypatch.setattr(main, "_project_snapshot_to_neo4j", fake_project_snapshot) monkeypatch.setattr(main, "_apply_delta_to_neo4j", fake_apply_delta) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 projection = client.post(f"/projects/{project_id}/graph/neo4j/project") assert projection.status_code == 200 assert projected[-1][0] == project_id module.write_text( """ Процедура Проведение() Движения.ОстаткиТоваров.Записать(); КонецПроцедуры """, encoding="utf-8", ) incremental = client.post( f"/projects/{project_id}/incremental/file", json={"path": str(module)}, ) assert incremental.status_code == 200 assert incremental.json()["neo4j_projected"] is True assert incremental.json()["neo4j_error"] is None assert len(projected) == 1 assert applied_deltas == [(project_id, incremental.json()["added_nodes"], incremental.json()["added_edges"])] def test_index_missing_path_returns_404(): client = TestClient(app) response = client.post("/projects/index", json={"path": "Z:/definitely/missing"}) assert response.status_code == 404 def test_index_demo_project_endpoint(): client = TestClient(app) response = client.post("/projects/demo/index") assert response.status_code == 200 assert response.json()["snapshot"]["project_id"] == "demo" def test_collaboration_endpoints(): client = TestClient(app) suffix = uuid4().hex user_id = f"user.api.collaboration.{suffix}" task_id = f"task.api.collaboration.{suffix}" session_id = f"session.api.collaboration.{suffix}" user = client.post( "/collaboration/users", json={"user_id": user_id, "display_name": "API Tester"}, ) assert user.status_code == 200 task = client.post( "/collaboration/tasks", json={ "task_id": task_id, "project_id": "demo-api", "title": "Проверить индекс", "assignee_user_id": user_id, }, ) assert task.status_code == 200 session = client.post( "/collaboration/sessions", json={ "session": { "session_id": session_id, "task_id": task_id, "user_id": user_id, } }, ) assert session.status_code == 200 finished = client.post(f"/collaboration/sessions/{session_id}/finish") assert finished.status_code == 200 assert finished.json()["finished_at"] is not None feed = client.get("/projects/demo-api/activity") assert feed.status_code == 200 assert feed.json()[0]["verb"] == "FINISH_SESSION" assert any(event["verb"] == "UPSERT_TASK" for event in feed.json()) permission = client.get(f"/security/users/{user_id}/permissions/READ_GRAPH") assert permission.status_code == 200 assert permission.json()["allowed"] is False grant = client.post(f"/security/users/{user_id}/roles/viewer") assert grant.status_code == 200 permission = client.get(f"/security/users/{user_id}/permissions/READ_GRAPH") assert permission.status_code == 200 assert permission.json()["allowed"] is True permissions = client.get(f"/security/users/{user_id}/permissions") assert permissions.status_code == 200 assert permissions.json()["permissions"] == ["READ_GRAPH"] def test_operations_endpoints(): client = TestClient(app) job = client.post("/operations/jobs", json={"job_id": "job.api", "kind": "INDEX_PROJECT"}) assert job.status_code == 200 update = client.patch( "/operations/jobs/job.api", json={"status": "SUCCEEDED", "result": {"indexed": True}}, ) assert update.status_code == 200 assert update.json()["status"] == "SUCCEEDED" metric = client.post( "/operations/metrics", json={"metric_id": "metric.api", "name": "api.requests", "value": 1.0}, ) assert metric.status_code == 200 packages = client.post( "/marketplace/packages", json={"package_id": "pack.api", "name": "BSP Knowledge", "version": "1.0.0"}, ) assert packages.status_code == 200 license_response = client.get("/license") assert license_response.status_code == 200 assert license_response.json()["valid"] is True summary = client.get("/admin/summary") assert summary.status_code == 200 assert "stored_snapshots" in summary.json() def test_ai_usage_endpoints(tmp_path: Path): project_id = f"demo-api-ai-usage-{uuid4()}" module = tmp_path / "module.bsl" module.write_text("Процедура Проверить()\nКонецПроцедуры\n", encoding="utf-8") client = TestClient(app) indexed = client.post( "/projects/index", json={"path": str(tmp_path), "project_id": project_id}, ) assert indexed.status_code == 200 usage_id = f"usage.{uuid4()}" usage = client.post( "/ai/usage", json={ "usage_id": usage_id, "project_id": project_id, "user_id": "user.ai", "model": "gpt-test", "operation": "review", "prompt_tokens": 120, "completion_tokens": 30, "cost": 0.02, }, ) assert usage.status_code == 200 assert usage.json()["total_tokens"] == 150 records = client.get("/ai/usage", params={"project_id": project_id}) assert records.status_code == 200 assert records.json()[0]["usage_id"] == usage_id summary = client.get("/ai/usage/summary", params={"project_id": project_id}) assert summary.status_code == 200 assert summary.json()["request_count"] == 1 assert summary.json()["total_tokens"] == 150 policy = client.get("/ai/policy", params={"user_id": "user.ai"}) assert policy.status_code == 200 assert policy.json()["used_tokens"] >= 150 assert policy.json()["remaining_tokens"] is not None report = client.get(f"/projects/{project_id}/report") assert report.status_code == 200 assert report.json()["ai_usage_request_count"] == 1 assert report.json()["ai_usage_total_tokens"] == 150 def test_ai_answer_policy_allows_with_knowledge_context(tmp_path: Path): project_id = f"demo-api-ai-answer-policy-{uuid4()}" (tmp_path / "module.bsl").write_text("Процедура Проверить()\nКонецПроцедуры\n", encoding="utf-8") client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 snapshot = client.get(f"/projects/{project_id}/snapshot/export").json() routine_lineage = next(node["lineage_id"] for node in snapshot["nodes"] if node["kind"] == "PROCEDURE") knowledge = client.post( "/knowledge", json={ "record_id": f"knowledge.answer.{uuid4()}", "scope": "PROJECT", "title": "Правила проверки", "body": "Процедура Проверить выполняет контроль.", "related_lineages": [routine_lineage], }, ) assert knowledge.status_code == 200 response = client.post( f"/projects/{project_id}/ai/answer-policy", json={ "user_id": "user.answer", "question": "Что делает процедура Проверить?", "related_lineages": [routine_lineage], "estimated_tokens": 100, }, ) assert response.status_code == 200 payload = response.json() assert payload["allowed"] is True assert payload["reasons"] == [] assert payload["knowledge_records"][0]["title"] == "Правила проверки" def test_ai_answer_policy_blocks_sensitive_context(tmp_path: Path): project_id = f"demo-api-ai-answer-policy-sensitive-{uuid4()}" (tmp_path / "metadata.xml").write_text( """ """, encoding="utf-8", ) client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.status_code == 200 snapshot = client.get(f"/projects/{project_id}/snapshot/export").json() inn_lineage = next( node["lineage_id"] for node in snapshot["nodes"] if node["kind"] == "ATTRIBUTE" and node["name"] == "ИНН" ) marker = client.post( f"/projects/{project_id}/privacy/markers", json={ "target_id": inn_lineage, "classification": "PERSONAL_DATA", "reason": "Налоговый идентификатор контрагента", }, ) assert marker.status_code == 200 response = client.post( f"/projects/{project_id}/ai/answer-policy", json={ "user_id": "user.answer", "question": "Объясни реквизит ИНН", "related_lineages": [inn_lineage], "estimated_tokens": 100, "require_knowledge": False, }, ) assert response.status_code == 200 payload = response.json() assert payload["allowed"] is False assert "Privacy policy blocks sensitive 1C context" in payload["reasons"] assert payload["privacy_markers"][0]["classification"] == "PERSONAL_DATA" def test_operation_job_runner_indexes_project(tmp_path: Path): module = tmp_path / "job_module.bsl" module.write_text("Процедура JobRun()\nКонецПроцедуры\n", encoding="utf-8") client = TestClient(app) job_id = f"job.index.{uuid4().hex}" job = client.post( "/operations/jobs", json={ "job_id": job_id, "kind": "INDEX_PROJECT", "payload": {"path": str(tmp_path), "project_id": "demo-job"}, }, ) assert job.status_code == 200 run = client.post(f"/operations/jobs/{job_id}/run") assert run.status_code == 200 payload = run.json() assert payload["status"] == "SUCCEEDED" assert payload["result"]["snapshot"]["project_id"] == "demo-job" def test_operation_job_runner_records_failure(): client = TestClient(app) job_id = f"job.fail.{uuid4().hex}" client.post( "/operations/jobs", json={"job_id": job_id, "kind": "INDEX_PROJECT", "payload": {}}, ) run = client.post(f"/operations/jobs/{job_id}/run") assert run.status_code == 200 assert run.json()["status"] == "FAILED" assert "payload.path" in run.json()["error"] def test_neo4j_endpoints_contract_for_unindexed_project(): client = TestClient(app) status = client.get("/graph/neo4j/status") assert status.status_code == 200 assert status.json()["status"] in {"ok", "unavailable"} projection = client.post("/projects/not-indexed/graph/neo4j/project") assert projection.status_code == 404 def test_neo4j_callees_endpoint_contract(monkeypatch): async def fake_query(project_id: str, routine_name: str, *, outgoing: bool): assert project_id == "demo" assert routine_name == "Проведение" assert outgoing is True return [ { "lineage_id": "lineage.procedure.check", "kind": "PROCEDURE", "name": "ПроверитьОстатки", "qualified_name": "Module.ПроверитьОстатки", } ] monkeypatch.setattr(main, "_neo4j_routine_query", fake_query) client = TestClient(app) response = client.get("/projects/demo/graph/neo4j/callees/Проведение") assert response.status_code == 200 assert response.json()["results"][0]["name"] == "ПроверитьОстатки" def test_neo4j_writes_endpoint_contract(monkeypatch): async def fake_query(project_id: str, routine_name: str, *, relation: str): assert project_id == "demo" assert routine_name == "Проведение" assert relation == "WRITES" return [ { "lineage_id": "lineage.register.stock", "kind": "REGISTER", "name": "ОстаткиТоваров", "qualified_name": "РегистрНакопления.ОстаткиТоваров", } ] monkeypatch.setattr(main, "_neo4j_relation_query", fake_query) client = TestClient(app) response = client.get("/projects/demo/graph/neo4j/writes/Проведение") assert response.status_code == 200 assert response.json()["results"][0]["name"] == "ОстаткиТоваров" def test_neo4j_integrations_endpoints_contract(monkeypatch): async def fake_integrations(project_id: str, *, kind: str | None = None): assert project_id == "demo" assert kind == "HTTP_SERVICE" return [ { "endpoint_id": "lineage.integration.http", "name": "https://api.example.local/orders", "kind": "HTTP_SERVICE", "direction": "OUTBOUND", "owner": "IntegrationModule", "attributes": {"url": "https://api.example.local/orders"}, } ] async def fake_modules(project_id: str, integration_name: str): assert project_id == "demo" assert integration_name == "https://api.example.local/orders" return [ { "lineage_id": "lineage.module.integration", "kind": "MODULE", "name": "IntegrationModule", "qualified_name": "IntegrationModule", } ] monkeypatch.setattr(main, "_neo4j_integrations_query", fake_integrations) monkeypatch.setattr(main, "_neo4j_integration_modules_query", fake_modules) client = TestClient(app) integrations = client.get( "/projects/demo/graph/neo4j/integrations", params={"kind": "HTTP_SERVICE"}, ) modules = client.get( "/projects/demo/graph/neo4j/integrations/https://api.example.local/orders/modules" ) assert integrations.status_code == 200 assert integrations.json()[0]["kind"] == "HTTP_SERVICE" assert modules.status_code == 200 assert modules.json()["results"][0]["name"] == "IntegrationModule" def test_neo4j_object_impact_endpoint_contract(monkeypatch): async def fake_query(project_id: str, object_name: str): assert project_id == "demo" assert object_name == "Документ.ЗаказПокупателя" return { "object": { "lineage_id": "lineage.document.order", "kind": "DOCUMENT", "name": "ЗаказПокупателя", "qualified_name": "Документ.ЗаказПокупателя", }, "modules": [ { "lineage_id": "lineage.module.object", "kind": "MODULE", "name": "ObjectModule", "qualified_name": "ObjectModule", } ], "routines": [ { "lineage_id": "lineage.procedure.posting", "kind": "PROCEDURE", "name": "Проведение", "qualified_name": "ObjectModule.Проведение", } ], "forms": [ { "lineage_id": "lineage.form.main", "kind": "FORM", "name": "ФормаДокумента", "qualified_name": "Документ.ЗаказПокупателя.ФормаДокумента", } ], "commands": [ { "lineage_id": "lineage.command.post", "kind": "COMMAND", "name": "Провести", "qualified_name": "Документ.ЗаказПокупателя.ФормаДокумента.Провести", } ], "attributes": [ { "lineage_id": "lineage.attribute.counterparty", "kind": "ATTRIBUTE", "name": "Контрагент", "qualified_name": "Документ.ЗаказПокупателя.Контрагент", } ], "tabular_sections": [ { "lineage_id": "lineage.tabular.goods", "kind": "TABULAR_SECTION", "name": "Товары", "qualified_name": "Документ.ЗаказПокупателя.Товары", } ], "tabular_section_columns": { "lineage.tabular.goods": [ { "lineage_id": "lineage.attribute.item", "kind": "ATTRIBUTE", "name": "Номенклатура", "qualified_name": "Документ.ЗаказПокупателя.Товары.Номенклатура", } ] }, "roles": [ { "lineage_id": "lineage.role.manager", "kind": "ROLE", "name": "Менеджер", "qualified_name": "Роль.Менеджер", } ], "role_access": [ { "role": { "lineage_id": "lineage.role.manager", "kind": "ROLE", "name": "Менеджер", "qualified_name": "Роль.Менеджер", }, "permissions": {"read": "true", "write": "true", "post": "true"}, } ], "callees": [], "query_tables": [], "writes": [ { "lineage_id": "lineage.register.stock", "kind": "REGISTER", "name": "ОстаткиТоваров", "qualified_name": "РегистрНакопления.ОстаткиТоваров", } ], } monkeypatch.setattr(main, "_neo4j_object_impact_query", fake_query) client = TestClient(app) response = client.get( "/projects/demo/graph/neo4j/objects/impact/Документ.ЗаказПокупателя" ) assert response.status_code == 200 payload = response.json() assert payload["object"]["name"] == "ЗаказПокупателя" assert payload["modules"][0]["name"] == "ObjectModule" assert payload["routines"][0]["name"] == "Проведение" assert payload["attributes"][0]["name"] == "Контрагент" assert payload["tabular_sections"][0]["name"] == "Товары" assert payload["tabular_section_columns"]["lineage.tabular.goods"][0]["name"] == "Номенклатура" assert payload["roles"][0]["name"] == "Менеджер" assert payload["role_access"][0]["permissions"]["post"] == "true" assert payload["writes"][0]["name"] == "ОстаткиТоваров" def test_neo4j_object_impact_endpoint_returns_404(monkeypatch): async def fake_query(project_id: str, object_name: str): return None monkeypatch.setattr(main, "_neo4j_object_impact_query", fake_query) client = TestClient(app) response = client.get("/projects/demo/graph/neo4j/objects/impact/НеСуществует") assert response.status_code == 404 def test_neo4j_object_ui_endpoint_contract(monkeypatch): async def fake_query(project_id: str, object_name: str): assert project_id == "demo" assert object_name == "Документ.ЗаказПокупателя" return { "object": { "lineage_id": "lineage.document.order", "kind": "DOCUMENT", "name": "ЗаказПокупателя", "qualified_name": "Документ.ЗаказПокупателя", }, "forms": [ { "form": { "lineage_id": "lineage.form.order", "kind": "FORM", "name": "ФормаДокумента", "qualified_name": "Документ.ЗаказПокупателя.ФормаДокумента", }, "commands": [ { "lineage_id": "lineage.command.post", "kind": "COMMAND", "name": "Провести", "qualified_name": "Документ.ЗаказПокупателя.ФормаДокумента.Провести", } ], "elements": [], "command_handlers": { "lineage.command.post": { "lineage_id": "lineage.procedure.post", "kind": "PROCEDURE", "name": "ПровестиКоманда", "qualified_name": "ObjectModule.ПровестиКоманда", } }, } ], } monkeypatch.setattr(main, "_neo4j_object_ui_query", fake_query) client = TestClient(app) response = client.get("/projects/demo/graph/neo4j/objects/ui/Документ.ЗаказПокупателя") assert response.status_code == 200 payload = response.json() assert payload["object"]["name"] == "ЗаказПокупателя" assert payload["forms"][0]["commands"][0]["name"] == "Провести" assert next(iter(payload["forms"][0]["command_handlers"].values()))["name"] == "ПровестиКоманда" def test_neo4j_object_attributes_endpoint_contract(monkeypatch): async def fake_query(project_id: str, object_name: str): assert project_id == "demo" assert object_name == "Документ.ЗаказПокупателя" return [ { "lineage_id": "lineage.attribute.counterparty", "kind": "ATTRIBUTE", "name": "Контрагент", "qualified_name": "Документ.ЗаказПокупателя.Контрагент", } ] monkeypatch.setattr(main, "_neo4j_object_attributes_query", fake_query) client = TestClient(app) response = client.get("/projects/demo/graph/neo4j/objects/attributes/Документ.ЗаказПокупателя") assert response.status_code == 200 assert response.json()["results"][0]["name"] == "Контрагент" def test_neo4j_object_schema_endpoint_contract(monkeypatch): async def fake_query(project_id: str, object_name: str): assert project_id == "demo" assert object_name == "Документ.ЗаказПокупателя" return { "object": { "lineage_id": "lineage.document.order", "kind": "DOCUMENT", "name": "ЗаказПокупателя", "qualified_name": "Документ.ЗаказПокупателя", }, "attributes": [ { "lineage_id": "lineage.attribute.counterparty", "kind": "ATTRIBUTE", "name": "Контрагент", "qualified_name": "Документ.ЗаказПокупателя.Контрагент", } ], "tabular_sections": [ { "tabular_section": { "lineage_id": "lineage.tabular.goods", "kind": "TABULAR_SECTION", "name": "Товары", "qualified_name": "Документ.ЗаказПокупателя.Товары", }, "columns": [ { "lineage_id": "lineage.attribute.item", "kind": "ATTRIBUTE", "name": "Номенклатура", "qualified_name": "Документ.ЗаказПокупателя.Товары.Номенклатура", } ], } ], } monkeypatch.setattr(main, "_neo4j_object_schema_query", fake_query) client = TestClient(app) response = client.get("/projects/demo/graph/neo4j/objects/schema/Документ.ЗаказПокупателя") assert response.status_code == 200 payload = response.json() assert payload["object"]["name"] == "ЗаказПокупателя" assert payload["attributes"][0]["name"] == "Контрагент" assert payload["tabular_sections"][0]["columns"][0]["name"] == "Номенклатура" def test_neo4j_object_tabular_sections_endpoint_contract(monkeypatch): async def fake_query(project_id: str, object_name: str): assert project_id == "demo" assert object_name == "Документ.ЗаказПокупателя" return [ { "lineage_id": "lineage.tabular.goods", "kind": "TABULAR_SECTION", "name": "Товары", "qualified_name": "Документ.ЗаказПокупателя.Товары", } ] monkeypatch.setattr(main, "_neo4j_object_tabular_sections_query", fake_query) client = TestClient(app) response = client.get("/projects/demo/graph/neo4j/objects/tabular-sections/Документ.ЗаказПокупателя") assert response.status_code == 200 assert response.json()["results"][0]["name"] == "Товары" def test_neo4j_object_tabular_section_columns_endpoint_contract(monkeypatch): async def fake_query(project_id: str, object_name: str): assert project_id == "demo" assert object_name == "Документ.ЗаказПокупателя" return [ { "tabular_section": { "lineage_id": "lineage.tabular.goods", "kind": "TABULAR_SECTION", "name": "Товары", "qualified_name": "Документ.ЗаказПокупателя.Товары", }, "columns": [ { "lineage_id": "lineage.attribute.item", "kind": "ATTRIBUTE", "name": "Номенклатура", "qualified_name": "Документ.ЗаказПокупателя.Товары.Номенклатура", } ], } ] monkeypatch.setattr(main, "_neo4j_object_tabular_section_columns_query", fake_query) client = TestClient(app) response = client.get("/projects/demo/graph/neo4j/objects/tabular-sections/Документ.ЗаказПокупателя/columns") assert response.status_code == 200 assert response.json()[0]["tabular_section"]["name"] == "Товары" assert response.json()[0]["columns"][0]["name"] == "Номенклатура" def test_neo4j_access_endpoints_contract(monkeypatch): async def fake_object_access(project_id: str, object_name: str): assert project_id == "demo" assert object_name == "Документ.ЗаказПокупателя" return { "object": { "lineage_id": "lineage.document.order", "kind": "DOCUMENT", "name": "ЗаказПокупателя", "qualified_name": "Документ.ЗаказПокупателя", }, "grants": [ { "role": { "lineage_id": "lineage.role.manager", "kind": "ROLE", "name": "Менеджер", "qualified_name": "Роль.Менеджер", }, "permissions": {"read": "true", "write": "true"}, } ], } async def fake_role_access(project_id: str, role_name: str): assert project_id == "demo" assert role_name == "Роль.Менеджер" return { "role": { "lineage_id": "lineage.role.manager", "kind": "ROLE", "name": "Менеджер", "qualified_name": "Роль.Менеджер", }, "objects": [ { "lineage_id": "lineage.document.order", "kind": "DOCUMENT", "name": "ЗаказПокупателя", "qualified_name": "Документ.ЗаказПокупателя", } ], "grants": [ { "object": { "lineage_id": "lineage.document.order", "kind": "DOCUMENT", "name": "ЗаказПокупателя", "qualified_name": "Документ.ЗаказПокупателя", }, "permissions": {"post": "true"}, } ], } monkeypatch.setattr(main, "_neo4j_object_access_query", fake_object_access) monkeypatch.setattr(main, "_neo4j_role_access_query", fake_role_access) client = TestClient(app) object_response = client.get( "/projects/demo/graph/neo4j/access/objects/Документ.ЗаказПокупателя/roles" ) role_response = client.get("/projects/demo/graph/neo4j/access/roles/Роль.Менеджер/objects") assert object_response.status_code == 200 assert object_response.json()["grants"][0]["role"]["name"] == "Менеджер" assert role_response.status_code == 200 assert role_response.json()["objects"][0]["name"] == "ЗаказПокупателя"