Files
gandalf/tests/test_app.py
T
jaredandClaude Opus 5.5 f655836446
Lint / JS (eslint) (push) Successful in 7s
Lint / Python (flake8) (push) Successful in 50s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 50s
Test / Python Tests (pytest) (push) Successful in 1m2s
Lint / Python (flake8) (pull_request) Successful in 52s
Lint / JS (eslint) (pull_request) Successful in 7s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 1m42s
Test / Python Tests (pytest) (pull_request) Successful in 2m2s
Add a rack drawing (/rack)
The rack front is drawn from the rack units on cutsheet items. Switches
with a faceplate use it, with live port state; patch panels get a generic
panel with port labels; everything else is a labelled plate. Each
documented port shows its cutsheet state, and hovering a port lights up
every port on the same cable run (through patch panels) and names the
cables. Clicking opens the switch port or the cable.

Overlapping items and items past the top of the rack are listed instead
of drawn over another; where two claim the same top unit, the first one
recorded wins. rack.units sets the height (default 42).

/api/rack builds on the cutsheet view; cutsheet.rack_layout and
cutsheet.port_marks hold the placement and per-port logic.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-10 20:59:30 -05:00

471 lines
23 KiB
Python

"""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