gandalf/tests/test_app.py

"""Tests for app.py routes: database and LDAP are mocked."""
import json
import time
from datetime import datetime, timedelta, timezone
import os
import sys
from unittest.mock import patch

import pytest

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

import app as gandalf  # noqa: E402

PROXY = '10.0.0.27'
# What the local nginx forwards for a request that came through the Authelia proxy
AUTH = {'Remote-User': 'jared', 'Remote-Name': 'Jared V', 'Remote-Groups': 'admin', 'X-Real-IP': PROXY}
CONFIG = {'auth': {'allowed_groups': ['admin'], 'trusted_proxies': [PROXY]}, 'ticket_api': {}}
STATE = {
    'network_snapshot': json.dumps({
        'hosts': {'large1': {'ip': '10.0.0.2', 'status': 'up', 'source': 'prometheus', 'interfaces': {'eth0': 'up'}}},
        'unifi': [{'name': 'USW Flex', 'type': 'usw', 'model': 'USF5P', 'ip': '10.0.0.3', 'mac': '',
                   'connected': True}],
    }),
    'link_stats': json.dumps({'hosts': {}, 'unifi_switches': {}}),
}


@pytest.fixture
def client():
    with patch.object(gandalf, 'db') as db:
        db.get_state.side_effect = lambda key, default=None: STATE.get(key, default)
        db.get_active_events.return_value = []
        db.get_recent_resolved.return_value = []
        db.get_active_suppressions.return_value = []
        db.get_events_for_target.return_value = []
        db.count_active_events.return_value = 0
        db.get_status_summary.return_value = {'critical': 0, 'warning': 0, 'info': 0, 'resolved_24h': 0}
        db.check_suppressed.return_value = False
        db.get_device_link.return_value = ''
        db.cutsheet_racks.return_value = []
        db.get_unused_interfaces.return_value = {}
        with patch.object(gandalf, '_cfg', CONFIG):
            yield gandalf.app.test_client()


class TestIdentityHeaderTrust:
    def test_proxy_identity_is_accepted(self, client):
        assert client.get('/api/devices', headers=AUTH).status_code == 200

    def test_lan_client_through_nginx_cannot_claim_a_user(self, client):
        spoofed = dict(AUTH, **{'X-Real-IP': '10.0.0.99'})
        assert client.get('/api/devices', headers=spoofed).status_code == 401

    def test_loopback_without_nginx_source_is_untrusted(self, client):
        spoofed = {k: v for k, v in AUTH.items() if k != 'X-Real-IP'}
        assert client.get('/api/devices', headers=spoofed).status_code == 401

    def test_direct_peer_cannot_borrow_the_proxy_address(self, client):
        resp = client.get('/api/devices', headers=AUTH, environ_base={'REMOTE_ADDR': '10.0.0.99'})
        assert resp.status_code == 401

    def test_proxy_connecting_directly_is_trusted(self, client):
        resp = client.get('/api/devices', headers=AUTH, environ_base={'REMOTE_ADDR': PROXY})
        assert resp.status_code == 200

    def test_fails_closed_without_trusted_proxies(self, client):
        with patch.object(gandalf, '_cfg', {'auth': {'allowed_groups': ['admin']}}):
            assert client.get('/api/devices', headers=AUTH).status_code == 401

    def test_cidr_entries_and_bad_entries(self, client):
        cfg = {'auth': {'allowed_groups': ['admin'], 'trusted_proxies': ['not-an-ip', '10.0.0.0/24']}}
        with patch.object(gandalf, '_cfg', cfg):
            assert client.get('/api/devices', headers=AUTH).status_code == 200

    def test_garbage_source_header_is_untrusted(self, client):
        assert client.get('/api/devices', headers=dict(AUTH, **{'X-Real-IP': 'not-an-ip'})).status_code == 401

    def test_rejection_is_logged(self, client, caplog):
        client.get('/api/devices', headers=dict(AUTH, **{'X-Real-IP': '10.0.0.99'}))
        assert "untrusted source '10.0.0.99'" in caplog.text

    def test_avatar_ignores_untrusted_identity(self, client):
        with patch.object(gandalf, '_config', return_value=dict(CONFIG, ldap={'host': 'h', 'bind_dn': 'b'})):
            resp = client.get('/api/avatar', headers=dict(AUTH, **{'X-Real-IP': '10.0.0.99'}))
        assert resp.status_code == 401


