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
        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'] == []