import io from numbers import Integral from unittest.mock import patch import pandas as pd from queries.ciq_3g_schema_loader import ( WBTS_FINAL_COLUMNS, WBTS_FINAL_DEFAULTS, WCEL_FINAL_COLUMNS, WCEL_FINAL_DEFAULTS, ) from queries.process_ciq_3g import ( _get_excel_writer_engine, apply_final_schema, build_wbts_sheet, build_wcel_sheet, generate_ciq_3g_excel, ) def _sample_ciq_df() -> pd.DataFrame: return pd.DataFrame( [ { "Sites": "0273_TBC_DIRE-HAMDALLAYE_3G", "NodeB_ID": 273, "NOM_CELLULE": "0273_TBC_DIRE-HAMDALLAYE_1_U2100", "CELLID": 2731, "SAC": 11, "LAC": 30802, "RAC": 250, "FREQUENCE": 10713, "PSCRAMBCODE": 101, "RNC_id": 2202, }, { "Sites": "0273_TBC_DIRE-HAMDALLAYE_3G", "NodeB_ID": 273, "NOM_CELLULE": "0273_TBC_DIRE-HAMDALLAYE_2_U2100", "CELLID": 2732, "SAC": 12, "LAC": 30802, "RAC": 250, "FREQUENCE": 10713, "PSCRAMBCODE": 102, "RNC_id": 2202, }, { "Sites": "0273_TBC_DIRE-HAMDALLAYE_3G", "NodeB_ID": 273, "NOM_CELLULE": "0273_TBC_DIRE-HAMDALLAYE_4_U900", "CELLID": 2734, "SAC": 14, "LAC": 30802, "RAC": 250, "FREQUENCE": 3087, "PSCRAMBCODE": 104, "RNC_id": 2202, }, ] ) def _sample_ciq_u900_only_df() -> pd.DataFrame: return pd.DataFrame( [ { "Sites": "0660_SKS_HAMADALLAYE_3G", "NodeB_ID": 660, "NOM_CELLULE": "0660_SKS_HAMADALLAYE_1_U900", "CELLID": 6601, "SAC": 11, "LAC": 30802, "RAC": 250, "FREQUENCE": 3087, "PSCRAMBCODE": 101, "RNC_id": 2202, }, { "Sites": "0660_SKS_HAMADALLAYE_3G", "NodeB_ID": 660, "NOM_CELLULE": "0660_SKS_HAMADALLAYE_2_U900", "CELLID": 6602, "SAC": 12, "LAC": 30802, "RAC": 250, "FREQUENCE": 3087, "PSCRAMBCODE": 102, "RNC_id": 2202, }, ] ) def test_apply_final_schema_wbts_keeps_core_values_and_order(): df_wbts = build_wbts_sheet(_sample_ciq_df(), year_suffix="26", bands="U9U21") final_df = apply_final_schema(df_wbts, "WBTS") assert list(final_df.columns) == WBTS_FINAL_COLUMNS assert len(final_df.columns) == 67 assert final_df.loc[0, "S"] == 273 assert final_df.loc[0, "RncId"] == 2202 assert final_df.loc[0, "WBTSId"] == 273 assert final_df.loc[0, "Name"] == "0273_TBC_DIRE-HAMDALLAYE_26_U9U21_NA" assert final_df.loc[0, "name"] == "0273_TBC_DIRE-HAMDALLAYE_3G_NA" assert final_df.loc[0, "WBTSName"] == "0273_TBC_DIRE-HAMDALLAYE_3G_NA" assert final_df.loc[0, "BTSSupportForHSPACM"] == WBTS_FINAL_DEFAULTS["BTSSupportForHSPACM"] assert final_df.loc[0, "DelayThresholdMax"] == WBTS_FINAL_DEFAULTS["DelayThresholdMax"] assert final_df.loc[0, "HSDPAUserThreshold"] == "" assert final_df.loc[0, "IPNBId"] == "" assert isinstance(final_df.loc[0, "WinLCHSDPA"], Integral) def test_apply_final_schema_wbts_uses_50000_offset_for_u900_only_sites(): df_wbts = build_wbts_sheet(_sample_ciq_u900_only_df(), year_suffix="26", bands="U9") final_df = apply_final_schema(df_wbts, "WBTS") assert final_df.loc[0, "S"] == 660 assert final_df.loc[0, "WBTSId"] == 50660 assert final_df.loc[0, "Name"] == "0660_SKS_HAMADALLAYE_26_U9_NA" assert final_df.loc[0, "WBTSName"] == "0660_SKS_HAMADALLAYE_3G_NA" def test_build_wbts_sheet_uses_profile_specific_bands_for_name(): df_wbts = build_wbts_sheet( _sample_ciq_df(), year_suffix="26", bands="G9G18U9U21L8L18L26", profile_definitions={ "densif": "G9G18U9U21L7L8L18L26", "rural": "G9U9L7L8", }, site_profile_mapping={"273": "densif"}, ) assert df_wbts.loc[0, "Name"] == "0273_TBC_DIRE-HAMDALLAYE_26_G9G18U9U21L7L8L18L26_NA" def test_apply_final_schema_wcel_keeps_core_values_and_does_not_override_existing_columns(): df_wcel = build_wcel_sheet(_sample_ciq_df()) df_wcel["PtxTarget"] = "custom" final_df = apply_final_schema(df_wcel, "WCEL") assert list(final_df.columns) == WCEL_FINAL_COLUMNS assert len(final_df.columns) == 516 assert list(final_df["SectorID"]) == [1, 2, 1] assert list(final_df["LcrId"]) == [1, 2, 10] assert list(final_df["Band"]) == ["U2100 (U21)", "U2100 (U21)", "U900 (U9)"] assert list(final_df["UARFCN"]) == [10713, 10713, 3087] assert list(final_df["PriScrCode"]) == [101, 102, 104] assert list(final_df["PtxTarget"]) == ["custom", "custom", "custom"] assert final_df.loc[0, "AdminCellState"] == WCEL_FINAL_DEFAULTS["AdminCellState"] assert final_df.loc[0, "CellAdditionalInfo"] == "" assert final_df.loc[0, "AllowedPreambleSignatures"] == "" assert isinstance(final_df.loc[0, "PrxTarget"], Integral) def test_build_wcel_uses_same_wbtsid_for_u900_only_sites(): df_wcel = build_wcel_sheet(_sample_ciq_u900_only_df()) assert list(df_wcel["Site"]) == [660, 660] assert list(df_wcel["WBTSId"]) == [50660, 50660] assert list(df_wcel["Band"]) == ["U900 (U9)", "U900 (U9)"] assert list(df_wcel["LcrId"]) == [10, 11] def test_get_excel_writer_engine_prefers_xlsxwriter_then_openpyxl(): with patch("queries.process_ciq_3g.importlib.util.find_spec") as mock_find_spec: mock_find_spec.side_effect = lambda name: object() if name == "xlsxwriter" else None assert _get_excel_writer_engine() == "xlsxwriter" with patch("queries.process_ciq_3g.importlib.util.find_spec") as mock_find_spec: mock_find_spec.side_effect = lambda name: object() if name == "openpyxl" else None assert _get_excel_writer_engine() == "openpyxl" def test_get_excel_writer_engine_raises_clear_error_when_missing(): with patch("queries.process_ciq_3g.importlib.util.find_spec", return_value=None): try: _get_excel_writer_engine() except RuntimeError as exc: assert "xlsxwriter" in str(exc) assert "openpyxl" in str(exc) else: raise AssertionError("Expected RuntimeError when no Excel writer engine is installed") def test_generate_ciq_3g_excel_returns_final_schema_workbook(): captured_sheet_names = [] captured_engine = None class DummyWriter: def __enter__(self): return self def __exit__(self, exc_type, exc, tb): return False def _capture_excel_writer(*args, **kwargs): nonlocal captured_engine captured_engine = kwargs.get("engine") return DummyWriter() def _capture_to_excel(self, writer, sheet_name=None, index=True, **kwargs): captured_sheet_names.append(sheet_name) with ( patch("queries.process_ciq_3g.read_ciq_3g_brut", return_value=_sample_ciq_df()), patch("queries.process_ciq_3g.importlib.util.find_spec") as mock_find_spec, patch("queries.process_ciq_3g.pd.ExcelWriter", side_effect=_capture_excel_writer), patch.object(pd.DataFrame, "to_excel", autospec=True, side_effect=_capture_to_excel), ): mock_find_spec.side_effect = lambda name: object() if name == "xlsxwriter" else None sheets, excel_bytes = generate_ciq_3g_excel( io.BytesIO(b"placeholder"), year_suffix="26", bands="U9U21" ) assert list(sheets["WBTS"].columns) == WBTS_FINAL_COLUMNS assert list(sheets["WCEL"].columns) == WCEL_FINAL_COLUMNS assert captured_sheet_names == ["WBTS", "WCEL"] assert captured_engine == "xlsxwriter" assert isinstance(excel_bytes, bytes) assert sheets["WBTS"].loc[0, "WinLCHSDPA"] == WBTS_FINAL_DEFAULTS["WinLCHSDPA"] assert sheets["WCEL"].loc[0, "PrxTarget"] == WCEL_FINAL_DEFAULTS["PrxTarget"]