class TestPagesAndNav:
    def test_home_has_search_and_no_link_debug(self, client):
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        assert 'id="device-search-input"' in html
        assert 'Link Debug' not in html

    def test_home_section_order(self, client):
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        order = [html.index(m) for m in (
            'id="device-search"', 'Alert Queue', 'WAN / Fiber', 'g-section-title">Hosts<',
            'Device Inventory',
        )]
        assert order == sorted(order)

    def test_dashboard_topology_moved_to_its_own_page(self, client):
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        assert 'topology-diagram' not in html and 'Network Topology' not in html
        assert 'href="/topology"' in html

    def test_old_links_url_redirects_home(self, client):
        resp = client.get('/links', headers=AUTH)
        assert resp.status_code == 302 and resp.headers['Location'].endswith('/')

    def test_device_page_escapes_name(self, client):
        html = client.get('/device/%3Cscript%3Ex', headers=AUTH).get_data(as_text=True)
        assert '<script>x' not in html

    def test_old_inspector_url_redirects_home(self, client):
        resp = client.get('/inspector', headers=AUTH)
        assert resp.status_code == 302 and resp.headers['Location'].endswith('/')

    def test_nav_has_no_inspector_link(self, client):
        assert '>Inspector<' not in client.get('/', headers=AUTH).get_data(as_text=True)
        assert 'Inspector' not in client.get('/device/large1', headers=AUTH).get_data(as_text=True).split('</nav>')[0]

    def test_device_page_loads_the_port_inspector(self, client):
        html = client.get('/device/large1', headers=AUTH).get_data(as_text=True)
        assert 'port-inspector.js' in html and 'id="inspector-panel"' in html

    def test_faceplates_load_before_the_port_inspector(self, client):
        html = client.get('/device/large1', headers=AUTH).get_data(as_text=True)
        assert html.index('faceplates.js') < html.index('port-inspector.js')

    def test_link_data_api_remains(self, client):
        assert client.get('/api/links', headers=AUTH).status_code == 200


class TestDeviceApi:
    def test_requires_auth(self, client):
        assert client.get('/api/devices').status_code == 401
        assert client.get('/api/device/large1').status_code == 401

    def test_index_lists_devices(self, client):
        names = [d['name'] for d in client.get('/api/devices', headers=AUTH).get_json()['devices']]
        assert names == ['large1', 'USW Flex']

    def test_name_with_a_space(self, client):
        d = client.get('/api/device/USW%20Flex', headers=AUTH).get_json()
        assert d['kind'] == 'switch' and d['events'] == []

    def test_unknown_device_is_404(self, client):
        assert client.get('/api/device/nope', headers=AUTH).status_code == 404


class TestAvatarLogging:
    def test_missing_ldap_config_says_why(self, client, caplog):
        assert client.get('/api/avatar', headers=AUTH).status_code == 404
        assert 'no ldap' in caplog.text


class TestDatabaseTimeZone:
    def test_sessions_run_in_utc(self):
        import db as real_db
        with patch.object(real_db.pymysql, 'connect') as connect:
            real_db._new_conn({'host': 'h', 'user': 'u', 'password': 'p', 'name': 'n'})
        assert connect.call_args.kwargs['init_command'] == "SET time_zone='+00:00'"


class TestDiagnoseWithResolvedNeighbor:
    LINKS = json.dumps({
        'hosts': {'large1': {'enp35s0': {'host_ip': '10.0.0.2'}}},
        'unifi_switches': {'sw': {'ports': {'Port 4': {
            'port_idx': 4, 'up': True,
            'lldp': {'system_name': 'large1', 'port_id': 'enp35s0', 'chassis_id': 'aa:bb:cc:dd:ee:ff'},
        }}}},
    })

    def test_port_with_a_named_neighbor_starts_a_job(self, client):
        STATE['link_stats'], saved = self.LINKS, STATE['link_stats']
        try:
            with patch.object(gandalf.threading, 'Thread'):
                resp = client.post('/api/diagnose', json={'switch_name': 'sw', 'port_idx': 4}, headers=AUTH)
        finally:
            STATE['link_stats'] = saved
        assert resp.status_code == 200 and resp.get_json()['job_id']


CUTSHEET = {
    'items': [{'id': 1, 'name': 'PP1', 'kind': 'patch_panel', 'port_count': 24, 'location': 'rack',
               'rack_u': 25, 'rack_height': 1, 'notes': ''}],
    'cables': [{'id': 7, 'label': 'C07', 'a_device': 'USW Flex', 'a_port': '2', 'a_side': '',
                'b_device': 'PP1', 'b_port': '5', 'b_side': 'front', 'status': 'installed',
                'cable_type': 'Cat6', 'length_m': None, 'color': '', 'notes': '', 'label_applied': False}],
    'port_labels': [{'item_id': 1, 'item_name': 'PP1', 'port': '5', 'label': 'OFFICE', 'applied': False}],
}
USER = dict(AUTH, **{'Remote-Groups': 'staff'})


@pytest.fixture
def cs_client(client):
    gandalf._cutsheet_schema_ready = True
    db = gandalf.db
    db.cutsheet_load.side_effect = lambda: json.loads(json.dumps(CUTSHEET))
    db.cutsheet_history.return_value = []
    db.cutsheet_create_cable.return_value = 8
    db.CutsheetConflict = type('CutsheetConflict', (Exception,), {})
    with patch.object(gandalf, '_cfg', dict(CONFIG, auth={'allowed_groups': ['admin', 'staff'],
                                                          'trusted_proxies': [PROXY]})):
        yield client, db


