jaredandClaude Opus 5.5
9d561db703
Lint / JS (eslint) (push) Successful in 6s
Lint / Python (flake8) (push) Successful in 1m56s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 2m3s
Test / Python Tests (pytest) (push) Successful in 1m0s
Lint / Python (flake8) (pull_request) Successful in 48s
Lint / JS (eslint) (pull_request) Successful in 6s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 2m7s
Test / Python Tests (pytest) (pull_request) Successful in 2m6s
/events lists every alert from the last 24 hours to 90 days, open or cleared, with how long each lasted, failed checks and its ticket. Filters (device, type, severity, state, range) live in the URL, so device pages and tickets link straight to a device's history. A "Recurring" table lists alerts that fired more than once on the same target. On phones the rows become cards. /api/events/history backs it (100 per page). The existing /api/events is unchanged. /health and the dashboard's monitor indicator now call the monitor stalled after five missed polls (at least 10 minutes) instead of a fixed 20, and /health reports the monitor's age. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
379 lines
18 KiB
Python
379 lines
18 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
|
|
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
|