from __future__ import annotations import csv from datetime import datetime import io import json from pathlib import Path import tempfile import unittest 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", "network_type", "cgi", "cell_name", "interference_dbm", "longitude", "latitude", "azimuth", ], ) rows_by_type = {row["cell_name"].removesuffix("-小区"): row for row in rows} self.assertEqual(rows_by_type["5G干扰监控"]["network_type"], "2.6G") self.assertEqual(rows_by_type["700M干扰监控"]["network_type"], "700M") self.assertEqual(rows_by_type["SDR_FDD干扰监控"]["network_type"], "4G") 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.assertEqual(rows_by_type["5G干扰监控"]["azimuth"], "30") self.assertTrue(all(not row["longitude"] and not row["latitude"] for row in rows if row["cgi"] == "460-00-100-1")) self.assertTrue(all(row["azimuth"] == "0" 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["threshold_filtered_rows"], 0) 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_interference_thresholds_remove_only_lower_values(self) -> None: for network_type, threshold in (("2.6G", "-107"), ("700M", "-110"), ("4G", "-110")): row = database_row() row["network_type"] = network_type row["interference_dbm"] = threshold self.assertTrue(main.passes_interference_threshold(row)) row["interference_dbm"] = str(float(threshold) - 0.1) self.assertFalse(main.passes_interference_threshold(row)) with tempfile.TemporaryDirectory() as temp: root = Path(temp) / "source" for source_type in main.EXPECTED_TYPES: interference = -107.1 if source_type == main.EXPECTED_TYPES[0] else -100.5 create_archive(root, source_type, COMPLETE_WINDOW, interference_dbm=interference) create_cell_data_sources(root) result = main.process(main.LocalSource(root), Path(temp) / "output", 3) manifest = json.loads((result / "manifest.json").read_text(encoding="utf-8")) self.assertEqual(manifest["summary_rows"], 6) self.assertEqual(manifest["threshold_filtered_rows"], 1) 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_metadata_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 metadata"): main.load_cell_metadata([("first.xlsx", first), ("second.xlsx", second)]) def test_empty_cell_data_azimuth_defaults_to_zero(self) -> None: metadata = main.parse_cell_data_workbook(create_cell_data_workbook(azimuth=None), "cell-data.xlsx") self.assertEqual(metadata["460-00-200-1"], ("113.123456", "22.654321", "0")) def test_api_store_replaces_same_hour_and_deletes_other_hours(self) -> None: client = FakeApiClient("2026-07-31T10:00:00") store = main.ApiSummaryStore(client, "db_share_mysql") 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) queries = [payload["sql"] for endpoint, payload in client.posts if endpoint.endswith("/query")] self.assertTrue(any("CREATE DATABASE IF NOT EXISTS `interference_etl`" in query for query in queries)) self.assertTrue(any("ADD COLUMN `network_type`" in query for query in queries)) self.assertTrue(any("ADD COLUMN `longitude`" in query for query in queries)) self.assertTrue(any("ADD COLUMN `latitude`" in query for query in queries)) self.assertTrue(any("ADD COLUMN `azimuth`" in query for query in queries)) script = next(payload for endpoint, payload in client.posts if endpoint.endswith("/run-script")) self.assertTrue(script["single_session"]) self.assertIn("START TRANSACTION", script["content"]) self.assertIn("NULL, NULL, 0", script["content"]) self.assertIn("CONVERT(0x", script["content"]) self.assertNotIn("source_type", script["content"]) self.assertNotIn("source_path", script["content"]) def test_api_store_refuses_to_replace_a_newer_hour(self) -> None: client = FakeApiClient("2026-07-31T11:00:00") store = main.ApiSummaryStore(client, "db_share_mysql") with self.assertRaisesRegex(main.ProcessingError, "already contains newer metric time"): store.replace_latest([database_row()]) self.assertFalse(any(endpoint.endswith("/run-script") for endpoint, _ in client.posts)) def test_api_store_propagates_script_failure(self) -> None: client = FakeApiClient(None, fail_script=True) store = main.ApiSummaryStore(client, "db_share_mysql") with self.assertRaisesRegex(main.ProcessingError, "statement 3: insert failed"): store.replace_latest([database_row()]) def test_api_store_rejects_invalid_decimal(self) -> None: client = FakeApiClient(None) store = main.ApiSummaryStore(client, "db_share_mysql") row = database_row() row["interference_dbm"] = "not-a-number" with self.assertRaisesRegex(main.ProcessingError, "Invalid decimal value for interference_dbm"): store.replace_latest([row]) 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 FakeApiClient: def __init__(self, latest_time: str | None, fail_script: bool = False) -> None: self.latest_time = latest_time self.fail_script = fail_script self.posts: list[tuple[str, dict[str, object]]] = [] def get_json(self, endpoint: str, query: dict[str, object] | None = None, timeout: int = 30) -> object: del endpoint, query, timeout return [ {"name": "metric_time"}, {"name": "cgi"}, {"name": "cell_name"}, {"name": "interference_dbm"}, ] def post_json(self, endpoint: str, payload: dict[str, object], timeout: int = 30) -> object: del timeout self.posts.append((endpoint, payload)) if endpoint.endswith("/query"): if str(payload["sql"]).lstrip().startswith("SELECT MAX"): return {"rows": [{"latest_time": self.latest_time}]} return {"affected_rows": 0} if self.fail_script: return { "results": [ {"index": 1, "ok": True, "affected_rows": 0}, {"index": 2, "ok": True, "affected_rows": 7}, {"index": 3, "ok": False, "message": "insert failed", "affected_rows": 0}, ], "stopped": True, } return { "results": [ {"index": 1, "ok": True, "affected_rows": 0}, {"index": 2, "ok": True, "affected_rows": 7}, {"index": 3, "ok": True, "affected_rows": 1}, {"index": 4, "ok": True, "affected_rows": 20}, {"index": 5, "ok": True, "affected_rows": 0}, ], "stopped": False, } def database_row() -> dict[str, str]: return { "hour_start": "2026-07-31 10:00:00", "hour_end": "2026-07-31 11:00:00", "network_type": "2.6G", "cgi": "460-00-200-1", "cell_name": "测试小区", "interference_dbm": "-100.5", "longitude": "", "latitude": "", "azimuth": "0", } def create_archive( root: Path, source_type: str, window: str, bad_header: bool = False, interference_dbm: float = -100.5, ) -> 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, interference_dbm)) 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, interference_dbm: float = -100.5) -> 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)": interference_dbm, "小区上行平均干扰电平(dBm)": interference_dbm, } ) 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, azimuth: float | None = 30, ) -> bytes: workbook = Workbook() sheet = workbook.active sheet.title = "小区信息表" header = ["小区名称", "eNB/gNB", "CI", "经度"] if include_latitude: header.append("纬度") header.append("方向角") sheet.append(header) row: list[object] = ["测试小区", 200, 1, longitude] if include_latitude: row.append(22.654321) row.append(azimuth) sheet.append(row) content = io.BytesIO() workbook.save(content) workbook.close() return content.getvalue() if __name__ == "__main__": unittest.main()