class TestCutsheetApi:
    def test_page_and_view(self, cs_client):
        client, _ = cs_client
        assert client.get('/cutsheet', headers=AUTH).status_code == 200
        view = client.get('/api/cutsheet', headers=AUTH).get_json()
        c = view['cables'][0]
        assert c['b_text'] == 'PP1 port 5 (front)' and c['verdict'] in view['verdicts']
        assert view['next_label'] == 'C01'
        assert view['labels']['cables'][0]['text'] == 'C07'
        assert any(d['name'] == 'PP1' for d in view['catalog'])

    def test_create_cable(self, cs_client):
        client, db = cs_client
        resp = client.post('/api/cutsheet/cables', headers=AUTH,
                           json={'a_device': 'USW Flex', 'a_port': '3', 'b_device': 'PP1', 'b_port': '6',
                                 'b_side': 'front', 'label': 'C08'})
        assert resp.status_code == 200 and resp.get_json()['id'] == 8
        saved, user = db.cutsheet_create_cable.call_args.args
        assert saved['b_side'] == 'front' and user == 'jared'

    def test_viewers_cannot_edit(self, cs_client):
        client, db = cs_client
        assert client.get('/api/cutsheet', headers=USER).status_code == 200
        resp = client.post('/api/cutsheet/cables', headers=USER, json={'a_device': 'x'})
        assert resp.status_code == 403 and not db.cutsheet_create_cable.called

    def test_writes_require_json(self, cs_client):
        client, db = cs_client
        resp = client.post('/api/cutsheet/cables', headers=AUTH, data='a_device=x',
                           content_type='application/x-www-form-urlencoded')
        assert resp.status_code == 400 and not db.cutsheet_create_cable.called

    def test_validation_error_is_400(self, cs_client):
        client, _ = cs_client
        resp = client.post('/api/cutsheet/cables', headers=AUTH, json={'a_device': 'PP1', 'a_port': '3'})
        assert resp.status_code == 400 and 'front or rear' in resp.get_json()['error']

    def test_conflict_is_409(self, cs_client):
        client, db = cs_client
        db.cutsheet_create_item.side_effect = db.CutsheetConflict('Another item already has that name')
        resp = client.post('/api/cutsheet/items', headers=AUTH, json={'name': 'PP1', 'kind': 'shelf'})
        assert resp.status_code == 409 and 'already' in resp.get_json()['error']

    def test_missing_records_are_404(self, cs_client):
        client, db = cs_client
        db.cutsheet_delete_cable.return_value = False
        assert client.delete('/api/cutsheet/cables/99', headers=AUTH).status_code == 404

    def test_port_label(self, cs_client):
        client, db = cs_client
        resp = client.put('/api/cutsheet/items/1/ports/5', headers=AUTH, json={'label': 'GUEST BR', 'applied': True})
        assert resp.status_code == 200
        assert db.cutsheet_set_port_label.call_args.args == (1, '5', 'GUEST BR', True, 'jared')

    def test_csv_export_and_all_or_nothing_import(self, cs_client):
        client, db = cs_client
        text = client.get('/api/cutsheet/cables.csv', headers=AUTH).get_data(as_text=True)
        assert text.startswith('label,a_device') and 'C07' in text
        bad = client.post('/api/cutsheet/import', headers=AUTH, json={'csv': text + 'C09,PP1,99,front\n'})
        assert bad.status_code == 400 and bad.get_json()['errors'] and not db.cutsheet_import_cables.called
        dry = client.post('/api/cutsheet/import', headers=AUTH, json={'csv': text, 'dry_run': True})
        assert dry.get_json()['rows'] == 1 and not db.cutsheet_import_cables.called
        db.cutsheet_import_cables.return_value = {'added': 0, 'updated': 1}
        done = client.post('/api/cutsheet/import', headers=AUTH, json={'csv': text})
        assert done.get_json()['updated'] == 1

    def test_device_page_and_search_include_the_cutsheet(self, cs_client):
        client, _ = cs_client
        d = client.get('/api/device/USW%20Flex', headers=AUTH).get_json()
        assert d['cutsheet'][0]['label'] == 'C07' and d['cutsheet'][0]['port'] == '2'
        found = client.get('/api/devices', headers=AUTH).get_json()['devices']
        by_name = {x['name']: x for x in found}
        assert by_name['C07']['url'] == '/cutsheet#cable-7'
        assert by_name['PP1']['url'] == '/cutsheet#item-1'

    def test_device_page_survives_a_cutsheet_failure(self, cs_client):
        client, db = cs_client
        db.cutsheet_load.side_effect = RuntimeError('db down')
        d = client.get('/api/device/USW%20Flex', headers=AUTH).get_json()
        assert d['cutsheet'] == []


class TestCutsheetLivePorts:
    def _fake(self, result):
        class Fake:
            calls = 0

            def __init__(self, cfg):
                pass

            def get_switch_ports(self):
                Fake.calls += 1
                return result
        return Fake

    def test_reads_unifi_and_caches_briefly(self, cs_client):
        client, _ = cs_client
        fake = self._fake({'sw': {'ports': {'Office TV': {'port_idx': 13, 'up': True},
                                            'Port 14': {'port_idx': 14, 'up': False}}}})
        gandalf._live_ports_cache.update(at=0.0, data=None, client=None)
        with patch.object(gandalf, 'UnifiClient', fake), \
                patch.dict(gandalf._cfg, {'unifi': {'controller': 'https://u', 'api_key': 'k'}}):
            first = client.get('/api/cutsheet/live-ports', headers=AUTH).get_json()
            client.get('/api/cutsheet/live-ports', headers=AUTH)
        assert first['switches'] == {'sw': {'13': {'up': True, 'name': 'Office TV'},
                                            '14': {'up': False, 'name': 'Port 14'}}}
        assert fake.calls == 1

    def test_controller_failure_is_502_and_not_cached(self, cs_client):
        client, _ = cs_client
        gandalf._live_ports_cache.update(at=0.0, data=None, client=None)
        with patch.object(gandalf, 'UnifiClient', self._fake(None)), \
                patch.dict(gandalf._cfg, {'unifi': {'controller': 'https://u', 'api_key': 'k'}}):
            assert client.get('/api/cutsheet/live-ports', headers=AUTH).status_code == 502
        assert gandalf._live_ports_cache['data'] is None

    def test_admin_only(self, cs_client):
        client, _ = cs_client
        assert client.get('/api/cutsheet/live-ports', headers=USER).status_code == 403


