108 lines
3.8 KiB
Python
108 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from server.aps_domain.kangni_intake import (
|
|
DEFAULT_ROUTE_XLSX,
|
|
_find_file,
|
|
build_flex_bundle,
|
|
build_site_payload_from_data_dir,
|
|
parse_orders_workbook,
|
|
)
|
|
from tests.external_data import external_dir
|
|
|
|
KANGNI_DATA_DIR = external_dir("KANGNI_DATA_DIR", "kangni")
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not KANGNI_DATA_DIR.exists() or not Path(DEFAULT_ROUTE_XLSX).exists(),
|
|
reason="康尼现场只读数据不在本机",
|
|
)
|
|
def test_real_kangni_10_order_import_aggregates_bom_usage_read_only():
|
|
source_files = sorted(KANGNI_DATA_DIR.glob("*.xlsx"))
|
|
assert len(source_files) == 9
|
|
source_hashes_before = {path.name: _sha256(path) for path in source_files}
|
|
world_path = Path(__file__).resolve().parents[2] / "server" / "data" / "world.json"
|
|
world_hash_before = _sha256(world_path) if world_path.exists() else None
|
|
orders_path = _find_file(KANGNI_DATA_DIR, "orders")
|
|
assert orders_path is not None
|
|
assert len(parse_orders_workbook(orders_path)) == 10
|
|
|
|
bundle = build_flex_bundle(
|
|
DEFAULT_ROUTE_XLSX,
|
|
KANGNI_DATA_DIR,
|
|
include_sibling_orders=True,
|
|
)
|
|
meta = bundle["_meta"]
|
|
assert meta["orderCount"] == 10
|
|
assert meta["routingRecordCount"] == 72
|
|
assert meta["operationCount"] == 26
|
|
assert meta["bomSourceRowCount"] == 965
|
|
assert meta["bomCount"] == 823
|
|
assert sum(
|
|
int(row.get("aggregatedRows") or 0)
|
|
for row in meta["bomMappings"].values()
|
|
) > 0
|
|
assert any("BOM 多行用量" in line for line in meta["inferences"])
|
|
|
|
assert {path.name: _sha256(path) for path in source_files} == source_hashes_before
|
|
assert (_sha256(world_path) if world_path.exists() else None) == world_hash_before
|
|
|
|
|
|
@pytest.mark.skipif(
|
|
not KANGNI_DATA_DIR.exists(),
|
|
reason="康尼现场只读数据不在本机",
|
|
)
|
|
def test_real_kangni_data_dir_aggregation_is_deterministic(tmp_path: Path):
|
|
source_files = sorted(KANGNI_DATA_DIR.glob("*.xlsx"))
|
|
assert len(source_files) == 9
|
|
source_hashes_before = {path.name: _sha256(path) for path in source_files}
|
|
|
|
upload_a = tmp_path / "upload-a"
|
|
upload_b = tmp_path / "another-temporary-root"
|
|
upload_a.mkdir()
|
|
upload_b.mkdir()
|
|
for source in source_files:
|
|
shutil.copy2(source, upload_a / source.name)
|
|
shutil.copy2(source, upload_b / source.name)
|
|
assert not list(upload_a.glob("*完整生产路线*.xlsx"))
|
|
copy_hashes_before = {
|
|
path.name: _sha256(path)
|
|
for path in sorted(upload_a.glob("*.xlsx"))
|
|
}
|
|
orders_path = _find_file(upload_a, "orders")
|
|
assert orders_path is not None
|
|
assert len(parse_orders_workbook(orders_path)) == 10
|
|
|
|
payload_a = build_site_payload_from_data_dir(upload_a)
|
|
payload_b = build_site_payload_from_data_dir(upload_b)
|
|
canonical_a = json.dumps(payload_a, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
canonical_b = json.dumps(payload_b, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
assert canonical_a == canonical_b
|
|
assert payload_a["meta"]["orderCount"] == 10
|
|
assert payload_a["meta"]["routingRecordCount"] == 72
|
|
assert payload_a["meta"]["bomSourceRowCount"] == 965
|
|
assert payload_a["meta"]["projectedBomCount"] == 823
|
|
assert str(upload_a) not in canonical_a
|
|
assert str(upload_b) not in canonical_b
|
|
assert str(KANGNI_DATA_DIR) not in canonical_a
|
|
|
|
assert {path.name: _sha256(path) for path in source_files} == source_hashes_before
|
|
assert {
|
|
path.name: _sha256(path)
|
|
for path in sorted(upload_a.glob("*.xlsx"))
|
|
} == copy_hashes_before
|