gandalf/tests/test_modem.py

"""Tests for modem.py parsing. Fixtures copy the BGW320-505 (fw 6.35.8) markup with sanitized values."""
import os
import sys
from unittest.mock import MagicMock

import pytest
import requests

sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))

import modem  # noqa: E402


def dom_table(heading, current, flags=(0, 0, 0, 0), limits=(-292, -90, -279, -100)):
    return f'''
    <h1>{heading}&nbsp;&nbsp;Currently {current}</h1>
    <table class="grid table100" summary="{heading} table">
    <thead><tr><th scope="col" width="30%">&nbsp;</th><th scope="col">Low</th><th scope="col">High</th></tr></thead>
    <tbody>
    <tr class="a"><td class="heading" scope="row">Alarm</td>
    <td>{flags[0]}                 (Threshold {limits[0]})</td>
    <td>{flags[1]}                 (Threshold {limits[1]})</td>
    </tr>
    <tr class="b"><td class="heading" scope="row">Warning</td>
    <td>{flags[2]}                 (Threshold {limits[2]})</td>
    <td>{flags[3]}                 (Threshold {limits[3]})</td>
    </tr>
    </tbody></table>'''


def fiber_page(optical='Up', rx=-169, rx_flags=(0, 0, 0, 0), include_rx=True):
    rx_table = dom_table('Rx Power', rx, rx_flags) if include_rx else ''
    return f'''<html><title>Fiber Status</title>
    <tr><th scope="row" width="47%">Optical WAN Operational Status</th>
    <td class="col2">{optical}                 </td></tr>
    <tr><th scope="row">Wave Length</th><td class="col2">1270 nm </td></tr>
    <tr><th scope="row">Rx LOS State</th><td class="col2">1</td></tr>
    <tr><th scope="row">Vendor PN</th><td class="col2">TESTPN</td></tr>
    {dom_table('Temperature', 34, limits=(-50, 95, -45, 90))}
    {dom_table('Vcc', 3, limits=(3, 3, 3, 3))}
    {dom_table('Tx Bias', 6, limits=(0, 500, 0, 400))}
    {dom_table('Tx Power', 33, limits=(-10, 59, 0, 49))}
    {rx_table}</html>'''


def broadband_page(connection='Up', pon='OPERATION (O5)'):
    return f'''<html><title>Broadband Status</title>
    <tr><th scope="row" width="47%">Broadband Connection Source</th><td class="col2">FIBER</td></tr>
    <tr><th scope="row" width="47%">Broadband Connection</th>
        <td class="col2">
{connection}
        </td></tr>
    <tr><th scope="row">Broadband IPv4 Address</th><td class="col2">203.0.113.10</td></tr>
    <tr class="a"><th scope="row" class="heading a">Current Speed (Mbps)</th><td>10000</td></tr>
    <tr><th scope="row">Transmit Bytes</th><td>200</td></tr>
    <tr><th scope="row">Receive Bytes</th><td>100</td></tr>
    <tr><th scope="row">Receive Errors</th><td>1</td></tr>
    <tr><th scope="row">Transmit Errors</th><td>2</td></tr>
    <tr><th scope="row">Receive Drops</th><td>3</td></tr>
    <tr><th scope="row">Transmit Drops</th><td>4</td></tr>
    <tr><th scope="row">Transmit Packets</th><td>7</td></tr>
    <tr><th scope="row">Transmit Packets</th><td>999</td></tr>
    <tr><th scope="row">PON Link Status</th><td>{pon}</td></tr>
    <tr><th scope="row">UNI Status</th><td>up</td></tr></html>'''


SYSINFO = '''<tr><th scope="row">Model Number</th><td class="col2">BGW320-505</td></tr>
<tr><th scope="row">Software Version</th><td class="col2">6.35.8</td></tr>
<tr><th scope="row">Time Since Last Reboot</th><td class="col2">28951</td></tr>'''