class TestAlertHistory:
    HISTORY = {'events': [], 'total': 0, 'by_type': {}}

    def _db(self):
        db = gandalf.db
        db.get_events_history.return_value = dict(self.HISTORY)
        db.get_flappers.return_value = []
        db.get_event_types.return_value = ['host_unreachable']
        return db

    def test_page_and_nav(self, client):
        html = client.get('/events', headers=AUTH).get_data(as_text=True)
        assert 'Alert History' in html and 'events.js' in html
        assert 'href="/events"' in client.get('/', headers=AUTH).get_data(as_text=True)

    def test_defaults_to_seven_days(self, client):
        db = self._db()
        data = client.get('/api/events/history', headers=AUTH).get_json()
        assert db.get_events_history.call_args.args == (168, '', '', '', 'all')
        assert data['types'] == ['host_unreachable'] and data['page_size'] == 100

    def test_filters_are_passed_through(self, client):
        db = self._db()
        client.get('/api/events/history?range=90d&type=interface_down&target=+large1+&severity=critical'
                   '&status=resolved&offset=200', headers=AUTH)
        assert db.get_events_history.call_args.args == (2160, 'interface_down', 'large1', 'critical', 'resolved')
        assert db.get_events_history.call_args.kwargs == {'limit': 100, 'offset': 200}
        assert db.get_flappers.call_args.args == (2160, 'interface_down', 'large1', 'critical')

    @pytest.mark.parametrize('query', ['range=1y', 'severity=bad', 'status=open', 'offset=x'])
    def test_bad_filters_are_rejected(self, client, query):
        self._db()
        assert client.get(f'/api/events/history?{query}', headers=AUTH).status_code == 400

    def test_needs_login(self, client):
        assert client.get('/api/events/history').status_code == 401


class TestHealth:
    def _health(self, client, age_s, poll=120):
        from datetime import datetime, timedelta, timezone
        stamp = (datetime.now(timezone.utc) - timedelta(seconds=age_s)).strftime('%Y-%m-%d %H:%M:%S UTC')
        gandalf.db.get_state.side_effect = lambda key, default=None: stamp if key == 'last_check' else default
        with patch.object(gandalf, '_cfg', dict(CONFIG, monitor={'poll_interval': poll})):
            return client.get('/health')

    def test_recent_check_is_healthy(self, client):
        resp = self._health(client, 150)
        assert resp.status_code == 200 and resp.get_json()['checks']['monitor'] == 'ok'

    def test_stalled_monitor_fails_after_ten_minutes(self, client):
        resp = self._health(client, 660)
        assert resp.status_code == 503
        assert resp.get_json()['checks'] == {'db': 'ok', 'monitor': 'stale', 'monitor_age_s': 660}

    def test_limit_follows_a_long_poll_interval(self, client):
        assert self._health(client, 1000, poll=300).status_code == 200
        assert self._health(client, 1600, poll=300).status_code == 503


class TestPageTitles:
    @pytest.mark.parametrize('path,title', [('/', 'Dashboard | GANDALF'), ('/events', 'Alert History | GANDALF'),
                                            ('/cutsheet', 'Cutsheet | GANDALF'),
                                            ('/topology', 'Topology | GANDALF')])
    def test_name_appears_once(self, client, path, title):
        gandalf._cutsheet_schema_ready = True
        html = client.get(path, headers=AUTH).get_data(as_text=True)
        assert f'<title>{title}</title>' in html


