from pathlib import Path import subprocess import semantic_kernel from semantic_kernel import index_project, parse_bsl_module, parse_bsl_module_from_rust_json from sir import EdgeKind, NodeKind def test_parse_bsl_module_supports_english_1c_syntax() -> None: source = Path("tests/golden/english_module.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert [routine.name for routine in routines] == ["Posting", "CheckStock"] assert routines[1].is_function assert routines[0].calls == (("CheckStock", 2),) assert routines[0].writes[0].target == "StockBalance" assert routines[1].queries[0].tables == ("AccumulationRegister.StockBalance",) def test_parse_bsl_module_supports_inline_query_assignment_with_pipes() -> None: source = Path("tests/golden/query_inline_pipes.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert len(routines) == 1 assert routines[0].queries[0].tables == ("Справочник.Номенклатура",) assert routines[0].queries[0].text.startswith("ВЫБРАТЬ") assert "|ИЗ" not in routines[0].queries[0].text def test_parse_bsl_module_supports_from_and_table_on_same_line() -> None: source = Path("tests/golden/query_inline_from.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert len(routines) == 1 assert routines[0].queries[0].tables == ("Справочник.Контрагенты",) def test_parse_bsl_module_extracts_assignment_function_calls() -> None: source = Path("tests/golden/assignment_call.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert routines[0].calls == (("ПроверитьОстатки", 2),) def test_parse_bsl_module_extracts_condition_function_calls() -> None: source = Path("tests/golden/condition_call.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert routines[0].calls == (("ПроверитьОстатки", 2),) def test_parse_bsl_module_extracts_join_tables() -> None: source = Path("tests/golden/query_join_tables.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert routines[0].queries[0].tables == ( "Документ.ЗаказПокупателя", "Справочник.Контрагенты", ) def test_parse_bsl_module_extracts_object_write_targets() -> None: source = Path("tests/golden/object_write.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert routines[0].writes[0].target == "Справочник.Номенклатура" assert routines[0].writes[0].write_type == "OBJECT_WRITE" assert routines[1].writes[0].target == "Документ.CustomerOrder" assert routines[1].writes[0].write_type == "OBJECT_WRITE" def test_parse_bsl_module_extracts_recordset_write_targets() -> None: source = Path("tests/golden/recordset_write.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert routines[0].writes[0].target == "РегистрСведений.Цены" assert routines[0].writes[0].write_type == "REGISTER_WRITE" assert routines[1].writes[0].target == "РегистрНакопления.StockBalance" assert routines[1].writes[0].write_type == "REGISTER_WRITE" def test_parse_bsl_module_preserves_export_flag() -> None: source = Path("tests/golden/common_module_export.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert [routine.export for routine in routines] == [True, True] def test_parse_bsl_module_extracts_calls_and_writes_inside_control_flow() -> None: source = Path("tests/golden/control_flow_calls.bsl").read_text(encoding="utf-8") routines = parse_bsl_module(source) assert routines[0].calls == (("ПроверитьСтроку", 3), ("СообщитьОбОшибке", 9)) assert routines[0].writes[0].target == "ОстаткиТоваров" def test_index_project_links_english_register_write(tmp_path: Path) -> None: module = tmp_path / "english_module.bsl" module.write_text(Path("tests/golden/english_module.bsl").read_text(encoding="utf-8"), encoding="utf-8") snapshot = index_project(tmp_path, project_id="english") register = next(node for node in snapshot.nodes if node.kind == NodeKind.REGISTER) assert register.name == "StockBalance" assert any(edge.kind == EdgeKind.WRITES and edge.target_lineage == register.lineage_id for edge in snapshot.edges) def test_index_project_links_object_write_to_metadata_node(tmp_path: Path) -> None: module = tmp_path / "object_write.bsl" module.write_text(Path("tests/golden/object_write.bsl").read_text(encoding="utf-8"), encoding="utf-8") snapshot = index_project(tmp_path, project_id="object-write") catalog = next(node for node in snapshot.nodes if node.kind == NodeKind.CATALOG) document = next(node for node in snapshot.nodes if node.kind == NodeKind.DOCUMENT) assert catalog.qualified_name == "Справочник.Номенклатура" assert document.qualified_name == "Документ.CustomerOrder" assert any(edge.kind == EdgeKind.WRITES and edge.target_lineage == catalog.lineage_id for edge in snapshot.edges) assert any(edge.kind == EdgeKind.WRITES and edge.target_lineage == document.lineage_id for edge in snapshot.edges) def test_index_project_links_recordset_write_to_register_node(tmp_path: Path) -> None: module = tmp_path / "recordset_write.bsl" module.write_text(Path("tests/golden/recordset_write.bsl").read_text(encoding="utf-8"), encoding="utf-8") snapshot = index_project(tmp_path, project_id="recordset-write") registers = {node.qualified_name: node for node in snapshot.nodes if node.kind == NodeKind.REGISTER} assert "РегистрСведений.Цены" in registers assert "РегистрНакопления.StockBalance" in registers assert any( edge.kind == EdgeKind.WRITES and edge.target_lineage == registers["РегистрСведений.Цены"].lineage_id for edge in snapshot.edges ) assert any( edge.kind == EdgeKind.WRITES and edge.target_lineage == registers["РегистрНакопления.StockBalance"].lineage_id for edge in snapshot.edges ) def test_index_project_stores_routine_export_attribute(tmp_path: Path) -> None: module = tmp_path / "common_module_export.bsl" module.write_text( Path("tests/golden/common_module_export.bsl").read_text(encoding="utf-8"), encoding="utf-8", ) snapshot = index_project(tmp_path, project_id="exports") exported = { node.name for node in snapshot.nodes if node.kind in {NodeKind.PROCEDURE, NodeKind.FUNCTION} and node.attributes.get("export") } assert exported == {"ОтправитьЧек", "BuildPayload"} def test_parse_bsl_module_from_rust_json_contract() -> None: payload = { "procedures": [ { "name": "Posting", "is_function": False, "source_range": {"line_start": 1, "line_end": 4}, }, { "name": "CheckStock", "is_function": True, "source_range": {"line_start": 6, "line_end": 13}, }, ], "calls": [ { "caller": "Posting", "callee": "CheckStock", "source_range": {"line_start": 2, "line_end": 2}, } ], "queries": [ { "owner_procedure": "CheckStock", "query_text": "SELECT\nStock.Item\nFROM\nAccumulationRegister.StockBalance AS Stock", "tables": ["AccumulationRegister.StockBalance"], "source_range": {"line_start": 8, "line_end": 12}, } ], "writes": [ { "owner_procedure": "Posting", "target": "StockBalance", "write_type": "REGISTER_WRITE", "source_range": {"line_start": 3, "line_end": 3}, } ], "diagnostics": [], } routines = parse_bsl_module_from_rust_json(payload) assert [routine.name for routine in routines] == ["Posting", "CheckStock"] assert routines[0].line_end == 4 assert routines[0].calls == (("CheckStock", 2),) assert routines[0].writes[0].target == "StockBalance" assert routines[1].queries[0].tables == ("AccumulationRegister.StockBalance",) def test_index_project_can_use_rust_parser_contract(monkeypatch, tmp_path: Path) -> None: module = tmp_path / "english_module.bsl" module.write_text(Path("tests/golden/english_module.bsl").read_text(encoding="utf-8"), encoding="utf-8") rust_json = r""" { "source_path": "english_module.bsl", "procedures": [ {"name": "Posting", "export": false, "is_function": false, "parameters": [], "source_range": {"line_start": 1, "line_end": 4, "column_start": 1, "column_end": 13}}, {"name": "CheckStock", "export": false, "is_function": true, "parameters": [], "source_range": {"line_start": 6, "line_end": 13, "column_start": 1, "column_end": 12}} ], "calls": [{"caller": "Posting", "callee": "CheckStock", "source_range": {"line_start": 2, "line_end": 2, "column_start": 1, "column_end": 18}}], "queries": [{"owner_procedure": "CheckStock", "query_text": "SELECT\nStock.Item\nFROM\nAccumulationRegister.StockBalance AS Stock", "tables": ["AccumulationRegister.StockBalance"], "source_range": {"line_start": 8, "line_end": 12, "column_start": 1, "column_end": 53}}], "writes": [{"owner_procedure": "Posting", "target": "StockBalance", "write_type": "REGISTER_WRITE", "source_range": {"line_start": 3, "line_end": 3, "column_start": 1, "column_end": 36}}], "diagnostics": [] } """ def fake_run(command, check, capture_output, text, encoding): assert command[0] == "bsl-parser" assert Path(command[1]) == module return subprocess.CompletedProcess(command, 0, stdout=rust_json, stderr="") monkeypatch.setenv("SFERA_BSL_PARSER", "bsl-parser") monkeypatch.setattr(semantic_kernel.subprocess, "run", fake_run) snapshot = index_project(tmp_path, project_id="rust-contract") assert any(node.kind == NodeKind.FUNCTION and node.name == "CheckStock" for node in snapshot.nodes) assert any(edge.kind == EdgeKind.CALLS for edge in snapshot.edges) assert any(edge.kind == EdgeKind.WRITES for edge in snapshot.edges) def test_index_project_auto_discovers_rust_parser(monkeypatch, tmp_path: Path) -> None: module = tmp_path / "english_module.bsl" module.write_text(Path("tests/golden/english_module.bsl").read_text(encoding="utf-8"), encoding="utf-8") rust_json = r""" { "source_path": "english_module.bsl", "procedures": [ {"name": "Posting", "export": false, "is_function": false, "parameters": [], "source_range": {"line_start": 1, "line_end": 4, "column_start": 1, "column_end": 13}}, {"name": "CheckStock", "export": false, "is_function": true, "parameters": [], "source_range": {"line_start": 6, "line_end": 13, "column_start": 1, "column_end": 12}} ], "calls": [{"caller": "Posting", "callee": "CheckStock", "source_range": {"line_start": 2, "line_end": 2, "column_start": 1, "column_end": 18}}], "queries": [{"owner_procedure": "CheckStock", "query_text": "SELECT\nStock.Item\nFROM\nAccumulationRegister.StockBalance AS Stock", "tables": ["AccumulationRegister.StockBalance"], "source_range": {"line_start": 8, "line_end": 12, "column_start": 1, "column_end": 53}}], "writes": [{"owner_procedure": "Posting", "target": "StockBalance", "write_type": "REGISTER_WRITE", "source_range": {"line_start": 3, "line_end": 3, "column_start": 1, "column_end": 36}}], "diagnostics": [] } """ def fake_run(command, check, capture_output, text, encoding): assert command[0] == "auto-bsl-parser" assert Path(command[1]) == module return subprocess.CompletedProcess(command, 0, stdout=rust_json, stderr="") monkeypatch.delenv("SFERA_BSL_PARSER", raising=False) monkeypatch.setattr(semantic_kernel, "_auto_discovered_rust_bsl_parser", lambda source_file: "auto-bsl-parser") monkeypatch.setattr(semantic_kernel.subprocess, "run", fake_run) snapshot = index_project(tmp_path, project_id="rust-auto") assert any(node.kind == NodeKind.FUNCTION and node.name == "CheckStock" for node in snapshot.nodes) assert any(edge.kind == EdgeKind.CALLS for edge in snapshot.edges) assert any(edge.kind == EdgeKind.WRITES for edge in snapshot.edges)