gandalf/tests/test_app.py

"""Tests for app.py routes: database and LDAP are mocked."""
import json
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 = ''
        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', 'Network Topology',
        )]
        assert order == sorted(order)

    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')])
    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 'switch' not in sw  # no port data in this snapshot
        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