class TestInventoryLinks:
    def _put(self, client, url, headers=AUTH, name='USW Flex'):
        gandalf._cutsheet_schema_ready = True
        return client.put(f'/api/device/{name}/inventory', json={'url': url}, headers=headers)

    def test_device_api_has_link_and_search(self, client):
        gandalf._cutsheet_schema_ready = True
        gandalf.db.get_device_link.return_value = 'https://inventory.lotusguild.org/item/abc'
        inv = client.get('/api/device/USW Flex', headers=AUTH).get_json()['inventory']
        assert inv == {'url': 'https://inventory.lotusguild.org/item/abc',
                       'search_url': 'https://inventory.lotusguild.org/items?q=USF5P'}

    def test_admin_sets_and_clears(self, client):
        assert self._put(client, ' https://inventory.lotusguild.org/item/abc ').status_code == 200
        gandalf.db.set_device_link.assert_called_with('USW Flex', 'https://inventory.lotusguild.org/item/abc', 'jared')
        assert self._put(client, '').status_code == 200
        gandalf.db.set_device_link.assert_called_with('USW Flex', '', 'jared')

    @pytest.mark.parametrize('url', ['http://inventory.lotusguild.org/item/a', 'https://evil.example/item/a',
                                     'https://inventory.lotusguild.org.evil.example/x', 'javascript:alert(1)',
                                     'https://user:pw@inventory.lotusguild.org/x',
                                     'https://inventory.lotusguild.org/' + 'a' * 600])
    def test_only_inventory_links_are_accepted(self, client, url):
        assert self._put(client, url).status_code == 400
        assert not gandalf.db.set_device_link.called

    def test_configured_host(self, client):
        cfg = dict(CONFIG, inventory={'base_url': 'https://homebox.example'})
        with patch.object(gandalf, '_cfg', cfg):
            assert self._put(client, 'https://homebox.example/item/1').status_code == 200
            assert self._put(client, 'https://inventory.lotusguild.org/item/1').status_code == 400

    def test_needs_admin_and_json(self, client):
        user = dict(AUTH, **{'Remote-Groups': 'staff'})
        with patch.object(gandalf, '_cfg', dict(CONFIG, auth=dict(CONFIG['auth'], allowed_groups=['admin', 'staff']))):
            assert self._put(client, 'https://inventory.lotusguild.org/x', headers=user).status_code == 403
        gandalf._cutsheet_schema_ready = True
        resp = client.put('/api/device/USW Flex/inventory', data='url=x', headers=AUTH)
        assert resp.status_code == 400


class TestRackApi:
    def test_rack_combines_items_switches_and_marks(self, client):
        gandalf._cutsheet_schema_ready = True
        db = gandalf.db
        db.cutsheet_load.return_value = {
            'items': [
                {'id': 1, 'name': 'USW Flex', 'kind': 'device', 'port_count': None, 'rack_u': 20, 'rack_height': 1,
                 'location': '', 'notes': ''},
                {'id': 2, 'name': 'PP1', 'kind': 'patch_panel', 'port_count': 24, 'rack_u': 21, 'rack_height': 1,
                 'location': '', 'notes': ''},
                {'id': 3, 'name': 'Shelf', 'kind': 'shelf', 'port_count': None, 'rack_u': 20, 'rack_height': 1,
                 'location': '', 'notes': ''},
                {'id': 4, 'name': 'Loose', 'kind': 'other', 'port_count': None, 'rack_u': None, 'rack_height': None,
                 'location': '', 'notes': ''},
            ],
            'cables': [{'id': 7, 'label': 'C07', 'a_device': 'USW Flex', 'a_port': '2', 'a_side': '',
                        'b_device': 'PP1', 'b_port': '5', 'b_side': 'front', 'status': 'installed',
                        'cable_type': '', 'length_m': None, 'color': '', 'notes': '', 'label_applied': False}],
            'port_labels': [{'item_id': 2, 'item_name': 'PP1', 'port': '5', 'label': 'OFFICE', 'applied': False}],
        }
        data = client.get('/api/rack', headers=AUTH).get_json()
        assert data['units'] == 42 and data['unplaced'] == ['Loose']
        assert [i['name'] for i in data['items']] == ['PP1', 'USW Flex']
        assert data['overlaps'] == [{'name': 'Shelf', 'units': 'U20', 'with': ['USW Flex']}]
        pp, sw = data['items']
        assert pp['labels'] == {'5': 'OFFICE'} and pp['url'] == '/cutsheet#item-2'
        assert pp['ports']['5']['run'] == [7] and pp['ports']['5']['cable_id'] == 7
        assert sw['url'] == '/device/USW%20Flex' and sw['switch']['model'] == 'USF5P'  # model from UniFi
        assert sw['ports']['2']['mark'] in ('ok', 'bad', 'doc')

    def test_rack_page_and_auth(self, client):
        assert 'rack.js' in client.get('/rack', headers=AUTH).get_data(as_text=True)
        assert client.get('/api/rack').status_code == 401

    def test_units_from_config(self, client):
        gandalf._cutsheet_schema_ready = True
        gandalf.db.cutsheet_load.return_value = {'items': [], 'cables': [], 'port_labels': []}
        with patch.object(gandalf, '_cfg', dict(CONFIG, rack={'units': 24})):
            assert client.get('/api/rack', headers=AUTH).get_json()['units'] == 24


