| 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"] |
|
|