class TestBuildStatus:
    def test_scales_optical_power_from_tenths_of_dbm(self):
        d = modem.build_status(fiber_page(), broadband_page(), SYSINFO)
        assert d['optics']['rx_power']['value'] == -16.9
        assert d['optics']['tx_power']['value'] == 3.3
        assert d['optics']['temperature']['value'] == 34.0

    def test_scales_thresholds(self):
        rx = modem.build_status(fiber_page(), broadband_page())['optics']['rx_power']
        assert (rx['alarm_low'], rx['alarm_high'], rx['warn_low'], rx['warn_high']) == (-29.2, -9.0, -27.9, -10.0)

    def test_unit_ambiguous_readings_are_dropped_but_flags_kept(self):
        d = modem.build_status(fiber_page(), broadband_page())
        assert d['optics']['tx_bias']['value'] is None
        assert d['optics']['vcc']['alarm_low'] is None
        assert d['optics']['vcc']['alarm_flag'] is False

    def test_system_and_wan_fields(self):
        d = modem.build_status(fiber_page(), broadband_page(), SYSINFO)
        assert d['model'] == 'BGW320-505'
        assert d['firmware'] == '6.35.8'
        assert d['uptime_s'] == 28951
        assert d['broadband_state'] == 'Up'
        assert d['line_speed_mbps'] == 10000
        assert d['counters'] == {
            'rx_bytes': 100, 'tx_bytes': 200, 'rx_errors': 1,
            'tx_errors': 2, 'rx_drops': 3, 'tx_drops': 4,
        }

    def test_repeated_label_keeps_first_value(self):
        assert modem.parse_rows(broadband_page())['Transmit Packets'] == '7'

    def test_sysinfo_is_optional(self):
        d = modem.build_status(fiber_page(), broadband_page(), '')
        assert d['model'] == '' and d['uptime_s'] is None

    def test_unknown_page_is_invalid(self):
        with pytest.raises(modem.ModemDataInvalid):
            modem.build_status('<html>nope</html>', broadband_page())
        with pytest.raises(modem.ModemDataInvalid):
            modem.build_status(fiber_page(), '<html>nope</html>')

    def test_up_link_without_rx_power_is_invalid(self):
        with pytest.raises(modem.ModemDataInvalid):
            modem.build_status(fiber_page(include_rx=False), broadband_page())

    def test_cut_fiber_without_rx_power_is_not_invalid(self):
        d = modem.build_status(fiber_page(optical='Down', include_rx=False), broadband_page(connection='Down'))
        assert 'rx_power' not in d['optics']
        assert modem.wan_problems(d)


class TestWanProblems:
    def test_healthy_link_has_none(self):
        assert modem.wan_problems(modem.build_status(fiber_page(), broadband_page())) == []

    @pytest.mark.parametrize('field,value,expected', [
        ('broadband_state', 'Down', 'broadband connection is Down'),
        ('optical_state', 'Down', 'optical WAN is Down'),
        ('pon_state', 'INITIAL (O1)', 'PON link state is INITIAL (O1)'),
    ])
    def test_each_failure_is_reported(self, field, value, expected):
        d = {'broadband_state': 'Up', 'optical_state': 'Up', 'pon_state': 'OPERATION (O5)', field: value}
        assert modem.wan_problems(d) == [expected]

    def test_missing_pon_field_is_not_a_problem(self):
        assert modem.wan_problems({'broadband_state': 'Up', 'optical_state': 'Up', 'pon_state': ''}) == []

    def test_rx_los_pin_is_ignored(self):
        # Real gateways report Rx LOS State=1 on a healthy link (see fixture).
        assert modem.wan_problems(modem.build_status(fiber_page(), broadband_page())) == []


class TestMetricStatus:
    @pytest.mark.parametrize('rx,expected', [
        (-169, 'ok'),
        (-279, 'warning'),
        (-285, 'warning'),
        (-292, 'alarm'),
        (-400, 'alarm'),
        (-100, 'warning'),
        (-90, 'alarm'),
    ])
    def test_rx_thresholds(self, rx, expected):
        rx_metric = modem.parse_dom_metrics(fiber_page(rx=rx))['rx_power']
        assert modem.metric_status(rx_metric) == expected

    def test_module_flags_win_over_reading(self):
        alarm = modem.parse_dom_metrics(fiber_page(rx_flags=(1, 0, 0, 0)))['rx_power']
        warn = modem.parse_dom_metrics(fiber_page(rx_flags=(0, 0, 1, 0)))['rx_power']
        assert modem.metric_status(alarm) == 'alarm'
        assert modem.metric_status(warn) == 'warning'

    def test_flag_only_metric(self):
        assert modem.metric_status(modem.parse_dom_metrics(fiber_page())['vcc']) == 'ok'