class TestRackRacksShelvesLinks:
    def _items(self, *items):
        base = {'port_count': None, 'rack_u': None, 'rack_height': None, 'rack': '', 'shelf': '', 'device': '',
                'location': '', 'notes': ''}
        gandalf._cutsheet_schema_ready = True
        gandalf.db.cutsheet_load.return_value = {
            'items': [dict(base, id=n, **i) for n, i in enumerate(items, 1)], 'cables': [], 'port_labels': []}

    def test_product_named_item_links_to_its_device(self, client):
        self._items({'name': 'Flex (desk)', 'kind': 'device', 'rack_u': 5, 'rack_height': 1, 'device': 'USW Flex'})
        item = client.get('/api/rack', headers=AUTH).get_json()['items'][0]
        assert item['device'] == 'USW Flex' and item['linked'] and item['url'] == '/device/USW%20Flex'
        assert item['status'] == 'up' and item['switch']['model'] == 'USF5P'

    def test_unlinked_item(self, client):
        self._items({'name': 'UNVR-Pro', 'kind': 'device', 'rack_u': 5, 'rack_height': 2})
        item = client.get('/api/rack', headers=AUTH).get_json()['items'][0]
        assert not item['linked'] and item['url'] == '/cutsheet#item-1' and 'switch' not in item

    def test_drawn_front_is_passed_through(self, client):
        self._items({'name': 'UNVR-Pro', 'kind': 'device', 'rack_u': 5, 'rack_height': 2, 'chassis': 'unvr-pro'})
        assert client.get('/api/rack', headers=AUTH).get_json()['items'][0]['chassis'] == 'unvr-pro'

    def test_shelf_lists_what_is_on_it(self, client):
        self._items({'name': 'Shelf', 'kind': 'shelf', 'rack_u': 3, 'rack_height': 1},
                    {'name': 'large1', 'kind': 'device', 'shelf': 'Shelf'},
                    {'name': 'Mini', 'kind': 'device', 'shelf': 'Shelf', 'device': 'USW Flex'})
        data = client.get('/api/rack', headers=AUTH).get_json()
        assert [i['name'] for i in data['items']] == ['Shelf'] and data['unplaced'] == []
        assert [(c['name'], c['status']) for c in data['items'][0]['contents']] == [('large1', 'up'), ('Mini', 'up')]

    def test_racks_tabs_and_selection(self, client):
        self._items({'name': 'A', 'kind': 'other', 'rack_u': 1, 'rack_height': 1},
                    {'name': 'B', 'kind': 'other', 'rack_u': 1, 'rack_height': 1, 'rack': 'Garage'})
        cfg = dict(CONFIG, racks=[{'name': 'Main', 'units': 24}])
        with patch.object(gandalf, '_cfg', cfg):
            data = client.get('/api/rack', headers=AUTH).get_json()
            assert [(r['id'], r['name'], r['units'], r['count']) for r in data['racks']] == [
                (None, 'Main', 24, 1), (None, 'Garage', 42, 1)]
            assert data['rack'] == 'Main' and [i['name'] for i in data['items']] == ['A']
            garage = client.get('/api/rack?rack=Garage', headers=AUTH).get_json()
            assert [i['name'] for i in garage['items']] == ['B'] and garage['units'] == 42
            assert client.get('/api/rack?rack=Nope', headers=AUTH).status_code == 404


class TestRackFacesAndSettings:
    def _items(self, *items):
        base = {'port_count': None, 'rack_u': None, 'rack_height': None, 'rack': '', 'shelf': '', 'device': '',
                'mount': 'front', 'location': '', 'notes': ''}
        gandalf._cutsheet_schema_ready = True
        gandalf.db.cutsheet_load.return_value = {
            'items': [dict(base, id=n, **i) for n, i in enumerate(items, 1)], 'cables': [], 'port_labels': []}

    def test_three_views(self, client):
        # A server through both posts, a PDU on the rear posts facing the front (as in the StarTech),
        # a fan tray on the rear posts facing the rear, and a switch on the front posts
        self._items({'name': 'Server', 'kind': 'device', 'rack_u': 5, 'rack_height': 4, 'mount': 'full'},
                    {'name': 'PDU', 'kind': 'device', 'rack_u': 12, 'rack_height': 1, 'mount': 'rear'},
                    {'name': 'Fans', 'kind': 'device', 'rack_u': 14, 'rack_height': 1, 'mount': 'rear',
                     'facing': 'rear'},
                    {'name': 'Switch', 'kind': 'device', 'rack_u': 12, 'rack_height': 1})

        def view(face):
            return [(i['name'], i['through']) for i in
                    client.get(f'/api/rack?face={face}', headers=AUTH).get_json()['items']]
        assert view('front') == [('Switch', False), ('Server', False)]
        assert view('behind') == [('Fans', True), ('PDU', False), ('Server', False)]
        assert view('rear') == [('Fans', False), ('PDU', True), ('Server', True)]
        racks = client.get('/api/rack', headers=AUTH).get_json()['racks'][0]
        assert (racks['count'], racks['rear'], racks['behind']) == (4, 3, 2)
        assert client.get('/api/rack?face=side', headers=AUTH).status_code == 400

    def test_shelf_contents_ordered_with_nics(self, client):
        self._items({'name': 'Shelf', 'kind': 'shelf', 'rack_u': 3, 'rack_height': 2, 'mount': 'rear'},
                    {'name': 'b-box', 'kind': 'device', 'shelf': 'Shelf', 'slot': 1, 'chassis': 'zimaboard-832'},
                    {'name': 'large1', 'kind': 'device', 'shelf': 'Shelf', 'slot': 2})
        data = client.get('/api/rack?face=behind', headers=AUTH).get_json()
        things = data['items'][0]['contents']
        assert [(c['name'], c['slot'], c['chassis']) for c in things] == [('b-box', 1, 'zimaboard-832'),
                                                                          ('large1', 2, '')]
        assert things[1]['interfaces'] == [{'name': 'eth0', 'up': True, 'unused': False}]
        assert things[0]['device'] == 'b-box' and things[0]['ports'] == {}

    def test_saved_racks_replace_config(self, client):
        self._items()
        gandalf.db.cutsheet_racks.return_value = [{'id': 4, 'name': 'StarTech 25U', 'units': 25, 'notes': 'open frame'}]
        data = client.get('/api/rack', headers=AUTH).get_json()
        assert (data['rack'], data['rack_id'], data['units'], data['notes']) == ('StarTech 25U', 4, 25, 'open frame')

    def test_first_save_carries_the_default_racks_items(self, client):
        gandalf._cutsheet_schema_ready = True
        gandalf.db.cutsheet_save_rack.return_value = 7
        resp = client.post('/api/cutsheet/racks', json={'name': 'StarTech 25U', 'units': 25, 'previous': 'Rack 1'},
                           headers=AUTH)
        assert resp.get_json() == {'success': True, 'id': 7}
        args, kwargs = gandalf.db.cutsheet_save_rack.call_args
        assert args[0] is None and args[1]['units'] == 25 and kwargs == {'previous': 'Rack 1'}

    def test_later_racks_ignore_previous(self, client):
        gandalf._cutsheet_schema_ready = True
        gandalf.db.cutsheet_racks.return_value = [{'id': 4, 'name': 'A', 'units': 25, 'notes': ''}]
        client.post('/api/cutsheet/racks', json={'name': 'B', 'units': 42, 'previous': 'A'}, headers=AUTH)
        assert gandalf.db.cutsheet_save_rack.call_args.kwargs == {'previous': ''}

    def test_rack_writes_need_admin_and_valid_data(self, client):
        gandalf._cutsheet_schema_ready = True
        assert client.post('/api/cutsheet/racks', json={'name': 'x', 'units': 99}, headers=AUTH).status_code == 400
        user = dict(AUTH, **{'Remote-Groups': 'staff'})
        with patch.object(gandalf, '_cfg', dict(CONFIG, auth=dict(CONFIG['auth'], allowed_groups=['admin', 'staff']))):
            assert client.put('/api/cutsheet/racks/1', json={'name': 'x', 'units': 2}, headers=user).status_code == 403
        gandalf.db.cutsheet_delete_rack.return_value = False
        assert client.delete('/api/cutsheet/racks/9', headers=AUTH).status_code == 404


