from __future__ import annotations import csv from datetime import datetime import io import json from pathlib import Path import tempfile import types import unittest from unittest.mock import patch import zipfile import main from openpyxl import Workbook COMPLETE_WINDOW = "2026073110001100" INCOMPLETE_WINDOW = "2026073111001200" class PipelineTest(unittest.TestCase): def test_latest_complete_window_is_converted_and_merged(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) / "source" output = Path(temp) / "output" for source_type in main.EXPECTED_TYPES: create_archive(root, source_type, COMPLETE_WINDOW) create_archive(root, main.EXPECTED_TYPES[0], INCOMPLETE_WINDOW) create_cell_data_sources(root) store = RecordingStore() result = main.process(main.LocalSource(root), output, lookback_days=3, store=store) self.assertEqual(result.name, COMPLETE_WINDOW) converted = sorted((result / "converted").glob("*.csv")) self.assertEqual(len(converted), 7) with (result / f"interference_summary_{COMPLETE_WINDOW}.csv").open(encoding="utf-8-sig", newline="") as file: rows = list(csv.DictReader(file)) self.assertEqual(len(rows), 7) self.assertEqual( list(rows[0]), ["hour_start", "hour_end", "cgi", "cell_name", "interference_dbm", "longitude", "latitude"], ) rows_by_type = {row["cell_name"].removesuffix("-小区"): row for row in rows} self.assertEqual(rows_by_type["5G干扰监控"]["cgi"], "460-00-200-1") self.assertEqual(rows_by_type["700M干扰监控"]["cgi"], "460-00-200-1") for source_type in set(main.EXPECTED_TYPES) - {"5G干扰监控", "700M干扰监控"}: self.assertEqual(rows_by_type[source_type]["cgi"], "460-00-100-1") self.assertTrue(all(row["interference_dbm"] == "-100.5" for row in rows)) self.assertEqual(rows_by_type["5G干扰监控"]["longitude"], "113.123456") self.assertEqual(rows_by_type["5G干扰监控"]["latitude"], "22.654321") self.assertTrue(all(not row["longitude"] and not row["latitude"] for row in rows if row["cgi"] == "460-00-100-1")) manifest = json.loads((result / "manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["source_file_count"], 7) self.assertEqual(manifest["cell_data_file_count"], 5) self.assertEqual(manifest["summary_rows"], 7) self.assertEqual(manifest["coordinate_matched_rows"], 2) self.assertEqual(manifest["coordinate_unmatched_rows"], 5) self.assertEqual(len(manifest["warnings"]), 1) self.assertIn(INCOMPLETE_WINDOW, manifest["warnings"][0]) self.assertNotIn("source_types", manifest) self.assertTrue(all("source_type" not in item and "source_path" not in item for item in manifest["files"])) self.assertEqual(manifest["database"]["metric_time"], "2026-07-31 10:00:00") self.assertEqual(len(store.rows), 7) def test_requested_incomplete_window_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) / "source" create_archive(root, main.EXPECTED_TYPES[0], INCOMPLETE_WINDOW) with self.assertRaisesRegex(main.ProcessingError, "Requested window is incomplete"): main.process(main.LocalSource(root), Path(temp) / "output", 3, INCOMPLETE_WINDOW) def test_schema_change_is_rejected(self) -> None: with tempfile.TemporaryDirectory() as temp: root = Path(temp) / "source" for source_type in main.EXPECTED_TYPES: create_archive(root, source_type, COMPLETE_WINDOW, bad_header=source_type == main.EXPECTED_TYPES[0]) with self.assertRaisesRegex(main.ProcessingError, "Unexpected Sheet0 header"): main.process(main.LocalSource(root), Path(temp) / "output", 3) def test_cross_midnight_window(self) -> None: self.assertEqual( main.window_bounds("2026073023000000"), ("2026-07-30 23:00:00", "2026-07-31 00:00:00"), ) def test_latest_dated_cell_data_file_is_selected(self) -> None: entries = [ {"name": "江门5G小区信息表20260727.xlsx", "path": "/old.xlsx", "is_dir": False}, {"name": "江门5G小区信息表20260728.xlsx", "path": "/latest.xlsx", "is_dir": False}, {"name": "说明.txt", "path": "/说明.txt", "is_dir": False}, ] selected = main.select_latest_cell_data_file(entries, "/cell-data") self.assertEqual(selected["path"], "/latest.xlsx") def test_cell_data_schema_change_is_rejected(self) -> None: raw = create_cell_data_workbook(include_latitude=False) with self.assertRaisesRegex(main.ProcessingError, "missing columns.*纬度"): main.parse_cell_data_workbook(raw, "bad.xlsx") def test_conflicting_cell_data_coordinates_are_rejected(self) -> None: first = create_cell_data_workbook(longitude=113.1) second = create_cell_data_workbook(longitude=113.2) with self.assertRaisesRegex(main.ProcessingError, "Conflicting CellData coordinates"): main.load_cell_coordinates([("first.xlsx", first), ("second.xlsx", second)]) def test_mysql_store_replaces_same_hour_and_deletes_other_hours(self) -> None: connection = FakeConnection(datetime(2026, 7, 31, 10), same_hour_rows=7, old_rows=20) store = main.MySQLSummaryStore(connect_factory=lambda: connection) result = store.replace_latest([database_row()]) self.assertEqual(result["metric_time"], "2026-07-31 10:00:00") self.assertEqual(result["inserted_rows"], 1) self.assertEqual(result["refreshed_rows"], 7) self.assertEqual(result["old_rows_deleted"], 20) self.assertEqual(connection.cursor_instance.inserted[0][0], datetime(2026, 7, 31, 10)) self.assertIsNone(connection.cursor_instance.inserted[0][4]) self.assertIsNone(connection.cursor_instance.inserted[0][5]) self.assertTrue(connection.committed) self.assertFalse(connection.rolled_back) self.assertTrue(connection.closed) self.assertNotIn("source_type", " ".join(connection.cursor_instance.queries)) self.assertNotIn("source_path", " ".join(connection.cursor_instance.queries)) self.assertTrue(any("ADD COLUMN `longitude`" in query for query in connection.cursor_instance.queries)) self.assertTrue(any("ADD COLUMN `latitude`" in query for query in connection.cursor_instance.queries)) def test_mysql_store_refuses_to_replace_a_newer_hour(self) -> None: connection = FakeConnection(datetime(2026, 7, 31, 11)) store = main.MySQLSummaryStore(connect_factory=lambda: connection) with self.assertRaisesRegex(main.ProcessingError, "already contains newer metric time"): store.replace_latest([database_row()]) self.assertFalse(connection.committed) self.assertTrue(connection.rolled_back) self.assertEqual(connection.cursor_instance.inserted, []) self.assertTrue(connection.closed) def test_mysql_store_creates_and_connects_to_own_database(self) -> None: bootstrap = BootstrapConnection() target = object() calls: list[dict[str, object]] = [] def connect(**kwargs: object) -> object: calls.append(kwargs) return bootstrap if len(calls) == 1 else target with patch.dict("sys.modules", {"pymysql": types.SimpleNamespace(connect=connect)}): connection = main.MySQLSummaryStore()._connect() self.assertIs(connection, target) self.assertNotIn("database", calls[0]) self.assertEqual(calls[1]["database"], "interference_etl") self.assertIn("CREATE DATABASE IF NOT EXISTS `interference_etl`", bootstrap.cursor_instance.executed[0]) self.assertTrue(bootstrap.cursor_instance.closed) self.assertTrue(bootstrap.closed) class RecordingStore: def __init__(self) -> None: self.rows: list[dict[str, str]] = [] def replace_latest(self, rows: list[dict[str, str]]) -> dict[str, object]: self.rows = list(rows) return { "enabled": True, "table": main.DATABASE_TABLE, "metric_time": rows[0]["hour_start"], "inserted_rows": len(rows), "refreshed_rows": 0, "old_rows_deleted": 0, } class FakeCursor: def __init__(self, latest_time: datetime | None, same_hour_rows: int, old_rows: int) -> None: self.latest_time = latest_time self.same_hour_rows = same_hour_rows self.old_rows = old_rows self.rowcount = 0 self.inserted: list[tuple[object, ...]] = [] self.queries: list[str] = [] self.closed = False def execute(self, query: str, params: tuple[object, ...] | None = None) -> None: normalized = " ".join(query.split()) self.queries.append(normalized) if normalized.startswith("DELETE") and "metric_time = %s" in normalized: self.rowcount = self.same_hour_rows elif normalized.startswith("DELETE") and "metric_time <> %s" in normalized: self.rowcount = self.old_rows def executemany(self, query: str, values: list[tuple[object, ...]]) -> None: del query self.inserted = list(values) self.rowcount = len(values) def fetchone(self) -> tuple[datetime | None]: return (self.latest_time,) def fetchall(self) -> list[tuple[str]]: return [(name,) for name in ("metric_time", "cgi", "cell_name", "interference_dbm")] def close(self) -> None: self.closed = True class FakeConnection: def __init__(self, latest_time: datetime | None, same_hour_rows: int = 0, old_rows: int = 0) -> None: self.cursor_instance = FakeCursor(latest_time, same_hour_rows, old_rows) self.committed = False self.rolled_back = False self.closed = False def cursor(self) -> FakeCursor: return self.cursor_instance def commit(self) -> None: self.committed = True def rollback(self) -> None: self.rolled_back = True def close(self) -> None: self.closed = True class BootstrapCursor: def __init__(self) -> None: self.executed: list[str] = [] self.closed = False def execute(self, query: str) -> None: self.executed.append(" ".join(query.split())) def close(self) -> None: self.closed = True class BootstrapConnection: def __init__(self) -> None: self.cursor_instance = BootstrapCursor() self.closed = False def cursor(self) -> BootstrapCursor: return self.cursor_instance def close(self) -> None: self.closed = True def database_row() -> dict[str, str]: return { "hour_start": "2026-07-31 10:00:00", "hour_end": "2026-07-31 11:00:00", "cgi": "460-00-200-1", "cell_name": "测试小区", "interference_dbm": "-100.5", "longitude": "", "latitude": "", } def create_archive(root: Path, source_type: str, window: str, bad_header: bool = False) -> None: date_dir = root / f"{window[:4]}-{window[4:6]}-{window[6:8]}" date_dir.mkdir(parents=True, exist_ok=True) filename = f"{source_type}_LWP_每小时_过滤110_{window}" workbook = Workbook() sheet = workbook.active sheet.title = "Sheet0" header = list(main.EXPECTED_HEADERS[source_type]) if bad_header: header[-1] = "unexpected" sheet.append(header) sheet.append(mock_row(source_type, window)) metadata = workbook.create_sheet("指标(计数器)") metadata.append(["指标或计数器", "指标或计数器描述", "指标公式", "指标或计数器状态"]) content = io.BytesIO() workbook.save(content) workbook.close() with zipfile.ZipFile(date_dir / f"{filename}.zip", "w", zipfile.ZIP_DEFLATED) as archive: archive.writestr(f"{filename}.xlsx", content.getvalue()) def mock_row(source_type: str, window: str) -> list[object]: header = main.EXPECTED_HEADERS[source_type] values: dict[str, object] = {column: "mock" for column in header} values.update( { "开始时间": datetime.strptime(window[:12], "%Y%m%d%H%M"), "粒度": "1 小时", "eNodeBId": 100, "eNodeBID": 100, "gNBId": 200, "gNBplmn": "460-00", "cellId": 1, "小区ID": 1, "masterOperatorId": "unused-source-value", "E-UTRAN FDD小区名称": f"{source_type}-小区", "E-UTRAN TDD小区名称": f"{source_type}-小区", "CU小区配置名称": f"{source_type}-小区", "小区名称": f"{source_type}-小区", "载波平均噪声干扰(dBm)": -100.5, "小区上行平均干扰电平(dBm)": -100.5, } ) return [values[column] for column in header] def create_cell_data_sources(root: Path) -> None: raw = create_cell_data_workbook() for index, configured_directory in enumerate(main.CELL_DATA_DIRECTORIES, start=1): directory = root / Path(configured_directory.lstrip("/")) directory.mkdir(parents=True, exist_ok=True) (directory / f"江门小区信息表{index}-20260728.xlsx").write_bytes(raw) def create_cell_data_workbook(longitude: float = 113.123456, include_latitude: bool = True) -> bytes: workbook = Workbook() sheet = workbook.active sheet.title = "小区信息表" header = ["小区名称", "eNB/gNB", "CI", "经度"] if include_latitude: header.append("纬度") sheet.append(header) row: list[object] = ["测试小区", 200, 1, longitude] if include_latitude: row.append(22.654321) sheet.append(row) content = io.BytesIO() workbook.save(content) workbook.close() return content.getvalue() if __name__ == "__main__": unittest.main()