class TestCounterRates:
    def test_rate_from_delta(self):
        assert modem.counter_rates({'rx_bytes': 300}, {'rx_bytes': 100}, 100)['rx_bytes'] == 2.0

    def test_first_poll_is_none(self):
        assert modem.counter_rates({'rx_bytes': 300}, None, 100)['rx_bytes'] is None

    def test_reboot_counter_reset_is_none(self):
        assert modem.counter_rates({'rx_bytes': 5}, {'rx_bytes': 100}, 100)['rx_bytes'] is None


class TestClient:
    def _client(self, **response):
        c = modem.ModemClient({'url': 'http://modem.test/'})
        resp = MagicMock(status_code=response.get('status', 200), text=response.get('text', ''))
        c.session.get = MagicMock(return_value=resp)
        return c

    def test_network_failure_is_unreachable(self):
        c = self._client()
        c.session.get = MagicMock(side_effect=requests.ConnectionError('boom'))
        with pytest.raises(modem.ModemUnreachable):
            c.poll()

    def test_redirect_is_invalid_and_not_followed(self):
        c = self._client(status=302)
        with pytest.raises(modem.ModemDataInvalid):
            c.poll()
        assert c.session.get.call_args.kwargs['allow_redirects'] is False

    def test_login_page_is_invalid(self):
        with pytest.raises(modem.ModemDataInvalid, match='login'):
            self._client(text='<html><title>Login</title></html>').poll()


class TestWanHistory:
    def ev(self, key, ts, device='RPDU'):
        return {'key': key, 'ts': ts, 'device': device, 'wan': 'WAN1'}

    def test_outage_opens_and_closes(self):
        h = {}
        modem.apply_wan_events(h, [self.ev(modem.LOG_WAN_FAILED, 100), self.ev(modem.LOG_WAN_RESTORED, 200)])
        assert h['outages'] == [{'failed_at': 100, 'restored_at': 200, 'wan': 'WAN1'}]

    def test_events_are_handled_in_time_order(self):
        h = {}
        modem.apply_wan_events(h, [self.ev(modem.LOG_WAN_RESTORED, 200), self.ev(modem.LOG_WAN_FAILED, 100)])
        assert h['outages'][0]['restored_at'] == 200

    def test_outage_spanning_two_batches(self):
        h = {}
        modem.apply_wan_events(h, [self.ev(modem.LOG_WAN_FAILED, 100)])
        assert h['outages'][0]['restored_at'] is None
        modem.apply_wan_events(h, [self.ev(modem.LOG_WAN_RESTORED, 200)])
        assert h['outages'][0]['restored_at'] == 200

    def test_repeat_failure_does_not_open_a_second_outage(self):
        h = {}
        modem.apply_wan_events(h, [self.ev(modem.LOG_WAN_FAILED, 100), self.ev(modem.LOG_WAN_FAILED, 150)])
        assert len(h['outages']) == 1

    def test_restored_without_failure_is_ignored(self):
        h = {}
        modem.apply_wan_events(h, [self.ev(modem.LOG_WAN_RESTORED, 200)])
        assert h['outages'] == []

    def test_blip_is_not_actionable_but_restart_and_instability_are(self):
        keys = [modem.LOG_WAN_FAILED, modem.LOG_WAN_RESTORED, modem.LOG_MODEM_RESTARTED, modem.LOG_WAN_UNSTABLE]
        got = modem.apply_wan_events({}, [self.ev(k, i) for i, k in enumerate(keys)])
        assert [e['key'] for e in got] == [modem.LOG_MODEM_RESTARTED, modem.LOG_WAN_UNSTABLE]

    def test_history_is_bounded(self):
        h = {}
        for i in range(60):
            modem.apply_wan_events(h, [
                self.ev(modem.LOG_WAN_FAILED, i * 2), self.ev(modem.LOG_WAN_RESTORED, i * 2 + 1),
            ])
        assert len(h['outages']) == 50

    def test_uptime_reset_confirms_logged_restart(self):
        h = {}
        modem.apply_wan_events(h, [self.ev(modem.LOG_MODEM_RESTARTED, 1000)])
        modem.record_uptime_reset(h, 1023)
        assert h['restarts'] == [{'at': 1000, 'by': 'RPDU', 'confirmed': True}]

    def test_unexplained_uptime_reset_is_recorded(self):
        h = {}
        modem.record_uptime_reset(h, 5000)
        assert h['restarts'] == [{'at': 5000, 'by': None, 'confirmed': True}]