class TestUnusedInterfaces:
    def _url(self, iface='nic1'):
        gandalf._cutsheet_schema_ready = True
        return f'/api/device/compute-storage-01/interfaces/{iface}/unused'

    def test_mark_and_clear(self, client):
        assert client.put(self._url(), json={'reason': ' never cabled '}, headers=AUTH).status_code == 200
        gandalf.db.set_unused_interface.assert_called_with('compute-storage-01', 'nic1', 'never cabled', 'jared')
        gandalf.db.clear_unused_interface.return_value = True
        resp = client.delete(self._url(), headers=AUTH)
        assert resp.get_json() == {'success': True, 'changed': True}

    def test_needs_admin_json_and_a_sane_name(self, client):
        user = dict(AUTH, **{'Remote-Groups': 'staff'})
        with patch.object(gandalf, '_cfg', dict(CONFIG, auth=dict(CONFIG['auth'], allowed_groups=['admin', 'staff']))):
            assert client.put(self._url(), json={}, headers=user).status_code == 403
        assert client.put(self._url(), data='x', headers=AUTH).status_code == 400
        assert client.put(self._url('bad name!'), json={}, headers=AUTH).status_code in (400, 404)

    def test_device_api_lists_them(self, client):
        gandalf._cutsheet_schema_ready = True
        gandalf.db.get_unused_interfaces.return_value = {'large1': {'enp36s0': {'reason': 'spare'}}}
        data = client.get('/api/device/large1', headers=AUTH).get_json()
        assert data['unused_interfaces'] == {'enp36s0': {'reason': 'spare'}}


class TestMaintenance:
    SUP = {'id': 7, 'target_type': 'all', 'target_name': '', 'reason': 'UniFi Network update',
           'suppressed_by': 'jared', 'expires_at': '2026-11-12T20:00:00Z'}

    def test_no_banner_and_a_start_button_by_default(self, client):
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        assert 'maint-banner' not in html
        assert 'data-action="start-maintenance"' in html and 'id="maint-modal"' in html

    def test_banner_on_every_page_while_on(self, client):
        gandalf.db.get_active_suppressions.return_value = [
            {'id': 3, 'target_type': 'host', 'target_name': 'large1', 'reason': 'x', 'expires_at': None},
            self.SUP,
        ]
        for page in ('/', '/events', '/rack'):
            html = client.get(page, headers=AUTH).get_data(as_text=True)
            assert 'class="maint-banner"' in html, page
            assert 'datetime="2026-11-12T20:00:00Z"' in html and 'UniFi Network update' in html
            assert 'data-sup-id="7"' in html and 'data-action="start-maintenance"' not in html

    def test_open_ended_maintenance_wins(self, client):
        gandalf.db.get_active_suppressions.return_value = [
            self.SUP, dict(self.SUP, id=9, expires_at=None, reason='Rewiring')]
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        assert 'until it is ended' in html and 'data-sup-id="9"' in html

    def test_viewers_see_the_banner_without_buttons(self, client):
        gandalf.db.get_active_suppressions.return_value = [self.SUP]
        viewer = dict(AUTH, **{'Remote-Groups': 'staff'})
        with patch.object(gandalf, '_cfg', dict(CONFIG, auth=dict(CONFIG['auth'], allowed_groups=['admin', 'staff']))):
            html = client.get('/', headers=viewer).get_data(as_text=True)
        assert 'class="maint-banner"' in html and 'data-action="end-maintenance"' not in html

    def test_a_db_error_means_no_banner(self, client):
        gandalf.db.get_active_suppressions.side_effect = RuntimeError('db down')
        assert gandalf.inject_maintenance() == {'maintenance': None}


