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-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 "Проверить" 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", ) client = TestClient(app) indexed = client.post("/projects/index", json={"path": str(tmp_path), "project_id": project_id}) assert indexed.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 "Контрагент" in context.text assert "Товары" 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"] == "ЗаказПокупателя"