tests/test_process_ciq_2g_final_schema.py

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"