class TestFollowUp:
    def _state(self, tickets, age_s=60):
        state = dict(STATE, open_tickets=json.dumps({'checked_at': time.time() - age_s, 'tickets': tickets}))
        gandalf.db.get_state.side_effect = lambda key, default=None: state.get(key, default)

    def _event(self, days_old, ticket_id=None):
        first = (datetime.now(timezone.utc) - timedelta(days=days_old)).strftime('%Y-%m-%dT%H:%M:%SZ')
        return {'id': 1, 'event_type': 'interface_down', 'severity': 'critical', 'target_name': 'micro1',
                'target_detail': 'enp3s0', 'description': 'down', 'first_seen': first, 'last_seen': first,
                'consecutive_failures': 3, 'ticket_id': ticket_id}

    def test_old_alert_is_stale_and_closed_ticket_is_flagged(self, client):
        self._state({'9': 'Open'})
        gandalf.db.get_active_events.return_value = [self._event(5, '8'), self._event(1, '9')]
        gandalf.db.get_cleared_tickets.return_value = []
        events = client.get('/api/status', headers=AUTH).get_json()['events']
        assert events[0]['stale_days'] == 5 and events[0]['ticket_closed'] is True
        assert events[1]['stale_days'] is None and events[1]['ticket_closed'] is False

    def test_threshold_comes_from_config(self, client):
        self._state({})
        gandalf.db.get_active_events.return_value = [self._event(1.5)]
        gandalf.db.get_cleared_tickets.return_value = []
        cfg = dict(CONFIG, monitor={'stale_alert_days': 1})
        with patch.object(gandalf, '_cfg', cfg):
            events = client.get('/api/status', headers=AUTH).get_json()['events']
        assert events[0]['stale_days'] == 1

    def test_tickets_left_open_are_listed(self, client):
        self._state({'281331508': 'Open'})
        left = [{'ticket_id': '281331508', 'event_type': 'host_unreachable', 'target_name': 'pbs',
                 'target_detail': '', 'severity': 'critical', 'first_seen': '2026-10-14T12:02:05Z',
                 'cleared_at': '2026-11-04T22:12:36Z', 'events': 3}]
        gandalf.db.get_cleared_tickets.return_value = left
        assert client.get('/api/status', headers=AUTH).get_json()['stale_tickets'] == left
        gandalf.db.get_cleared_tickets.assert_called_with(['281331508'], older_than_hours=24)
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        assert '#281331508' in html and 'TO FOLLOW UP' in html
        assert '<section class="g-section" id="followup">' in html

    def test_out_of_date_list_flags_nothing(self, client):
        self._state({}, age_s=3 * 3600)
        gandalf.db.get_active_events.return_value = [self._event(1, '8')]
        body = client.get('/api/status', headers=AUTH).get_json()
        assert body['stale_tickets'] == [] and body['events'][0]['ticket_closed'] is False
        gandalf.db.get_cleared_tickets.assert_not_called()

    def test_no_list_yet_hides_the_panel(self, client):
        html = client.get('/', headers=AUTH).get_data(as_text=True)
        assert '<section class="g-section" id="followup" hidden>' in html
        assert 'TO FOLLOW UP' not in html


class TestTopology:
    def test_page_needs_auth(self, client):
        assert client.get('/topology').status_code == 401
        assert client.get('/api/topology').status_code == 401

    def test_page_loads_the_drawings_first(self, client):
        html = client.get('/topology', headers=AUTH).get_data(as_text=True)
        assert 'id="tp-canvas"' in html
        assert html.index('faceplates.js') < html.index('fronts.js') < html.index('topology.js')

    def test_api_returns_the_graph(self, client):
        gandalf._cutsheet_schema_ready = True
        with patch.object(gandalf.db, 'cutsheet_load', return_value={'items': [], 'cables': []}):
            data = client.get('/api/topology', headers=AUTH).get_json()
        assert {'nodes', 'links', 'warnings', 'updated'} <= set(data)
        names = {n['name'] for n in data['nodes']}
        assert {'large1', 'USW Flex', 'Internet'} <= names

    def test_api_survives_a_cutsheet_failure(self, client):
        gandalf._cutsheet_schema_ready = True
        with patch.object(gandalf.db, 'cutsheet_load', side_effect=RuntimeError('db down')):
            assert client.get('/api/topology', headers=AUTH).status_code == 200

    def test_suppressed_alerts_are_not_drawn(self, client):
        gandalf._cutsheet_schema_ready = True
        ev = {'id': 1, 'event_type': 'host_down', 'severity': 'critical', 'target_name': 'large1',
              'target_detail': '', 'description': ''}
        with patch.object(gandalf.db, 'cutsheet_load', return_value={'items': [], 'cables': []}), \
                patch.object(gandalf.db, 'get_active_events', return_value=[ev]), \
                patch.object(gandalf.db, 'check_suppressed', return_value=True):
            data = client.get('/api/topology', headers=AUTH).get_json()
        assert next(n for n in data['nodes'] if n['name'] == 'large1')['alert_count'] == 0