136 lines
4.0 KiB
Python
136 lines
4.0 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from app.pipeline.column_registry import (
|
||
COLUMNS,
|
||
MAPPING_DATA_FILE,
|
||
AmbiguousColumnNameError,
|
||
ColumnRegistry,
|
||
ColumnSchemaError,
|
||
)
|
||
|
||
|
||
def test_registry_keeps_stable_file_name_column_id() -> None:
|
||
assert COLUMNS.index_of("legacy_001") == 1
|
||
assert COLUMNS.meta(1).name == "Имя XML файла"
|
||
|
||
|
||
def test_registry_covers_all_report_columns_without_gaps() -> None:
|
||
indexes = tuple(column.index for column in COLUMNS.ordered_columns())
|
||
|
||
assert indexes == tuple(range(1, 205))
|
||
assert all(column.block for column in COLUMNS.ordered_columns())
|
||
assert all(column.report_group for column in COLUMNS.ordered_columns())
|
||
|
||
|
||
def test_registry_assigns_card_columns_missing_from_original_plan() -> None:
|
||
assert COLUMNS.block("foreign_card") == tuple(range(97, 107))
|
||
|
||
|
||
def test_registry_explicitly_preserves_structured_lookup_compatibility() -> None:
|
||
assert COLUMNS.meta(180).allow_short_lookup is True
|
||
assert COLUMNS.meta(153).allow_short_lookup is False
|
||
|
||
|
||
def test_registry_rejects_ambiguous_display_name_lookup() -> None:
|
||
assert COLUMNS.indices_named("ОКАТО") == (4, 171)
|
||
|
||
with pytest.raises(AmbiguousColumnNameError):
|
||
COLUMNS.index_named("ОКАТО")
|
||
|
||
|
||
def test_registry_report_groups_are_contiguous_and_cover_schema() -> None:
|
||
covered_indexes: list[int] = []
|
||
|
||
for group in COLUMNS.report_groups():
|
||
assert group.indexes == tuple(range(group.start_index, group.end_index + 1))
|
||
covered_indexes.extend(group.indexes)
|
||
|
||
assert covered_indexes == list(range(1, 205))
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("field", "invalid_value"),
|
||
(
|
||
("index", True),
|
||
("column_id", None),
|
||
("column_name", ["Имя"]),
|
||
("source_scope", {"scope": "any"}),
|
||
("structured_order", False),
|
||
),
|
||
)
|
||
def test_registry_rejects_invalid_scalar_types(
|
||
tmp_path: Path, field: str, invalid_value: object
|
||
) -> None:
|
||
data = _mapping_data()
|
||
data[0][field] = invalid_value
|
||
|
||
with pytest.raises(ColumnSchemaError):
|
||
ColumnRegistry.load(_write_mapping(tmp_path, data))
|
||
|
||
|
||
def test_registry_rejects_unknown_report_group(tmp_path: Path) -> None:
|
||
data = _mapping_data()
|
||
data[0]["report_group"] = "unknown_group"
|
||
|
||
with pytest.raises(ColumnSchemaError):
|
||
ColumnRegistry.load(_write_mapping(tmp_path, data))
|
||
|
||
|
||
def test_registry_rejects_unknown_business_block(tmp_path: Path) -> None:
|
||
data = _mapping_data()
|
||
data[0]["block"] = "unknown_block"
|
||
|
||
with pytest.raises(ColumnSchemaError):
|
||
ColumnRegistry.load(_write_mapping(tmp_path, data))
|
||
|
||
|
||
def test_registry_rejects_duplicate_column_ids_and_index_gaps(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
data = _mapping_data()
|
||
data[1]["column_id"] = data[0]["column_id"]
|
||
data[1]["index"] = 3
|
||
|
||
with pytest.raises(ColumnSchemaError):
|
||
ColumnRegistry.load(_write_mapping(tmp_path, data))
|
||
|
||
|
||
def test_registry_rejects_structured_group_without_aggregate(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
data = _mapping_data()
|
||
aggregate = next(item for item in data if item["index"] == 50)
|
||
aggregate["structured_role"] = "component"
|
||
aggregate["structured_order"] = 10
|
||
aggregate["structured_value"] = False
|
||
|
||
with pytest.raises(ColumnSchemaError):
|
||
ColumnRegistry.load(_write_mapping(tmp_path, data))
|
||
|
||
|
||
def test_registry_rejects_duplicate_structured_component_order(
|
||
tmp_path: Path,
|
||
) -> None:
|
||
data = _mapping_data()
|
||
first = next(item for item in data if item["index"] == 51)
|
||
second = next(item for item in data if item["index"] == 52)
|
||
second["structured_order"] = first["structured_order"]
|
||
|
||
with pytest.raises(ColumnSchemaError):
|
||
ColumnRegistry.load(_write_mapping(tmp_path, data))
|
||
|
||
|
||
def _mapping_data() -> list[dict[str, object]]:
|
||
return json.loads(MAPPING_DATA_FILE.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _write_mapping(tmp_path: Path, data: list[dict[str, object]]) -> Path:
|
||
path = tmp_path / "mapping_table.json"
|
||
path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
||
return path
|