280 lines
12 KiB
Python
280 lines
12 KiB
Python
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)
|