import io from numbers import Integral from unittest.mock import patch import pandas as pd from openpyxl import load_workbook from queries.ciq_2g_schema_loader import ( BTS_AMR_FINAL_COLUMNS, BTS_AMR_FINAL_DEFAULTS, BTS_FINAL_COLUMNS, BTS_FINAL_DEFAULTS, BTS_GPRS_FINAL_COLUMNS, BTS_GPRS_FINAL_DEFAULTS, HOC_FINAL_COLUMNS, HOC_FINAL_DEFAULTS, POC_FINAL_COLUMNS, POC_FINAL_DEFAULTS, ) from queries.process_ciq_2g import ( _PlannedSite, _build_bts_amr_sheet_from_bts, _build_bts_gprs_sheet_from_bts, _build_bts_sheet_from_assigned_sites, _build_hoc_sheet_from_bts, _build_poc_sheet_from_bts, _get_excel_writer_engine, apply_final_schema, generate_ciq_2g_excel, ) def _sample_ciq_df() -> pd.DataFrame: return pd.DataFrame( [ { "Sites": "0273_TBC_DIRE-HAMDALLAYE_2G", "NOM_CELLULE": "0273_TBC_DIRE-HAMDALLAYE_1_900", "Nbre_TRE_DR": 4, "LAC": 30802, "RAC": 1, "CI": 2731, "Frequence": "GSM900", "BCCH": 5, "TRX": "47,51,58", "NCC": 5, "BCC": 5, "HSN": 56, "MAIO": 0, "Nom BSC": "ASBSCMSC3", "BSC ID": 403703, "band": "G9", "sector": 1, "site_number": 273, }, { "Sites": "0273_TBC_DIRE-HAMDALLAYE_2G", "NOM_CELLULE": "0273_TBC_DIRE-HAMDALLAYE_1_1800", "Nbre_TRE_DR": 3, "LAC": 30802, "RAC": 1, "CI": 2734, "Frequence": "GSM1800", "BCCH": 879, "TRX": "859,864", "NCC": 3, "BCC": 6, "HSN": 54, "MAIO": 0, "Nom BSC": "ASBSCMSC3", "BSC ID": 403703, "band": "G18", "sector": 1, "site_number": 273, }, ] ) def _sample_assigned_sites() -> list[_PlannedSite]: return [ _PlannedSite( site_name="0273_TBC_DIRE-HAMDALLAYE_2G", site_number=273, bsc=403703, bsc_name="ASBSCMSC3", name="0273_TBC_DIRE-HAMDALLAYE_2G_NA", configuration="G9-4, G18-3", assigned_bcf=200, needed_bts_ids=(201, 204), ) ] def _sample_bts_base_df() -> pd.DataFrame: return _build_bts_sheet_from_assigned_sites( _sample_ciq_df(), _sample_assigned_sites(), mcc=610, mnc=2 ) def test_apply_final_schema_bts_keeps_core_values_and_adds_formula_and_derived_fields(): df_bts = _sample_bts_base_df() final_df = apply_final_schema(df_bts, "BTS") assert list(final_df.columns) == BTS_FINAL_COLUMNS assert len(final_df.columns) == 109 assert list(final_df["site"]) == [273, 273] assert list(final_df["btsId"]) == [201, 204] assert list(final_df["template_name"]) == ["GSM900", "GSM1800"] assert list(final_df["frequencyBandInUse"]) == [0, 1] assert list(final_df["hoppingSequenceNumber1"]) == [56, 54] assert list(final_df["segmentId"]) == ["=E2", "=E3"] assert final_df.loc[0, "masterBCF"] == BTS_FINAL_DEFAULTS["masterBCF"] assert ( final_df.loc[0, "gprsCapacityThroughputFactor"] == BTS_FINAL_DEFAULTS["gprsCapacityThroughputFactor"] ) assert isinstance(final_df.loc[0, "fddRscpMin"], Integral) def test_apply_final_schema_bts_gprs_adds_defaults_without_overriding_base_columns(): final_df = apply_final_schema( _build_bts_gprs_sheet_from_bts(_sample_bts_base_df()), "BTS_GPRS" ) assert list(final_df.columns) == BTS_GPRS_FINAL_COLUMNS assert len(final_df.columns) == 56 assert list(final_df["Site"]) == [273, 273] assert list(final_df["btsId"]) == [201, 204] assert list(final_df["template_name"]) == ["All", "All"] assert ( final_df.loc[0, "dedicatedGPRScapacity"] == BTS_GPRS_FINAL_DEFAULTS["dedicatedGPRScapacity"] ) assert final_df.loc[0, "nsei"] == "" assert final_df.loc[0, "rac"] == BTS_GPRS_FINAL_DEFAULTS["rac"] def test_apply_final_schema_bts_amr_keeps_bscid_renamed_as_bscid_final_name(): final_df = apply_final_schema( _build_bts_amr_sheet_from_bts(_sample_bts_base_df()), "BTS_AMR" ) assert list(final_df.columns) == BTS_AMR_FINAL_COLUMNS assert len(final_df.columns) == 53 assert list(final_df["bscId"]) == [403703, 403703] assert list(final_df["template_name"]) == ["All", "All"] assert ( final_df.loc[0, "amrConfFrCodecModeSet"] == BTS_AMR_FINAL_DEFAULTS["amrConfFrCodecModeSet"] ) assert ( final_df.loc[0, "radioLinkTimeoutAmr"] == BTS_AMR_FINAL_DEFAULTS["radioLinkTimeoutAmr"] ) def test_apply_final_schema_hoc_and_poc_keep_identifiers_and_defaults(): df_bts = _sample_bts_base_df() hoc_df = apply_final_schema(_build_hoc_sheet_from_bts(df_bts), "HOC") poc_df = apply_final_schema(_build_poc_sheet_from_bts(df_bts), "POC") assert list(hoc_df.columns) == HOC_FINAL_COLUMNS assert len(hoc_df.columns) == 87 assert list(hoc_df["hocId"]) == [1, 1] assert hoc_df.loc[0, "enableSddchHandover"] == "" assert hoc_df.loc[0, "enableUmbrellaHo"] == HOC_FINAL_DEFAULTS["enableUmbrellaHo"] assert list(poc_df.columns) == POC_FINAL_COLUMNS assert len(poc_df.columns) == 65 assert "pocId" not in poc_df.columns assert list(poc_df["hocId"]) == [1, 1] assert poc_df.loc[0, "alpha"] == POC_FINAL_DEFAULTS["alpha"] assert ( poc_df.loc[0, "transmitPowerReduction"] == POC_FINAL_DEFAULTS["transmitPowerReduction"] ) def test_get_excel_writer_engine_prefers_xlsxwriter_then_openpyxl(): with patch("queries.process_ciq_2g.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_2g.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_2g.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_2g_excel_raises_clear_error_when_hsn_is_missing(): ciq_df = _sample_ciq_df().drop(columns=["HSN"]) with ( patch( "queries.process_ciq_2g._read_dump_bts_required_columns", return_value=pd.DataFrame(columns=["BSC", "BCF", "BTS", "usedMobileAllocation"]), ), patch("queries.process_ciq_2g._read_ciq_df", return_value=ciq_df), ): try: generate_ciq_2g_excel(io.BytesIO(b"dump"), io.BytesIO(b"ciq")) except ValueError as exc: assert "missing required column: HSN" in str(exc) assert "BTS.hoppingSequenceNumber1" in str(exc) else: raise AssertionError("Expected ValueError when HSN is missing") def test_generate_ciq_2g_excel_returns_finalized_target_sheets_only(): 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_2g._read_dump_bts_required_columns", return_value=pd.DataFrame(columns=["BSC", "BCF", "BTS", "usedMobileAllocation"]), ), patch("queries.process_ciq_2g._read_ciq_df", return_value=_sample_ciq_df()), patch("queries.process_ciq_2g._read_forbidden_bcfs", return_value=({}, pd.DataFrame())), patch("queries.process_ciq_2g._assign_bcfs", return_value=_sample_assigned_sites()), patch("queries.process_ciq_2g.importlib.util.find_spec") as mock_find_spec, patch("queries.process_ciq_2g.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_2g_excel(io.BytesIO(b"dump"), io.BytesIO(b"ciq")) assert list(sheets["BTS"].columns) == BTS_FINAL_COLUMNS assert list(sheets["BTS_GPRS"].columns) == BTS_GPRS_FINAL_COLUMNS assert list(sheets["BTS_AMR"].columns) == BTS_AMR_FINAL_COLUMNS assert list(sheets["HOC"].columns) == HOC_FINAL_COLUMNS assert list(sheets["POC"].columns) == POC_FINAL_COLUMNS assert len(sheets["MAL"].columns) == 16 assert len(sheets["TRX"].columns) == 78 assert captured_sheet_names == [ "BCF", "BCF_LIBRE", "BTS", "BTS_GPRS", "BTS_AMR", "HOC", "POC", "MAL", "BTS_PLMNPERMITTED", "TRX", ] assert captured_engine == "xlsxwriter" assert isinstance(excel_bytes, bytes) assert list(sheets["BTS"]["segmentId"]) == ["=E2", "=E3"] def test_generate_ciq_2g_excel_writes_real_workbook_with_segmentid_formula(): with ( patch( "queries.process_ciq_2g._read_dump_bts_required_columns", return_value=pd.DataFrame(columns=["BSC", "BCF", "BTS", "usedMobileAllocation"]), ), patch("queries.process_ciq_2g._read_ciq_df", return_value=_sample_ciq_df()), patch("queries.process_ciq_2g._read_forbidden_bcfs", return_value=({}, pd.DataFrame())), patch("queries.process_ciq_2g._assign_bcfs", return_value=_sample_assigned_sites()), ): sheets, excel_bytes = generate_ciq_2g_excel(io.BytesIO(b"dump"), io.BytesIO(b"ciq")) assert isinstance(excel_bytes, bytes) assert len(excel_bytes) > 0 assert list(sheets["BTS"]["segmentId"]) == ["=E2", "=E3"] workbook = load_workbook(io.BytesIO(excel_bytes), data_only=False) ws_bts = workbook["BTS"] headers = [cell.value for cell in ws_bts[1]] segment_col_idx = headers.index("segmentId") + 1 btsid_col_idx = headers.index("btsId") + 1 btsid_col_letter = ws_bts.cell(1, btsid_col_idx).column_letter assert ws_bts.cell(2, segment_col_idx).value == f"={btsid_col_letter}2" assert ws_bts.cell(3, segment_col_idx).value == f"={btsid_col_letter}3"