diff --git a/README.md b/README.md index 1e23e1f..a4c3329 100644 --- a/README.md +++ b/README.md @@ -189,6 +189,31 @@ neighbors that can't be matched show their MAC. Bonded NICs share a MAC, so only Port LEDs use the colors from each model's quick start guide. Other models use plain blocks until they get an entry in `FACEPLATES`. +### Cutsheet (`/cutsheet`) +Where every cable is meant to go, checked against what Gandalf actually sees. Anyone who can log in +can view it; only the `admin` group can edit, and every change is logged with who made it and when. + +- **Inventory**: things discovery can't see: patch panels, wall jacks, shelves, and rack positions + for devices (use a device's exact Gandalf name). Panels and jacks are passive: a cable on a + port's front continues through the same port's rear. +- **Cables**: two endpoints each (device or item, port, and front/rear for panels and jacks), plus + ID, type, length, color, status (`planned`, `installed`, `dark`, `retired`) and notes. Leave the + far end blank for a run that ends nowhere yet, such as an unterminated fibre. +- **Paths**: each cable shows its full run through any panels and jacks, e.g. switch port 13 → + panel port 5 → bedroom jack. +- **Plan vs. actual**: each installed run ending on a switch or router port is compared with that + port's discovered neighbor (LLDP, the UniFi client table, or a DAC's serial reported at both + ends): *matches*, *wrong neighbor*, *no link*, *missing port*, or *port conflict* (two installed + cables in one port). Runs that end at a jack are *unverified* while linked and *idle* when not. + Planned, dark and retired cables aren't checked. +- **Undocumented links**: live switch ports no cable mentions, with an Add button that pre-fills + what Gandalf knows. Each switch port panel also shows its planned cable, or links to add one. +- **Labels to make**: cable IDs (one per cord end) and panel-port labels not yet stuck on, sized for + typing on a handheld label maker. Tick each off once it's applied; **Print list** prints just this. +- **CSV**: export, and import that adds or updates cables by ID. A file with any invalid row imports + nothing. +- Cables, panels and jacks are included in the dashboard search. + ### Suppressions (`/suppressions`) - Create timed (30 min / 1 hr / 4 hr / 8 hr) or manual suppressions - Target types: host, interface, UniFi device, or global @@ -528,6 +553,7 @@ gandalf/ ├── monitor.py # Background daemon (Prometheus, UniFi, Pulse, alert logic) ├── modem.py # Read-only AT&T gateway status page client and parser ├── devices.py # Device inventory for search and device pages +├── cutsheet.py # Cutsheet validation, path tracing, plan-vs-actual checks, CSV ├── db.py # Database operations (MariaDB via pymysql, thread-local conn reuse) ├── schema.sql # Database schema (network_events, suppression_rules, monitor_state) ├── config.json # Runtime configuration (not committed with secrets) @@ -536,11 +562,13 @@ gandalf/ │ ├── style.css # Terminal aesthetic CSS (CRT scanlines, green-on-black) │ ├── app.js # Dashboard JS (auto-refresh, events, suppress modal, search, modem panel) │ ├── device.js # Device page rendering +│ ├── cutsheet.js # Cutsheet page │ └── port-inspector.js # Switch front panel, port panel, path check, diagnostics └── templates/ ├── base.html # Shared layout (header, nav, footer) ├── index.html # Dashboard page ├── device.html # Per-device page + ├── cutsheet.html # Cutsheet page and its editors ├── _modem_panel.html # WAN / Fiber panel, shared by the dashboard and modem page └── suppressions.html # Suppression management page ``` @@ -582,6 +610,9 @@ Take LED colors from the model's quick start guide. See the comment at the top o - `monitor_state` also holds `wan_history` (WAN outages, modem restarts) and `unifi_log_cursor` (how far the UniFi log has been read) - Database sessions run in UTC (`db.py`), whatever the server's time zone - `modem_samples`: optical readings per poll, kept 90 days; created automatically at monitor startup +- `cutsheet_items`, `cutsheet_cables`, `cutsheet_port_labels`, `cutsheet_history`: cutsheet records and + their change log, created automatically on first use. Cables refer to devices and items by name; renaming + an item updates its cables, but a device renamed in UniFi needs its cables edited to match ### Security Notes diff --git a/app.py b/app.py index 5fde4ed..26a22e1 100644 --- a/app.py +++ b/app.py @@ -20,6 +20,7 @@ from functools import wraps from flask import Flask, jsonify, make_response, redirect, render_template, request, send_file, url_for +import cutsheet import db import devices import diagnose @@ -409,10 +410,38 @@ def api_modem(): @app.route('/api/devices') @require_auth def api_devices(): - """Device list for the home page search.""" - return jsonify({'devices': devices.build_index( + """Device list for the home page search, plus cutsheet panels, jacks and cables.""" + found = devices.build_index( _state_json('network_snapshot'), _state_json('link_stats'), _state_json('modem_stats'), - )}) + ) + known = {d['name'] for d in found} + try: + _ensure_cutsheet_schema() + data = db.cutsheet_load() + except Exception as e: + logger.error(f'Cutsheet unavailable for search: {e}') + data = {'items': [], 'cables': []} + for item in data['items']: + if item['name'] in known: + continue + kind = item['kind'].replace('_', ' ') + found.append({ + 'name': item['name'], 'kind': kind, 'status': 'unknown', 'ip': '', 'mac': '', 'model': '', + 'url': f'/cutsheet#item-{item["id"]}', 'interface_count': item.get('port_count') or 0, + 'search': ' '.join(filter(None, [item['name'], kind, item.get('location'), 'cutsheet'])).lower(), + }) + for cable in data['cables']: + if not cable.get('label'): + continue + ends = ' '.join(filter(None, [cable['a_device'], cable['b_device']])) + found.append({ + 'name': cable['label'], 'kind': 'cable', 'status': 'unknown', 'ip': '', 'mac': '', + 'model': f'{cable["a_device"]} ↔ {cable["b_device"] or "unterminated"}', + 'url': f'/cutsheet#cable-{cable["id"]}', 'interface_count': 0, + 'search': ' '.join(filter(None, [cable['label'], ends, cable.get('cable_type'), + cable.get('color'), 'cable cutsheet'])).lower(), + }) + return jsonify({'devices': found}) @app.route('/api/device/') @@ -427,6 +456,11 @@ def api_device(name: str): events = db.get_events_for_target(name) _annotate_suppressions(events, db.get_active_suppressions()) detail['events'] = events + try: + detail['cutsheet'] = _cutsheet_for_device(name) + except Exception as e: + logger.error(f'Cutsheet unavailable for {name}: {e}') + detail['cutsheet'] = [] return jsonify(detail) @@ -769,6 +803,247 @@ def api_avatar(): return resp +# --------------------------------------------------------------------------- +# Cutsheets +# --------------------------------------------------------------------------- + +_cutsheet_schema_ready = False +_cutsheet_schema_lock = threading.Lock() + + +def _ensure_cutsheet_schema() -> None: + """Create the cutsheet tables on first use, so deploys need no migration step.""" + global _cutsheet_schema_ready + if _cutsheet_schema_ready: + return + with _cutsheet_schema_lock: + if not _cutsheet_schema_ready: + db.ensure_cutsheet_schema() + _cutsheet_schema_ready = True + + +def _node_json(node) -> dict: + if node is None: + return {'device': '', 'port': '', 'side': '', 'text': 'unterminated'} + return {'device': node[0], 'port': node[1], 'side': node[2], 'text': cutsheet.describe_node(node)} + + +def _cutsheet_catalog(link_stats: dict, items: list) -> list: + """Devices and their ports, for the cable form's suggestions.""" + catalog = [] + for name, sw in sorted((link_stats.get('unifi_switches') or {}).items()): + ports = sorted((p for p in (sw.get('ports') or {}).items() if p[1].get('port_idx') is not None), + key=lambda kv: kv[1]['port_idx']) + catalog.append({'name': name, 'kind': 'switch', 'ports': [ + {'value': str(p['port_idx']), 'text': f'{p["port_idx"]} · {pname}'} for pname, p in ports]}) + for name, ifaces in sorted((link_stats.get('hosts') or {}).items()): + catalog.append({'name': name, 'kind': 'host', + 'ports': [{'value': i, 'text': i} for i in sorted(ifaces)]}) + known = {c['name'] for c in catalog} + for item in items: + if item['name'] in known: + continue + ports = [{'value': str(n), 'text': str(n)} for n in range(1, (item.get('port_count') or 0) + 1)] + catalog.append({'name': item['name'], 'kind': item['kind'], 'ports': ports}) + return catalog + + +def _cutsheet_view() -> dict: + """Every cutsheet record with its traced path and plan-vs-actual verdict.""" + _ensure_cutsheet_schema() + data = db.cutsheet_load() + link_stats = _state_json('link_stats') + switches = link_stats.get('unifi_switches') or {} + items = {i['name']: i for i in data['items']} + results = cutsheet.evaluate(data['cables'], items, switches) + for cable in data['cables']: + r = results[cable['id']] + cable['verdict'] = r['verdict'] + cable['detail'] = r['detail'] + cable['problem'] = r['verdict'] in cutsheet.PROBLEM_VERDICTS + cable['path'] = [ + {'cable_id': h} if isinstance(h, int) else _node_json(h) for h in r['path']['hops'] + ] + cable['a_text'] = cutsheet.describe_node(cutsheet._node(cable, 'a')) + cable['b_text'] = cutsheet.describe_node(cutsheet._node(cable, 'b')) + data['suggestions'] = cutsheet.undocumented(data['cables'], switches) + data['labels'] = cutsheet.labels_to_make(data['cables'], items, data['port_labels']) + data['next_label'] = cutsheet.next_label(data['cables']) + data['catalog'] = _cutsheet_catalog(link_stats, data['items']) + data['verdicts'] = cutsheet.VERDICTS + data['options'] = {'kinds': cutsheet.ITEM_KINDS, 'passive_kinds': cutsheet.PASSIVE_KINDS, + 'statuses': cutsheet.STATUSES, 'cable_types': cutsheet.CABLE_TYPES} + data['updated'] = link_stats.get('updated') + return data + + +def _cutsheet_for_device(name: str) -> list: + """The cables that end on one device, for its port panel.""" + view = _cutsheet_view() + out = [] + for cable in view['cables']: + for end in ('a', 'b'): + if cable[f'{end}_device'] == name and cable['status'] != 'retired': + out.append({ + 'port': cable[f'{end}_port'], 'side': cable[f'{end}_side'], 'id': cable['id'], + 'label': cable['label'], 'status': cable['status'], 'verdict': cable['verdict'], + 'detail': cable['detail'], 'problem': cable['problem'], 'path': cable['path'], + }) + return out + + +def _json_body() -> dict: + """The request's JSON object. Requiring a JSON content type also keeps another + site's plain form post from reaching these write endpoints (no CORS preflight).""" + if not request.is_json: + raise cutsheet.ValidationError('Send the request as JSON') + data = request.get_json(silent=True) + if not isinstance(data, dict): + raise cutsheet.ValidationError('Request body must be a JSON object') + return data + + +def _cutsheet_write(fn): + """Turn validation and conflict errors from a write into 400 and 409 responses.""" + @wraps(fn) + def wrapper(*args, **kwargs): + try: + _ensure_cutsheet_schema() + return fn(*args, **kwargs) + except cutsheet.ValidationError as e: + return jsonify({'error': str(e)}), 400 + except db.CutsheetConflict as e: + return jsonify({'error': str(e)}), 409 + return wrapper + + +def _cutsheet_items() -> dict: + return {i['name']: i for i in db.cutsheet_load()['items']} + + +@app.route('/cutsheet') +@require_auth +def cutsheet_page(): + user = _get_user() + return render_template('cutsheet.html', user=user, can_edit='admin' in user['groups']) + + +@app.route('/api/cutsheet') +@require_auth +def api_cutsheet(): + view = _cutsheet_view() + view['history'] = db.cutsheet_history(limit=100) + return jsonify(view) + + +@app.route('/api/cutsheet/cables', methods=['POST']) +@require_auth +@require_admin +@_cutsheet_write +def api_cutsheet_create_cable(): + cable = cutsheet.validate_cable(_json_body(), _cutsheet_items()) + cable_id = db.cutsheet_create_cable(cable, _get_user()['username']) + return jsonify({'success': True, 'id': cable_id}) + + +@app.route('/api/cutsheet/cables/', methods=['PUT']) +@require_auth +@require_admin +@_cutsheet_write +def api_cutsheet_update_cable(cable_id: int): + cable = cutsheet.validate_cable(_json_body(), _cutsheet_items()) + if not db.cutsheet_update_cable(cable_id, cable, _get_user()['username']): + return jsonify({'error': 'No such cable'}), 404 + return jsonify({'success': True}) + + +@app.route('/api/cutsheet/cables/', methods=['DELETE']) +@require_auth +@require_admin +@_cutsheet_write +def api_cutsheet_delete_cable(cable_id: int): + if not db.cutsheet_delete_cable(cable_id, _get_user()['username']): + return jsonify({'error': 'No such cable'}), 404 + return jsonify({'success': True}) + + +@app.route('/api/cutsheet/items', methods=['POST']) +@require_auth +@require_admin +@_cutsheet_write +def api_cutsheet_create_item(): + item = cutsheet.validate_item(_json_body()) + item_id = db.cutsheet_create_item(item, _get_user()['username']) + return jsonify({'success': True, 'id': item_id}) + + +@app.route('/api/cutsheet/items/', methods=['PUT']) +@require_auth +@require_admin +@_cutsheet_write +def api_cutsheet_update_item(item_id: int): + item = cutsheet.validate_item(_json_body()) + if not db.cutsheet_update_item(item_id, item, _get_user()['username']): + return jsonify({'error': 'No such item'}), 404 + return jsonify({'success': True}) + + +@app.route('/api/cutsheet/items/', methods=['DELETE']) +@require_auth +@require_admin +@_cutsheet_write +def api_cutsheet_delete_item(item_id: int): + if not db.cutsheet_delete_item(item_id, _get_user()['username']): + return jsonify({'error': 'No such item'}), 404 + return jsonify({'success': True}) + + +@app.route('/api/cutsheet/items//ports/', methods=['PUT']) +@require_auth +@require_admin +@_cutsheet_write +def api_cutsheet_port_label(item_id: int, port: str): + data = _json_body() + label = str(data.get('label') or '').strip() + if len(label) > 64: + raise cutsheet.ValidationError('Port label must be 64 characters or fewer') + if len(port) > cutsheet.PORT_MAX: + raise cutsheet.ValidationError('Port is too long') + if not db.cutsheet_set_port_label(item_id, port, label, bool(data.get('applied')), _get_user()['username']): + return jsonify({'error': 'No such item'}), 404 + return jsonify({'success': True}) + + +@app.route('/api/cutsheet/cables.csv') +@require_auth +def api_cutsheet_export(): + _ensure_cutsheet_schema() + resp = make_response(cutsheet.cables_to_csv(db.cutsheet_load()['cables'])) + resp.headers['Content-Type'] = 'text/csv; charset=utf-8' + resp.headers['Content-Disposition'] = 'attachment; filename="gandalf-cutsheet.csv"' + return resp + + +@app.route('/api/cutsheet/import', methods=['POST']) +@require_auth +@require_admin +@_cutsheet_write +def api_cutsheet_import(): + data = _json_body() + text = data.get('csv') + if not isinstance(text, str) or not text.strip(): + raise cutsheet.ValidationError('Paste or choose a CSV file') + if len(text) > 1_000_000: + raise cutsheet.ValidationError('CSV is larger than 1 MB') + cables, errors = cutsheet.parse_csv(text, _cutsheet_items()) + if errors: + return jsonify({'error': 'Nothing was imported', 'errors': errors[:50]}), 400 + if data.get('dry_run'): + return jsonify({'success': True, 'dry_run': True, 'rows': len(cables)}) + result = db.cutsheet_import_cables(cables, _get_user()['username']) + return jsonify(dict(result, success=True)) + + @app.route('/health') def health(): """Health check endpoint (no auth). Checks DB and monitor freshness.""" diff --git a/cutsheet.py b/cutsheet.py new file mode 100644 index 0000000..1c1c3c1 --- /dev/null +++ b/cutsheet.py @@ -0,0 +1,471 @@ +"""Cutsheets: where every cable is meant to go, compared with what Gandalf sees. + +Pure functions with no I/O, like devices.py. app.py loads rows from the +database and the monitor's cached link stats and passes them in. + +Model +----- +* An **item** is something Gandalf can't discover: a patch panel, a wall jack, + a shelf, or a rack position for a device. Items of a passive kind (panels and + jacks) pass a signal straight through, from a port's front to the same port's + rear. +* A **cable** joins two endpoints. An endpoint is a device or item name, a port + (a switch's port number, a host's interface name, or a panel port), and for + passive items a side. The far end may be left blank for a cable that ends + nowhere yet, such as an unterminated fibre. +* A **path** is the chain of cables joined through passive items, e.g. switch + port 13 → panel port 5 (front) … panel port 5 (rear) → bedroom jack. +""" +import csv +import io +import re +from typing import Dict, List, Optional, Tuple + +ITEM_KINDS = ('patch_panel', 'wall_jack', 'device', 'shelf', 'other') +PASSIVE_KINDS = ('patch_panel', 'wall_jack') +SIDES = ('front', 'rear') +STATUSES = ('planned', 'installed', 'dark', 'retired') +CABLE_TYPES = ('Cat5e', 'Cat6', 'Cat6a', 'DAC', 'AOC', 'OM3', 'OM4', 'OS2', 'Power', 'Other') + +LABEL_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._-]{0,31}$') +NAME_MAX = 100 +PORT_MAX = 64 +TEXT_MAX = 255 +NOTES_MAX = 2000 + +# Comparison outcomes, and whether each counts as a problem +VERDICTS = { + 'match': 'Matches what Gandalf sees', + 'wrong_neighbor': 'A different device is connected', + 'no_link': 'Planned device is not linked', + 'missing_port': 'That port is not reported by the device', + 'unverified': "Gandalf can't confirm the far end", + 'idle': 'Ends at a jack or panel with nothing linked', + 'not_checked': 'No switch port to check against', + 'planned': 'Planned, not installed yet', + 'dark': 'Unused (dark)', + 'retired': 'Retired', + 'conflict': 'Another cable uses the same port', +} +PROBLEM_VERDICTS = ('wrong_neighbor', 'no_link', 'missing_port', 'conflict') + + +class ValidationError(ValueError): + """A request the cutsheet can't accept, with a message for the user.""" + + +# ── Validation ────────────────────────────────────────────────────────────── + +def _text(data: dict, key: str, limit: int = TEXT_MAX, required: bool = False, label: str = '') -> str: + value = data.get(key) + value = '' if value is None else str(value).strip() + if required and not value: + raise ValidationError(f'{label or key} is required') + if len(value) > limit: + raise ValidationError(f'{label or key} must be {limit} characters or fewer') + return value + + +def _int(data: dict, key: str, low: int, high: int, label: str) -> Optional[int]: + value = data.get(key) + if value in (None, ''): + return None + try: + number = int(value) + except (TypeError, ValueError): + raise ValidationError(f'{label} must be a whole number') + if not low <= number <= high: + raise ValidationError(f'{label} must be between {low} and {high}') + return number + + +def validate_item(data: dict) -> dict: + kind = _text(data, 'kind', 20) or 'other' + if kind not in ITEM_KINDS: + raise ValidationError(f'kind must be one of {", ".join(ITEM_KINDS)}') + item = { + 'name': _text(data, 'name', NAME_MAX, required=True, label='Name'), + 'kind': kind, + 'port_count': _int(data, 'port_count', 1, 512, 'Port count'), + 'location': _text(data, 'location'), + 'rack_u': _int(data, 'rack_u', 1, 60, 'Rack unit'), + 'rack_height': _int(data, 'rack_height', 1, 20, 'Height'), + 'notes': _text(data, 'notes', NOTES_MAX), + } + if item['rack_height'] and not item['rack_u']: + raise ValidationError('Set a rack unit to go with the height') + if item['rack_u'] and not item['rack_height']: + item['rack_height'] = 1 + if kind == 'patch_panel' and not item['port_count']: + raise ValidationError('A patch panel needs a port count') + return item + + +def _endpoint(data: dict, prefix: str, items: Dict[str, dict], required: bool) -> Tuple[str, str, str]: + device = _text(data, f'{prefix}_device', NAME_MAX, required=required, + label='From device' if prefix == 'a' else 'To device') + port = _text(data, f'{prefix}_port', PORT_MAX) + side = _text(data, f'{prefix}_side', 10) + if not device: + if port or side: + raise ValidationError('Choose a device before a port') + return '', '', '' + item = items.get(device) + passive = bool(item and item['kind'] in PASSIVE_KINDS) + if passive: + if side not in SIDES: + raise ValidationError(f'Choose the front or rear of {device}') + if not port: + raise ValidationError(f'Choose a port on {device}') + count = item.get('port_count') + if count and (not port.isdigit() or not 1 <= int(port) <= count): + raise ValidationError(f'{device} has ports 1 to {count}') + else: + side = '' + return device, port, side + + +def validate_cable(data: dict, items: Dict[str, dict]) -> dict: + label = _text(data, 'label', 32) + if label and not LABEL_RE.match(label): + raise ValidationError('Label may use letters, digits, dot, dash and underscore (up to 32)') + status = _text(data, 'status', 20) or 'installed' + if status not in STATUSES: + raise ValidationError(f'status must be one of {", ".join(STATUSES)}') + cable_type = _text(data, 'cable_type', 32) + if cable_type and cable_type not in CABLE_TYPES: + raise ValidationError(f'Type must be one of {", ".join(CABLE_TYPES)}') + length = data.get('length_m') + if length in (None, ''): + length = None + else: + try: + length = round(float(length), 2) + except (TypeError, ValueError): + raise ValidationError('Length must be a number of metres') + if not 0 < length <= 999: + raise ValidationError('Length must be between 0 and 999 metres') + a = _endpoint(data, 'a', items, required=True) + b = _endpoint(data, 'b', items, required=False) + if a == b: + raise ValidationError('Both ends are the same port') + return { + 'label': label or None, + 'a_device': a[0], 'a_port': a[1], 'a_side': a[2], + 'b_device': b[0], 'b_port': b[1], 'b_side': b[2], + 'cable_type': cable_type, 'length_m': length, + 'color': _text(data, 'color', 32), 'status': status, + 'notes': _text(data, 'notes', NOTES_MAX), + 'label_applied': bool(data.get('label_applied')), + } + + +def next_label(cables: List[dict], prefix: str = 'C') -> str: + """The next free cable ID: C01, C02, ... (wider once past 99).""" + used = set() + for c in cables: + m = re.match(rf'^{re.escape(prefix)}(\d+)$', c.get('label') or '', re.IGNORECASE) + if m: + used.add(int(m.group(1))) + n = 1 + while n in used: + n += 1 + return f'{prefix}{n:02d}' + + +# ── Paths ─────────────────────────────────────────────────────────────────── + +def _node(cable: dict, end: str) -> Optional[tuple]: + device = cable.get(f'{end}_device') or '' + if not device: + return None + return (device, cable.get(f'{end}_port') or '', cable.get(f'{end}_side') or '') + + +def _other_side(node: tuple, items: Dict[str, dict]) -> Optional[tuple]: + """Through a passive item, the same port's other side; otherwise the path ends.""" + item = items.get(node[0]) + if not item or item['kind'] not in PASSIVE_KINDS or node[2] not in SIDES: + return None + return (node[0], node[1], 'rear' if node[2] == 'front' else 'front') + + +def _live(cable: dict) -> bool: + return cable.get('status') != 'retired' + + +def node_index(cables: List[dict]) -> Dict[tuple, List[dict]]: + """Map each endpoint to the live cables plugged into it.""" + index: Dict[tuple, List[dict]] = {} + for c in cables: + if not _live(c): + continue + for end in ('a', 'b'): + node = _node(c, end) + if node: + index.setdefault(node, []).append(c) + return index + + +def trace(cable: dict, cables: List[dict], items: Dict[str, dict], + index: Optional[Dict[tuple, List[dict]]] = None) -> dict: + """Follow a cable through passive items in both directions. + + Returns {'hops': [node, cable_id, node, ...], 'cable_ids': [...]} with the + hops ordered from one end of the run to the other. A node is + (device, port, side), or None for an unterminated end. + """ + index = index if index is not None else node_index(cables) + + def walk(node: Optional[tuple], came_from: dict, seen: set) -> list: + hops = [] + while node is not None: + through = _other_side(node, items) + if through is None: + break + nxt = next((c for c in index.get(through, []) if c['id'] != came_from['id'] and c['id'] not in seen), + None) + hops.append(through) + if nxt is None: + break + seen.add(nxt['id']) + hops.append(nxt['id']) + far = 'b' if _node(nxt, 'a') == through else 'a' + node = _node(nxt, far) + hops.append(node) + came_from = nxt + return hops + + seen = {cable['id']} + toward_a = walk(_node(cable, 'a'), cable, seen) + toward_b = walk(_node(cable, 'b'), cable, seen) + hops = list(reversed(toward_a)) + [_node(cable, 'a'), cable['id'], _node(cable, 'b')] + toward_b + return {'hops': hops, 'cable_ids': [h for h in hops if isinstance(h, int)]} + + +# ── Plan vs. actual ───────────────────────────────────────────────────────── + +def _switch_port(node: Optional[tuple], switches: dict) -> Optional[dict]: + """The discovered port behind an endpoint, if it is a switch or router port.""" + if not node or node[0] not in switches or not node[1].isdigit(): + return None + for port in (switches[node[0]].get('ports') or {}).values(): + if port.get('port_idx') == int(node[1]): + return port + return {} # the device is known but doesn't report this port + + +def _same(a: str, b: str) -> bool: + return bool(a) and bool(b) and a.strip().lower() == b.strip().lower() + + +def _check_end(node: tuple, other: Optional[tuple], switches: dict, items: Dict[str, dict]) -> Optional[dict]: + port = _switch_port(node, switches) + if port is None: + return None + if port == {}: + return {'verdict': 'missing_port', 'detail': f'{node[0]} reports no port {node[1]}'} + other_passive = other is None or (items.get(other[0], {}).get('kind') in PASSIVE_KINDS) + expected = None if other_passive else other[0] + lldp = port.get('lldp') or {} + seen = lldp.get('system_name') or '' + sfp_peer = (port.get('sfp') or {}).get('cable_peer') + + if sfp_peer and other is not None and not other_passive: + if _same(sfp_peer.get('switch', ''), other[0]) and str(sfp_peer.get('port_idx')) == other[1]: + return {'verdict': 'match', 'detail': 'Both ends report the same DAC serial'} + if not port.get('up'): + if expected: + return {'verdict': 'no_link', 'detail': f'{node[0]} port {node[1]} is down'} + return {'verdict': 'idle', 'detail': f'{node[0]} port {node[1]} is down'} + if expected: + if not seen: + return {'verdict': 'unverified', 'detail': f'Link up on port {node[1]}, but no neighbor is identified'} + if _same(seen, expected): + return {'verdict': 'match', 'detail': f'{node[0]} port {node[1]} sees {seen}'} + return {'verdict': 'wrong_neighbor', 'detail': f'{node[0]} port {node[1]} sees {seen}, not {expected}'} + detail = f'Link up on port {node[1]}' + (f', sees {seen}' if seen else '') + return {'verdict': 'unverified', 'detail': detail} + + +_VERDICT_RANK = ['wrong_neighbor', 'no_link', 'missing_port', 'unverified', 'idle', 'match'] + + +def compare_path(path: dict, cables_by_id: Dict[int, dict], switches: dict, items: Dict[str, dict]) -> dict: + """Check a traced run against discovery. Returns {'verdict', 'detail'}.""" + run = [cables_by_id[i] for i in path['cable_ids']] + statuses = {c['status'] for c in run} + for status in ('retired', 'dark', 'planned'): + if status in statuses: + return {'verdict': status, 'detail': VERDICTS[status]} + ends = (path['hops'][0], path['hops'][-1]) + checks = [c for c in (_check_end(ends[0], ends[1], switches, items), + _check_end(ends[1], ends[0], switches, items)) if c] + if not checks: + return {'verdict': 'not_checked', 'detail': VERDICTS['not_checked']} + # A switch-to-switch run is checked from both ends: report the worst + return min(checks, key=lambda c: _VERDICT_RANK.index(c['verdict'])) + + +def evaluate(cables: List[dict], items: Dict[str, dict], switches: dict) -> Dict[int, dict]: + """Trace and compare every cable. Returns {cable_id: {'path', 'verdict', 'detail'}}.""" + switches = switches or {} + index = node_index(cables) + by_id = {c['id']: c for c in cables} + result: Dict[int, dict] = {} + for c in cables: + if not _live(c): + result[c['id']] = {'path': {'hops': [_node(c, 'a'), c['id'], _node(c, 'b')], 'cable_ids': [c['id']]}, + 'verdict': 'retired', 'detail': VERDICTS['retired']} + continue + conflict = next((n for n in (_node(c, 'a'), _node(c, 'b')) + if n and len([x for x in index.get(n, []) if x['status'] in ('installed', 'dark')]) > 1 + and c['status'] in ('installed', 'dark')), None) + path = trace(c, cables, items, index) + if conflict: + others = [x.get('label') or f'#{x["id"]}' for x in index[conflict] if x['id'] != c['id']] + result[c['id']] = {'path': path, 'verdict': 'conflict', + 'detail': f'{describe_node(conflict)} is also used by {", ".join(others)}'} + continue + result[c['id']] = dict(compare_path(path, by_id, switches, items), path=path) + return result + + +def describe_node(node: Optional[tuple]) -> str: + if node is None: + return 'unterminated' + device, port, side = node + text = device + if port: + text += f' port {port}' if port.isdigit() else f' {port}' + if side: + text += f' ({side})' + return text + + +# ── Undocumented links ────────────────────────────────────────────────────── + +def undocumented(cables: List[dict], switches: dict) -> List[dict]: + """Linked switch ports that no live cable mentions, as suggestions to add.""" + documented = set() + for c in cables: + if not _live(c): + continue + for end in ('a', 'b'): + node = _node(c, end) + if node: + documented.add((node[0].lower(), node[1])) + suggestions = [] + for sw_name, sw in sorted((switches or {}).items()): + for port_name, port in (sw.get('ports') or {}).items(): + idx = port.get('port_idx') + if not port.get('up') or idx is None or (sw_name.lower(), str(idx)) in documented: + continue + lldp = port.get('lldp') or {} + sfp = port.get('sfp') or {} + peer = sfp.get('cable_peer') + suggestion = { + 'switch': sw_name, 'port_idx': idx, 'port_name': port_name, + 'neighbor': lldp.get('system_name') or '', 'neighbor_port': lldp.get('port_id') or '', + 'speed_mbps': port.get('speed_mbps') or 0, + 'cable_type': 'DAC' if sfp.get('kind') == 'dac' else '', + 'via': 'lldp' if lldp.get('system_name') else '', + } + if lldp.get('source') == 'unifi-clients': + suggestion['via'] = 'client table' + if peer: + suggestion.update(neighbor=peer['switch'], neighbor_port=str(peer['port_idx']), via='DAC serial') + # Offer each DAC once, from the end that sorts first + if (peer['switch'].lower(), str(peer['port_idx'])) in documented: + continue + if (peer['switch'], peer['port_idx']) < (sw_name, idx): + continue + suggestions.append(suggestion) + return suggestions + + +# ── Labels to make ────────────────────────────────────────────────────────── + +def labels_to_make(cables: List[dict], items: Dict[str, dict], port_labels: List[dict]) -> dict: + """Cable IDs and panel-port labels that haven't been stuck on yet.""" + cords = [ + {'id': c['id'], 'text': c['label'], 'copies': 2 if c.get('b_device') else 1, + 'where': f'{describe_node(_node(c, "a"))} ↔ {describe_node(_node(c, "b"))}'} + for c in sorted(cables, key=lambda c: _natural(c.get('label') or '')) + if c.get('label') and not c.get('label_applied') and c['status'] in ('installed', 'dark') + ] + unlabeled = [c['id'] for c in cables if not c.get('label') and c['status'] in ('installed', 'dark')] + ports = [ + {'item': p['item_name'], 'port': p['port'], 'text': p['label']} + for p in sorted(port_labels, key=lambda p: (p['item_name'].lower(), _natural(p['port']))) + if p.get('label') and not p.get('applied') + ] + return {'cables': cords, 'ports': ports, 'unlabeled_cable_ids': unlabeled} + + +def _natural(text: str) -> list: + return [int(t) if t.isdigit() else t.lower() for t in re.split(r'(\d+)', text or '')] + + +# ── CSV ───────────────────────────────────────────────────────────────────── + +CSV_FIELDS = ['label', 'a_device', 'a_port', 'a_side', 'b_device', 'b_port', 'b_side', + 'cable_type', 'length_m', 'color', 'status', 'label_applied', 'notes'] + + +def cables_to_csv(cables: List[dict]) -> str: + out = io.StringIO() + writer = csv.DictWriter(out, fieldnames=CSV_FIELDS, extrasaction='ignore', lineterminator='\n') + writer.writeheader() + for c in sorted(cables, key=lambda c: _natural(c.get('label') or '')): + row = {k: c.get(k) for k in CSV_FIELDS} + row['label_applied'] = 'yes' if c.get('label_applied') else '' + row['length_m'] = '' if c.get('length_m') is None else f'{float(c["length_m"]):g}' + # Keep spreadsheet apps from running a cell as a formula + writer.writerow({k: _csv_safe(v) for k, v in row.items()}) + return out.getvalue() + + +def _csv_safe(value) -> str: + text = '' if value is None else str(value) + return "'" + text if text[:1] in ('=', '+', '-', '@', '\t', '\r') and not _is_number(text) else text + + +def _is_number(text: str) -> bool: + try: + float(text) + return True + except ValueError: + return False + + +def parse_csv(text: str, items: Dict[str, dict], max_rows: int = 2000) -> Tuple[List[dict], List[str]]: + """Validate a cable CSV. Returns (cables, errors); nothing is applied on any error.""" + reader = csv.DictReader(io.StringIO(text)) + missing = [f for f in ('a_device',) if f not in (reader.fieldnames or [])] + if missing: + return [], [f'Missing column: {", ".join(missing)}'] + rows, errors, labels = [], [], set() + for line, raw in enumerate(reader, start=2): + if line - 1 > max_rows: + errors.append(f'More than {max_rows} rows') + break + data = {k: (v or '').strip() for k, v in raw.items() if k} + for k in list(data): + if data[k].startswith("'") and data[k][1:2] in ('=', '+', '-', '@'): + data[k] = data[k][1:] + if not any(data.values()): + continue + data['label_applied'] = data.get('label_applied', '').lower() in ('yes', 'true', '1', 'y') + try: + cable = validate_cable(data, items) + except ValidationError as e: + errors.append(f'Line {line}: {e}') + continue + if cable['label']: + if cable['label'].lower() in labels: + errors.append(f'Line {line}: label {cable["label"]} appears twice') + continue + labels.add(cable['label'].lower()) + rows.append(cable) + return rows, errors diff --git a/db.py b/db.py index 1e1a461..ffa18aa 100644 --- a/db.py +++ b/db.py @@ -478,3 +478,312 @@ def is_suppressed(target_type: str, target_name: str, target_detail: str = '') - return True return False + + +# --------------------------------------------------------------------------- +# Cutsheets (see cutsheet.py). Every change writes a cutsheet_history row in +# the same transaction, so the log can't miss an edit. +# --------------------------------------------------------------------------- + +_CUTSHEET_SCHEMA = ( + """CREATE TABLE IF NOT EXISTS cutsheet_items ( + id INT AUTO_INCREMENT PRIMARY KEY, + name VARCHAR(100) NOT NULL, + kind VARCHAR(20) NOT NULL, + port_count SMALLINT NULL, + location VARCHAR(255) NOT NULL DEFAULT '', + rack_u SMALLINT NULL, + rack_height SMALLINT NULL, + notes TEXT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uq_item_name (name) + ) ENGINE=InnoDB""", + """CREATE TABLE IF NOT EXISTS cutsheet_cables ( + id INT AUTO_INCREMENT PRIMARY KEY, + label VARCHAR(32) NULL, + a_device VARCHAR(100) NOT NULL, + a_port VARCHAR(64) NOT NULL DEFAULT '', + a_side VARCHAR(5) NOT NULL DEFAULT '', + b_device VARCHAR(100) NOT NULL DEFAULT '', + b_port VARCHAR(64) NOT NULL DEFAULT '', + b_side VARCHAR(5) NOT NULL DEFAULT '', + cable_type VARCHAR(32) NOT NULL DEFAULT '', + length_m DECIMAL(5,2) NULL, + color VARCHAR(32) NOT NULL DEFAULT '', + status VARCHAR(10) NOT NULL DEFAULT 'installed', + label_applied BOOLEAN NOT NULL DEFAULT FALSE, + notes TEXT NULL, + created_by VARCHAR(255) NOT NULL DEFAULT '', + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_by VARCHAR(255) NOT NULL DEFAULT '', + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uq_cable_label (label), + INDEX idx_cable_a (a_device, a_port), + INDEX idx_cable_b (b_device, b_port) + ) ENGINE=InnoDB""", + """CREATE TABLE IF NOT EXISTS cutsheet_port_labels ( + item_id INT NOT NULL, + port VARCHAR(64) NOT NULL, + label VARCHAR(64) NOT NULL DEFAULT '', + applied BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY (item_id, port) + ) ENGINE=InnoDB""", + """CREATE TABLE IF NOT EXISTS cutsheet_history ( + id INT AUTO_INCREMENT PRIMARY KEY, + changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + changed_by VARCHAR(255) NOT NULL, + entity VARCHAR(12) NOT NULL, + entity_id INT NOT NULL, + action VARCHAR(10) NOT NULL, + summary VARCHAR(500) NOT NULL, + before_json TEXT NULL, + after_json TEXT NULL, + INDEX idx_hist_time (changed_at), + INDEX idx_hist_entity (entity, entity_id) + ) ENGINE=InnoDB""", +) + +_ITEM_FIELDS = ('name', 'kind', 'port_count', 'location', 'rack_u', 'rack_height', 'notes') +_CABLE_FIELDS = ('label', 'a_device', 'a_port', 'a_side', 'b_device', 'b_port', 'b_side', + 'cable_type', 'length_m', 'color', 'status', 'label_applied', 'notes') + + +class CutsheetConflict(Exception): + """A change that would clash with existing records (duplicate name or label, item in use).""" + + +def ensure_cutsheet_schema() -> None: + with get_conn() as conn: + with conn.cursor() as cur: + for statement in _CUTSHEET_SCHEMA: + cur.execute(statement) + + +def _cs_jsonable(row: Optional[dict]) -> Optional[dict]: + if row is None: + return None + out = {} + for k, v in row.items(): + if hasattr(v, 'isoformat'): + v = v.isoformat() + 'Z' + elif v is not None and k == 'length_m': + v = float(v) + elif k in ('label_applied', 'applied'): + v = bool(v) + out[k] = v + return out + + +def _cs_log(cur, user: str, entity: str, entity_id: int, action: str, summary: str, + before: Optional[dict], after: Optional[dict]) -> None: + cur.execute( + """INSERT INTO cutsheet_history (changed_by, entity, entity_id, action, summary, before_json, after_json) + VALUES (%s, %s, %s, %s, %s, %s, %s)""", + (user, entity, entity_id, action, summary[:500], + json.dumps(_cs_jsonable(before)) if before else None, + json.dumps(_cs_jsonable(after)) if after else None), + ) + + +@contextmanager +def _cs_transaction(): + with get_conn() as conn: + conn.begin() + try: + with conn.cursor() as cur: + yield cur + conn.commit() + except pymysql.err.IntegrityError as e: + conn.rollback() + raise CutsheetConflict(_cs_integrity_message(e)) from e + except Exception: + conn.rollback() + raise + + +def _cs_integrity_message(e: Exception) -> str: + text = str(e) + if 'uq_cable_label' in text: + return 'Another cable already has that label' + if 'uq_item_name' in text: + return 'Another item already has that name' + return 'That change conflicts with an existing record' + + +def cutsheet_load() -> dict: + """Every cutsheet record, for one page render or API response.""" + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute('SELECT * FROM cutsheet_items ORDER BY name') + items = [_cs_jsonable(r) for r in cur.fetchall()] + cur.execute('SELECT * FROM cutsheet_cables ORDER BY id') + cables = [_cs_jsonable(r) for r in cur.fetchall()] + cur.execute( + """SELECT l.item_id, i.name AS item_name, l.port, l.label, l.applied + FROM cutsheet_port_labels l JOIN cutsheet_items i ON i.id = l.item_id""" + ) + port_labels = [_cs_jsonable(r) for r in cur.fetchall()] + return {'items': items, 'cables': cables, 'port_labels': port_labels} + + +def cutsheet_history(limit: int = 100) -> list: + with get_conn() as conn: + with conn.cursor() as cur: + cur.execute( + """SELECT id, changed_at, changed_by, entity, entity_id, action, summary + FROM cutsheet_history ORDER BY id DESC LIMIT %s""", + (limit,), + ) + return [_cs_jsonable(r) for r in cur.fetchall()] + + +def _cs_get(cur, table: str, row_id: int) -> Optional[dict]: + cur.execute(f'SELECT * FROM {table} WHERE id=%s FOR UPDATE', (row_id,)) # nosec B608 (fixed names) + return cur.fetchone() + + +def cutsheet_create_item(item: dict, user: str) -> int: + with _cs_transaction() as cur: + cur.execute( + f'INSERT INTO cutsheet_items ({", ".join(_ITEM_FIELDS)}) ' # nosec B608 (fixed names) + f'VALUES ({", ".join(["%s"] * len(_ITEM_FIELDS))})', + tuple(item[f] for f in _ITEM_FIELDS), + ) + item_id = cur.lastrowid + _cs_log(cur, user, 'item', item_id, 'create', f'Added {item["kind"]} {item["name"]}', None, item) + return item_id + + +def cutsheet_update_item(item_id: int, item: dict, user: str) -> bool: + """Update an item. A rename carries over to the cables that reference it.""" + with _cs_transaction() as cur: + before = _cs_get(cur, 'cutsheet_items', item_id) + if before is None: + return False + cur.execute( + f'UPDATE cutsheet_items SET {", ".join(f"{f}=%s" for f in _ITEM_FIELDS)} WHERE id=%s', # nosec B608 + tuple(item[f] for f in _ITEM_FIELDS) + (item_id,), + ) + summary = f'Edited {item["name"]}' + if before['name'] != item['name']: + for end in ('a', 'b'): + cur.execute( + f'UPDATE cutsheet_cables SET {end}_device=%s, updated_by=%s WHERE {end}_device=%s', # nosec B608 + (item['name'], user, before['name']), + ) + summary = f'Renamed {before["name"]} to {item["name"]}' + _cs_log(cur, user, 'item', item_id, 'update', summary, before, item) + return True + + +def cutsheet_delete_item(item_id: int, user: str) -> bool: + with _cs_transaction() as cur: + before = _cs_get(cur, 'cutsheet_items', item_id) + if before is None: + return False + cur.execute( + 'SELECT COUNT(*) AS n FROM cutsheet_cables WHERE a_device=%s OR b_device=%s', + (before['name'], before['name']), + ) + used = cur.fetchone()['n'] + if used: + raise CutsheetConflict(f'{before["name"]} still has {used} cable(s); move or delete them first') + cur.execute('DELETE FROM cutsheet_port_labels WHERE item_id=%s', (item_id,)) + cur.execute('DELETE FROM cutsheet_items WHERE id=%s', (item_id,)) + _cs_log(cur, user, 'item', item_id, 'delete', f'Removed {before["name"]}', before, None) + return True + + +def _cable_summary(c: dict) -> str: + def end(device: str, port: str) -> str: + return ' '.join(filter(None, [device or 'unterminated', port])) + ends = f'{end(c["a_device"], c["a_port"])} ↔ {end(c["b_device"], c["b_port"])}' + return f'{c.get("label") or "unlabeled cable"}: {ends}' + + +def cutsheet_create_cable(cable: dict, user: str) -> int: + with _cs_transaction() as cur: + cur.execute( + f'INSERT INTO cutsheet_cables ({", ".join(_CABLE_FIELDS)}, created_by, updated_by) ' # nosec B608 + f'VALUES ({", ".join(["%s"] * len(_CABLE_FIELDS))}, %s, %s)', + tuple(cable[f] for f in _CABLE_FIELDS) + (user, user), + ) + cable_id = cur.lastrowid + _cs_log(cur, user, 'cable', cable_id, 'create', f'Added {_cable_summary(cable)}', None, cable) + return cable_id + + +def cutsheet_update_cable(cable_id: int, cable: dict, user: str) -> bool: + with _cs_transaction() as cur: + before = _cs_get(cur, 'cutsheet_cables', cable_id) + if before is None: + return False + cur.execute( + f'UPDATE cutsheet_cables SET {", ".join(f"{f}=%s" for f in _CABLE_FIELDS)}, updated_by=%s ' # nosec B608 + 'WHERE id=%s', + tuple(cable[f] for f in _CABLE_FIELDS) + (user, cable_id), + ) + _cs_log(cur, user, 'cable', cable_id, 'update', f'Edited {_cable_summary(cable)}', before, cable) + return True + + +def cutsheet_delete_cable(cable_id: int, user: str) -> bool: + with _cs_transaction() as cur: + before = _cs_get(cur, 'cutsheet_cables', cable_id) + if before is None: + return False + cur.execute('DELETE FROM cutsheet_cables WHERE id=%s', (cable_id,)) + _cs_log(cur, user, 'cable', cable_id, 'delete', f'Deleted {_cable_summary(before)}', before, None) + return True + + +def cutsheet_set_port_label(item_id: int, port: str, label: str, applied: bool, user: str) -> bool: + with _cs_transaction() as cur: + item = _cs_get(cur, 'cutsheet_items', item_id) + if item is None: + return False + cur.execute('SELECT * FROM cutsheet_port_labels WHERE item_id=%s AND port=%s', (item_id, port)) + before = cur.fetchone() + if label: + cur.execute( + """INSERT INTO cutsheet_port_labels (item_id, port, label, applied) VALUES (%s, %s, %s, %s) + ON DUPLICATE KEY UPDATE label=VALUES(label), applied=VALUES(applied)""", + (item_id, port, label, applied), + ) + else: + cur.execute('DELETE FROM cutsheet_port_labels WHERE item_id=%s AND port=%s', (item_id, port)) + after = {'item_id': item_id, 'port': port, 'label': label, 'applied': applied} if label else None + _cs_log(cur, user, 'port_label', item_id, 'update', + f'{item["name"]} port {port} label: {label or "(cleared)"}', before, after) + return True + + +def cutsheet_import_cables(cables: list, user: str) -> dict: + """Add or update cables from a CSV import, matching on label. All or nothing.""" + added = updated = 0 + with _cs_transaction() as cur: + for cable in cables: + before = None + if cable['label']: + cur.execute('SELECT * FROM cutsheet_cables WHERE label=%s FOR UPDATE', (cable['label'],)) + before = cur.fetchone() + if before: + cur.execute( + f'UPDATE cutsheet_cables SET {", ".join(f"{f}=%s" for f in _CABLE_FIELDS)}, ' # nosec B608 + 'updated_by=%s WHERE id=%s', + tuple(cable[f] for f in _CABLE_FIELDS) + (user, before['id']), + ) + _cs_log(cur, user, 'cable', before['id'], 'update', f'Imported {_cable_summary(cable)}', + before, cable) + updated += 1 + else: + cur.execute( + f'INSERT INTO cutsheet_cables ({", ".join(_CABLE_FIELDS)}, created_by, updated_by) ' # nosec + f'VALUES ({", ".join(["%s"] * len(_CABLE_FIELDS))}, %s, %s)', + tuple(cable[f] for f in _CABLE_FIELDS) + (user, user), + ) + _cs_log(cur, user, 'cable', cur.lastrowid, 'create', f'Imported {_cable_summary(cable)}', + None, cable) + added += 1 + return {'added': added, 'updated': updated} diff --git a/schema.sql b/schema.sql index 8a76659..af8b1ad 100644 --- a/schema.sql +++ b/schema.sql @@ -61,3 +61,65 @@ CREATE TABLE IF NOT EXISTS monitor_state ( value MEDIUMTEXT NOT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB; + +-- ── Cutsheets (planned cabling) ──────────────────────────────────────── +-- Also created automatically on first use (db.ensure_cutsheet_schema). +CREATE TABLE IF NOT EXISTS cutsheet_items ( -- panels, jacks, shelves, rack positions + id INT AUTO_INCREMENT PRIMARY KEY, + name VARCHAR(100) NOT NULL, + kind VARCHAR(20) NOT NULL, -- patch_panel, wall_jack, device, shelf, other + port_count SMALLINT NULL, + location VARCHAR(255) NOT NULL DEFAULT '', + rack_u SMALLINT NULL, -- bottom rack unit + rack_height SMALLINT NULL, + notes TEXT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uq_item_name (name) +) ENGINE=InnoDB; + +CREATE TABLE IF NOT EXISTS cutsheet_cables ( -- one row per physical cable + id INT AUTO_INCREMENT PRIMARY KEY, + label VARCHAR(32) NULL, -- the ID on the cable, e.g. C07 + a_device VARCHAR(100) NOT NULL, -- device or item name + a_port VARCHAR(64) NOT NULL DEFAULT '', -- switch port number, interface, or panel port + a_side VARCHAR(5) NOT NULL DEFAULT '', -- front/rear, for panels and jacks + b_device VARCHAR(100) NOT NULL DEFAULT '', -- blank = unterminated + b_port VARCHAR(64) NOT NULL DEFAULT '', + b_side VARCHAR(5) NOT NULL DEFAULT '', + cable_type VARCHAR(32) NOT NULL DEFAULT '', + length_m DECIMAL(5,2) NULL, + color VARCHAR(32) NOT NULL DEFAULT '', + status VARCHAR(10) NOT NULL DEFAULT 'installed', -- planned, installed, dark, retired + label_applied BOOLEAN NOT NULL DEFAULT FALSE, + notes TEXT NULL, + created_by VARCHAR(255) NOT NULL DEFAULT '', + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_by VARCHAR(255) NOT NULL DEFAULT '', + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uq_cable_label (label), + INDEX idx_cable_a (a_device, a_port), + INDEX idx_cable_b (b_device, b_port) +) ENGINE=InnoDB; + +CREATE TABLE IF NOT EXISTS cutsheet_port_labels ( -- e.g. the room name on a panel port + item_id INT NOT NULL, + port VARCHAR(64) NOT NULL, + label VARCHAR(64) NOT NULL DEFAULT '', + applied BOOLEAN NOT NULL DEFAULT FALSE, + PRIMARY KEY (item_id, port) +) ENGINE=InnoDB; + +CREATE TABLE IF NOT EXISTS cutsheet_history ( -- who changed what, and when + id INT AUTO_INCREMENT PRIMARY KEY, + changed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + changed_by VARCHAR(255) NOT NULL, + entity VARCHAR(12) NOT NULL, -- cable, item, port_label + entity_id INT NOT NULL, + action VARCHAR(10) NOT NULL, -- create, update, delete + summary VARCHAR(500) NOT NULL, + before_json TEXT NULL, + after_json TEXT NULL, + INDEX idx_hist_time (changed_at), + INDEX idx_hist_entity (entity, entity_id) +) ENGINE=InnoDB; diff --git a/static/app.js b/static/app.js index cfe0465..53b8da0 100644 --- a/static/app.js +++ b/static/app.js @@ -233,7 +233,7 @@ function initDeviceSearch() { } results.innerHTML = matches.map((d, i) => ` + aria-selected="${i === active}" href="${lt.escHtml(d.url || deviceUrl(d.name))}"> ${lt.escHtml(d.name)} ${lt.escHtml(d.kind)} @@ -250,7 +250,7 @@ function initDeviceSearch() { if (!matches.length) return; if (e.key === 'ArrowDown') { e.preventDefault(); active = (active + 1) % matches.length; render(); } if (e.key === 'ArrowUp') { e.preventDefault(); active = (active - 1 + matches.length) % matches.length; render(); } - if (e.key === 'Enter') { e.preventDefault(); window.location.href = deviceUrl(matches[active].name); } + if (e.key === 'Enter') { e.preventDefault(); window.location.href = matches[active].url || deviceUrl(matches[active].name); } }); document.addEventListener('click', e => { if (!e.target.closest('#device-search')) close(); diff --git a/static/cutsheet.js b/static/cutsheet.js new file mode 100644 index 0000000..99aaec5 --- /dev/null +++ b/static/cutsheet.js @@ -0,0 +1,566 @@ +'use strict'; +/* global CS_CAN_EDIT, deviceUrl */ +// Cutsheet page: renders /api/cutsheet and edits cables, items and panel-port labels. + +const esc = s => lt.escHtml(s == null ? '' : String(s)); +const VERDICT_CLASS = { + match: 'badge-ok', wrong_neighbor: 'badge-critical', no_link: 'badge-critical', missing_port: 'badge-critical', + conflict: 'badge-critical', unverified: 'badge-info', idle: 'badge-neutral', not_checked: 'badge-neutral', + planned: 'badge-purple', dark: 'badge-neutral', retired: 'badge-neutral', +}; +const VERDICT_SHORT = { + match: 'Matches', wrong_neighbor: 'Wrong neighbor', no_link: 'No link', missing_port: 'Missing port', + conflict: 'Port conflict', unverified: 'Unverified', idle: 'Idle', not_checked: 'Not checked', + planned: 'Planned', dark: 'Dark', retired: 'Retired', +}; +const KIND_TEXT = { patch_panel: 'Patch panel', wall_jack: 'Wall jack', device: 'Device', shelf: 'Shelf', other: 'Other' }; + +let _cs = null; +let _showAllSuggestions = false; +const SUGGESTIONS_SHOWN = 10; + +function natural(a, b) { + return String(a).localeCompare(String(b), undefined, { numeric: true, sensitivity: 'base' }); +} + +function itemByName(name) { + return (_cs.items || []).find(i => i.name === name); +} + +function isPassive(name) { + const item = itemByName(name); + return !!(item && _cs.options.passive_kinds.includes(item.kind)); +} + +function verdictBadge(c) { + return `${esc(VERDICT_SHORT[c.verdict] || c.verdict)}`; +} + +function deviceLink(name) { + if (!name) return 'unterminated'; + const item = itemByName(name); + const known = (_cs.catalog || []).some(d => d.name === name && (d.kind === 'switch' || d.kind === 'host')); + if (item && !known) return `${esc(name)}`; + return `${esc(name)}`; +} + +function endText(c, end) { + const device = c[`${end}_device`]; + if (!device) return deviceLink(''); + const port = c[`${end}_port`]; + const side = c[`${end}_side`]; + let text = deviceLink(device); + if (port) text += ` ${/^\d+$/.test(port) ? 'port ' : ''}${esc(port)}`; + if (side) text += ` ${esc(side)}`; + return text; +} + +function pathText(c) { + // The run beyond this cable, through any panels or jacks + const nodes = (c.path || []).filter(h => !('cable_id' in h)).map(h => h.text); + return nodes.length > 2 ? nodes.map(esc).join(' → ') : ''; +} + +// ── Rendering ─────────────────────────────────────────────────────────── +function renderSummary() { + const cables = _cs.cables.filter(c => c.status !== 'retired'); + const problems = cables.filter(c => c.problem).length; + const matches = cables.filter(c => c.verdict === 'match').length; + const labels = _cs.labels.cables.length + _cs.labels.ports.length; + const chip = (n, text, cls, target) => + `${n} ${esc(text)}`; + document.getElementById('cs-summary').innerHTML = [ + chip(cables.length, 'cables', '', 'cs-cables-section'), + chip(matches, 'match what Gandalf sees', 'cs-chip-ok', 'cs-cables-section'), + chip(problems, problems === 1 ? 'problem' : 'problems', problems ? 'cs-chip-bad' : '', 'cs-cables-section'), + chip(_cs.suggestions.length, 'undocumented links', _cs.suggestions.length ? 'cs-chip-warn' : '', 'cs-suggest-section'), + chip(labels, 'labels to make', labels ? 'cs-chip-warn' : '', 'cs-labels-section'), + ].join(''); +} + +function cableMatches(c, words, status, problemsOnly) { + if (status && c.status !== status) return false; + if (problemsOnly && !c.problem) return false; + if (!words.length) return true; + const hay = [c.label, c.a_text, c.b_text, c.cable_type, c.color, c.notes, c.detail, c.status] + .join(' ').toLowerCase(); + return words.every(w => hay.includes(w)); +} + +function renderCables() { + const words = document.getElementById('cs-filter').value.trim().toLowerCase().split(/\s+/).filter(Boolean); + const status = document.getElementById('cs-status-filter').value; + const problemsOnly = document.getElementById('cs-problems-only').checked; + const rows = _cs.cables + .filter(c => cableMatches(c, words, status, problemsOnly)) + .sort((a, b) => (b.problem - a.problem) || natural(a.label || '~', b.label || '~') || a.id - b.id); + document.getElementById('cs-cable-count').textContent = _cs.cables.length; + const wrap = document.getElementById('cs-cables'); + if (!_cs.cables.length) { + wrap.innerHTML = emptyState('🔌', 'No cables yet', + CS_CAN_EDIT ? 'Add one with + Cable, or start from the undocumented links below.' : 'An admin can add cables.'); + return; + } + if (!rows.length) { + wrap.innerHTML = emptyState('🔍', 'No cables match the filter', ''); + return; + } + wrap.innerHTML = ` +
+ + + ${CS_CAN_EDIT ? '' : ''} + ${rows.map(c => ` + + + + + + + + ${CS_CAN_EDIT ? `` : ''} + `).join('')} + +
Cables
IDFromToTypeStatusCheck
${c.label ? esc(c.label) : '–'}${c.label && !c.label_applied && c.status !== 'retired' ? ' not stuck on' : ''}${endText(c, 'a')}${endText(c, 'b')}${pathText(c) ? `
${pathText(c)}
` : ''}		${esc([c.cable_type, c.length_m ? `${c.length_m} m` : '', c.color].filter(Boolean).join(' · ')) || '–'}${esc(c.status)}${verdictBadge(c)}
${esc(c.detail)}
+
`; +} + +function renderSuggestions() { + const list = _cs.suggestions; + document.getElementById('cs-suggest-count').textContent = list.length; + const wrap = document.getElementById('cs-suggestions'); + if (!list.length) { + wrap.innerHTML = emptyState('✅', 'Every live switch port is documented', ''); + return; + } + wrap.innerHTML = ` +
+ + + ${CS_CAN_EDIT ? '' : ''} + ${list.slice(0, _showAllSuggestions ? list.length : SUGGESTIONS_SHOWN).map((s, i) => ` + + + + + + ${CS_CAN_EDIT ? `` : ''} + `).join('')} + +
Undocumented links
Switch portUniFi nameSeesFound by
${deviceLink(s.switch)} port ${esc(s.port_idx)}${esc(s.port_name)}${s.neighbor ? `${deviceLink(s.neighbor)}${s.neighbor_port ? ` ${esc(s.neighbor_port)}` : ''}` : 'unknown'}${esc(s.via || 'link only')}
+
+ ${list.length > SUGGESTIONS_SHOWN ? `` : ''}`; +} + +function cableAt(name, port, side) { + return _cs.cables.find(c => c.status !== 'retired' && + ['a', 'b'].some(e => c[`${e}_device`] === name && c[`${e}_port`] === String(port) && c[`${e}_side`] === side)); +} + +function farEnd(c, name, port, side) { + const end = (c.a_device === name && c.a_port === String(port) && c.a_side === side) ? 'b' : 'a'; + return endText(c, end); +} + +function renderPanels() { + const panels = _cs.items.filter(i => i.kind === 'patch_panel').sort((a, b) => natural(a.name, b.name)); + const wrap = document.getElementById('cs-panels'); + if (!panels.length) { + wrap.innerHTML = emptyState('🗄️', 'No patch panels yet', CS_CAN_EDIT ? 'Add one with + Panel / item.' : ''); + return; + } + const labels = {}; + for (const l of _cs.port_labels) labels[`${l.item_id}|${l.port}`] = l; + wrap.innerHTML = panels.map(p => { + const rows = []; + for (let n = 1; n <= (p.port_count || 0); n++) { + const front = cableAt(p.name, n, 'front'); + const rear = cableAt(p.name, n, 'rear'); + const lab = labels[`${p.id}|${n}`]; + const cell = (c, side) => c + ? `${esc(c.label || 'unlabeled')} → ${farEnd(c, p.name, n, side)}` + : '–'; + rows.push(` + + ${n} + ${lab ? esc(lab.label) + (lab.applied ? '' : ' not stuck on') : '–'} + ${CS_CAN_EDIT ? `` : ''} + ${cell(front, 'front')} + ${cell(rear, 'rear')} + `); + } + return ` +
+ ╚╝ +
${esc(p.name)}${p.rack_u ? ` · U${esc(p.rack_u)}` : ''}${p.location ? ` · ${esc(p.location)}` : ''}
+
+ + + + ${rows.join('')} +
Ports on ${esc(p.name)}
PortLabelFront (patch cord)Rear (run)
+
+
`; + }).join(''); +} + +function renderLabels() { + const { cables, ports, unlabeled_cable_ids: unlabeled } = _cs.labels; + document.getElementById('cs-label-count').textContent = cables.length + ports.length; + const wrap = document.getElementById('cs-labels'); + if (!cables.length && !ports.length) { + wrap.innerHTML = emptyState('🏷️', 'Nothing to label', unlabeled.length + ? `${unlabeled.length} cable(s) have no ID yet: give them one to add them here.` : ''); + return; + } + const tick = (attrs, label) => CS_CAN_EDIT + ? `` : ''; + wrap.innerHTML = ` +
+ ${cables.map(c => ` +
+ ${tick(`data-action="applied-cable" data-id="${esc(c.id)}"`, `${c.text} is stuck on`)} + ${esc(c.text)} + ×${esc(c.copies)} + ${esc(c.where)} +
`).join('')} + ${ports.map(p => ` +
+ ${tick(`data-action="applied-port" data-item="${esc(p.item)}" data-port="${esc(p.port)}"`, `${p.text} is stuck on`)} + ${esc(p.text)} + ×1 + ${esc(p.item)} port ${esc(p.port)} +
`).join('')} +
+ ${unlabeled.length ? `

${unlabeled.length} more cable(s) have no ID yet.

` : ''}`; +} + +function renderRack() { + const placed = _cs.items.filter(i => i.rack_u).sort((a, b) => (b.rack_u + (b.rack_height || 1)) - (a.rack_u + (a.rack_height || 1))); + const wrap = document.getElementById('cs-rack'); + if (!placed.length) { + wrap.innerHTML = emptyState('🗄️', 'No rack positions recorded', CS_CAN_EDIT ? 'Set a rack unit on a panel, shelf or device in the inventory.' : ''); + return; + } + wrap.innerHTML = ` +
+ ${placed.map(i => { + const top = i.rack_u + (i.rack_height || 1) - 1; + return ` +
+ ${i.rack_height > 1 ? `U${esc(top)}–${esc(i.rack_u)}` : `U${esc(i.rack_u)}`} + ${deviceLink(i.name)} + ${esc(KIND_TEXT[i.kind] || i.kind)} +
`; + }).join('')} +
`; +} + +function renderItems() { + const items = [..._cs.items].sort((a, b) => natural(a.kind, b.kind) || natural(a.name, b.name)); + document.getElementById('cs-item-count').textContent = items.length; + const wrap = document.getElementById('cs-items'); + if (!items.length) { + wrap.innerHTML = emptyState('📦', 'No items yet', CS_CAN_EDIT ? 'Add your patch panel first: + Panel / item.' : ''); + return; + } + wrap.innerHTML = ` +
+ + + ${CS_CAN_EDIT ? '' : ''} + ${items.map(i => ` + + + + + + + + ${CS_CAN_EDIT ? `` : ''} + `).join('')} + +
Inventory
NameKindPortsRackLocationNotes
${deviceLink(i.name)}${esc(KIND_TEXT[i.kind] || i.kind)}${esc(i.port_count || '–')}${i.rack_u ? `U${esc(i.rack_u)}${i.rack_height > 1 ? ` (${esc(i.rack_height)}U)` : ''}` : '–'}${esc(i.location || '–')}${esc(i.notes || '')}
+
`; +} + +function renderHistory() { + const rows = _cs.history || []; + const wrap = document.getElementById('cs-history'); + if (!rows.length) { + wrap.innerHTML = emptyState('📋', 'No changes yet', ''); + return; + } + wrap.innerHTML = ` +
+ + + + ${rows.map(h => ` + `).join('')} + +
Change history
WhenWhoChange
${esc(fmtWhen(h.changed_at))}${esc(h.changed_by)}${esc(h.summary)}
+
`; +} + +function fmtWhen(iso) { + const d = new Date(iso); + return isNaN(d) ? iso : d.toLocaleString(); +} + +function emptyState(icon, title, body) { + return `
${icon}
+
${esc(title)}
${body ? `
${esc(body)}
` : ''}
`; +} + +function renderDatalists() { + document.getElementById('cs-device-list').innerHTML = + (_cs.catalog || []).map(d => ``).join(''); +} + +function render() { + renderSummary(); + renderCables(); + renderSuggestions(); + renderPanels(); + renderLabels(); + renderRack(); + renderItems(); + renderHistory(); + renderDatalists(); +} + +async function load() { + try { + _cs = await lt.api.get('/api/cutsheet'); + } catch (err) { + document.getElementById('cs-summary').innerHTML = + `
Couldn't load the cutsheet: ${esc(err.message)}
`; + return; + } + render(); + const hash = window.location.hash; + if (hash && /^#(cable|item)-\d+$/.test(hash)) { + const el = document.getElementById(hash.slice(1)); + if (el) { el.classList.add('cs-flash'); el.scrollIntoView({ block: 'center' }); } + } +} + +// ── Editing ───────────────────────────────────────────────────────────── +function fillPorts(end) { + const device = document.querySelector(`#cs-cable-form [name="${end}_device"]`).value; + const entry = (_cs.catalog || []).find(d => d.name === device); + document.getElementById(`cs-${end}-port-list`).innerHTML = entry + ? entry.ports.map(p => ``).join('') : ''; + const side = document.querySelector(`.cs-side[data-end="${end}"]`); + side.hidden = !isPassive(device); +} + +function openCable(cable, prefill) { + const form = document.getElementById('cs-cable-form'); + form.reset(); + const types = ['', ..._cs.options.cable_types]; + form.cable_type.innerHTML = types.map(t => ``).join(''); + const data = Object.assign({ status: 'installed', label: cable ? '' : _cs.next_label }, cable || {}, prefill || {}); + for (const key of ['id', 'label', 'a_device', 'a_port', 'b_device', 'b_port', 'cable_type', 'length_m', 'color', 'status', 'notes']) { + if (form[key]) form[key].value = data[key] == null ? '' : data[key]; + } + form.a_side.value = data.a_side || 'front'; + form.b_side.value = data.b_side || 'front'; + form.label_applied.checked = !!data.label_applied; + form.querySelector('.cs-delete').hidden = !cable; + document.getElementById('cs-cable-title').textContent = cable ? `Edit ${cable.label || 'cable'}` : 'New cable'; + fillPorts('a'); + fillPorts('b'); + lt.modal.open('cs-cable-modal'); +} + +function openItem(item) { + const form = document.getElementById('cs-item-form'); + form.reset(); + const data = Object.assign({ kind: 'patch_panel' }, item || {}); + for (const key of ['id', 'name', 'kind', 'port_count', 'rack_u', 'rack_height', 'location', 'notes']) { + form[key].value = data[key] == null ? '' : data[key]; + } + if (!item) form.port_count.value = 24; + form.querySelector('.cs-delete').hidden = !item; + document.getElementById('cs-item-title').textContent = item ? `Edit ${item.name}` : 'New panel or item'; + lt.modal.open('cs-item-modal'); +} + +function openPortLabel(itemId, port) { + const form = document.getElementById('cs-port-form'); + form.reset(); + const item = _cs.items.find(i => String(i.id) === String(itemId)); + const lab = _cs.port_labels.find(l => String(l.item_id) === String(itemId) && String(l.port) === String(port)); + form.item_id.value = itemId; + form.port.value = port; + form.label.value = lab ? lab.label : ''; + form.applied.checked = !!(lab && lab.applied); + document.getElementById('cs-port-title').textContent = `${item ? item.name : 'Panel'} port ${port}`; + lt.modal.open('cs-port-modal'); +} + +function formData(form, fields) { + const out = {}; + for (const f of fields) out[f] = form[f].value; + return out; +} + +async function save(fn, okText, modalId) { + try { + await fn(); + if (modalId) lt.modal.close(modalId); + lt.toast.success(okText); + await load(); + } catch (err) { + lt.toast.error(err.message || 'Save failed'); + } +} + +async function saveCable(e) { + e.preventDefault(); + const form = e.target; + const body = formData(form, ['label', 'a_device', 'a_port', 'a_side', 'b_device', 'b_port', 'b_side', + 'cable_type', 'length_m', 'color', 'status', 'notes']); + body.label_applied = form.label_applied.checked; + const id = form.id.value; + await save(() => id ? lt.api.put(`/api/cutsheet/cables/${encodeURIComponent(id)}`, body) + : lt.api.post('/api/cutsheet/cables', body), 'Cable saved', 'cs-cable-modal'); +} + +async function saveItem(e) { + e.preventDefault(); + const form = e.target; + const body = formData(form, ['name', 'kind', 'port_count', 'rack_u', 'rack_height', 'location', 'notes']); + const id = form.id.value; + await save(() => id ? lt.api.put(`/api/cutsheet/items/${encodeURIComponent(id)}`, body) + : lt.api.post('/api/cutsheet/items', body), 'Saved', 'cs-item-modal'); +} + +async function savePortLabel(e) { + e.preventDefault(); + const form = e.target; + const url = `/api/cutsheet/items/${encodeURIComponent(form.item_id.value)}/ports/${encodeURIComponent(form.port.value)}`; + await save(() => lt.api.put(url, { label: form.label.value, applied: form.applied.checked }), 'Label saved', 'cs-port-modal'); +} + +async function markCableApplied(id, applied) { + const c = _cs.cables.find(x => String(x.id) === String(id)); + if (!c) return; + const body = {}; + for (const k of ['label', 'a_device', 'a_port', 'a_side', 'b_device', 'b_port', 'b_side', 'cable_type', 'length_m', 'color', 'status', 'notes']) { + body[k] = c[k] == null ? '' : c[k]; + } + body.label_applied = applied; + await save(() => lt.api.put(`/api/cutsheet/cables/${encodeURIComponent(id)}`, body), `${c.label} marked as labeled`); +} + +async function markPortApplied(itemName, port, applied) { + const item = itemByName(itemName); + const lab = item && _cs.port_labels.find(l => l.item_id === item.id && String(l.port) === String(port)); + if (!lab) return; + await save(() => lt.api.put(`/api/cutsheet/items/${encodeURIComponent(item.id)}/ports/${encodeURIComponent(port)}`, + { label: lab.label, applied }), `${lab.label} marked as labeled`); +} + +async function importCsv(e, dryRun) { + if (e) e.preventDefault(); + const errorsEl = document.getElementById('cs-import-errors'); + errorsEl.innerHTML = ''; + const fileInput = document.getElementById('cs-import-file'); + let text = document.getElementById('cs-import-text').value; + if (fileInput.files.length) text = await fileInput.files[0].text(); + let resp, data; + try { + // fetch directly: lt.api keeps only the message, and the row errors matter here + resp = await fetch('/api/cutsheet/import', { + method: 'POST', + headers: Object.assign({ 'Content-Type': 'application/json' }, lt.csrf.headers()), + body: JSON.stringify({ csv: text, dry_run: dryRun }), + }); + data = await resp.json(); + } catch (err) { + errorsEl.innerHTML = `
Import failed: ${esc(err.message)}
`; + return; + } + if (!resp.ok) { + errorsEl.innerHTML = `
${esc(data.error || `HTTP ${resp.status}`)}
` + + (data.errors ? `
    ${data.errors.map(x => `
  • ${esc(x)}
    • `).join('')}
` : ''); + return; + } + if (dryRun) { + errorsEl.innerHTML = `
${esc(data.rows)} row(s) look valid. Nothing imported yet.
`; + return; + } + lt.modal.close('cs-import-modal'); + lt.toast.success(`Imported: ${data.added} added, ${data.updated} updated`); + await load(); +} + +// ── Wiring ────────────────────────────────────────────────────────────── +document.addEventListener('click', e => { + const btn = e.target.closest('[data-action]'); + if (!btn) return; + const action = btn.dataset.action; + if (action === 'new-cable') openCable(null); + else if (action === 'new-item') openItem(null); + else if (action === 'edit-cable') openCable(_cs.cables.find(c => String(c.id) === btn.dataset.id)); + else if (action === 'edit-item') openItem(_cs.items.find(i => String(i.id) === btn.dataset.id)); + else if (action === 'edit-port-label') openPortLabel(btn.dataset.item, btn.dataset.port); + else if (action === 'print-labels') printLabels(); + else if (action === 'toggle-suggestions') { _showAllSuggestions = !_showAllSuggestions; renderSuggestions(); } + else if (action === 'import-check') importCsv(null, true); + else if (action === 'add-suggestion') { + const s = _cs.suggestions[Number(btn.dataset.i)]; + openCable(null, { + a_device: s.switch, a_port: String(s.port_idx), b_device: s.neighbor, b_port: s.neighbor_port, + cable_type: s.cable_type, + }); + } else if (action === 'delete-cable') { + const form = document.getElementById('cs-cable-form'); + if (!confirm('Delete this cable record? Retiring it keeps the history visible; deleting removes it.')) return; + save(() => lt.api.delete(`/api/cutsheet/cables/${encodeURIComponent(form.id.value)}`), 'Cable deleted', 'cs-cable-modal'); + } else if (action === 'delete-item') { + const form = document.getElementById('cs-item-form'); + if (!confirm(`Delete ${form.name.value}?`)) return; + save(() => lt.api.delete(`/api/cutsheet/items/${encodeURIComponent(form.id.value)}`), 'Deleted', 'cs-item-modal'); + } +}); + +document.addEventListener('change', e => { + const t = e.target; + if (t.dataset.action === 'applied-cable') markCableApplied(t.dataset.id, t.checked); + else if (t.dataset.action === 'applied-port') markPortApplied(t.dataset.item, t.dataset.port, t.checked); +}); + +function printLabels() { + document.body.classList.add('cs-print-labels'); + window.print(); + document.body.classList.remove('cs-print-labels'); +} + +function openFromQuery() { + const q = new URLSearchParams(window.location.search); + if (!CS_CAN_EDIT || !q.get('a_device')) return; + openCable(null, { a_device: q.get('a_device'), a_port: q.get('a_port') || '' }); + history.replaceState(null, '', window.location.pathname); +} + +document.getElementById('cs-filter').addEventListener('input', () => _cs && renderCables()); +document.getElementById('cs-status-filter').addEventListener('change', () => _cs && renderCables()); +document.getElementById('cs-problems-only').addEventListener('change', () => _cs && renderCables()); +if (CS_CAN_EDIT) { + document.getElementById('cs-cable-form').addEventListener('submit', saveCable); + document.getElementById('cs-item-form').addEventListener('submit', saveItem); + document.getElementById('cs-port-form').addEventListener('submit', savePortLabel); + document.getElementById('cs-import-form').addEventListener('submit', e => importCsv(e, false)); + for (const end of ['a', 'b']) { + const input = document.querySelector(`#cs-cable-form [name="${end}_device"]`); + input.addEventListener('input', () => fillPorts(end)); + input.addEventListener('change', () => fillPorts(end)); + } +} + +load().then(openFromQuery); diff --git a/static/port-inspector.js b/static/port-inspector.js index 46e7a58..2d3c37e 100644 --- a/static/port-inspector.js +++ b/static/port-inspector.js @@ -89,6 +89,29 @@ function faceplatePortTitle(port, idx, role) { return escHtml(parts.join(' · ')); } +// ── Cutsheet section of the port panel ────────────────────────────────── +const CUTSHEET_PROBLEM = new Set(['wrong_neighbor', 'no_link', 'missing_port', 'conflict']); + +function cutsheetPanelHtml(swName, idx) { + const entries = ((_apiData && _apiData.cutsheet) || []).filter(e => e.port === String(idx)); + const head = '
Cutsheet
'; + if (!entries.length) { + const add = `/cutsheet?a_device=${encodeURIComponent(swName)}&a_port=${encodeURIComponent(idx)}`; + return `${head}
Planned + Not documented · Add cable
`; + } + return head + entries.map(e => { + const cls = CUTSHEET_PROBLEM.has(e.verdict) ? 'val-crit' : (e.verdict === 'match' ? 'val-good' : 'val-neutral'); + const path = (e.path || []).filter(h => !('cable_id' in h)).map(h => escHtml(h.text)).join(' → '); + const name = e.label ? escHtml(e.label) : 'unlabeled'; + return ` +
Cable + ${name} · ${escHtml(e.status)}
+
${escHtml(e.detail)}
+ ${path ? `
${path}
` : ''}`; + }).join(''); +} + // ── SFP module section of the port panel ──────────────────────────────── function sfpPanelHtml(sfp) { if (!sfp) return ''; @@ -351,6 +374,7 @@ function renderPanel(swName, idx) {
Auto-neg${autoneg}
Media${escHtml(mediaStr)}
+ ${cutsheetPanelHtml(swName, idx)} ${sfpPanelHtml(d.sfp)} ${poeHtml} ${trafficHtml} @@ -442,6 +466,7 @@ function renderPortInspector(detail) { outlets: detail.info && detail.info.outlets, online: detail.status === 'up', } }, hosts: detail.neighbor_hosts || {}, + cutsheet: detail.cutsheet || [], }; const chassis = document.getElementById('device-chassis'); // Auto-refresh redraws the panel, so keep a scrolled panel where the user left it diff --git a/static/style.css b/static/style.css index 67b89d3..90e77fb 100644 --- a/static/style.css +++ b/static/style.css @@ -1313,3 +1313,75 @@ polygon.fp-led-off { fill: #6b6c71; } .fp-fan-strut { fill: none; stroke: #c9c9cc; stroke-width: 5; } .fp-epo { fill: #1c1c1f; } .fp-bracket { fill: none; stroke: #55565c; stroke-width: 1; } + +/* ── Cutsheet ─────────────────────────────────────────────────────────── */ +.cutsheet-check { font-size: 12px; padding: 2px 0; overflow-wrap: anywhere; } +.cutsheet-path { font-size: 11px; color: var(--text-muted, #8aa); padding: 2px 0 8px; overflow-wrap: anywhere; } +.cs-more { margin-top: 8px; } +.cs-actions { display: flex; flex-wrap: wrap; gap: 8px; align-items: center; } +.cs-summary { display: flex; flex-wrap: wrap; gap: 8px; margin: 0 0 18px; } +.cs-chip { padding: 5px 10px; border: 1px solid var(--border-color, #234); border-radius: 3px; font-size: .82rem; + color: inherit; text-decoration: none; } +.cs-chip b { font-size: .95rem; } +.cs-chip-ok b { color: var(--status-ok, #2ecc71); } +.cs-chip-bad { border-color: var(--status-critical, #e0312b); } +.cs-chip-bad b { color: var(--status-critical, #e0312b); } +.cs-chip-warn b { color: var(--accent-amber, #f0a020); } +.cs-filters { display: flex; flex-wrap: wrap; gap: 10px; align-items: center; margin-bottom: 10px; } +.cs-filters .lt-input { flex: 1 1 240px; min-width: 0; } +.cs-filters .lt-select { flex: 0 1 180px; } +.cs-check { display: inline-flex; gap: 6px; align-items: center; font-size: .85rem; cursor: pointer; } +.cs-hint { font-size: .82rem; color: var(--text-muted, #8aa); margin: 0 0 10px; max-width: 75ch; } +.cs-table td { vertical-align: top; } +.cs-id { font-weight: 700; white-space: nowrap; } +.cs-port, .cs-side-tag { font-size: .8rem; color: var(--text-muted, #8aa); } +.cs-side-tag { text-transform: uppercase; letter-spacing: .04em; } +.cs-path, .cs-detail { font-size: .76rem; color: var(--text-muted, #8aa); margin-top: 3px; } +.cs-tag { font-size: .68rem; padding: 1px 5px; border: 1px solid var(--accent-amber, #f0a020); color: var(--accent-amber, #f0a020); + border-radius: 2px; white-space: nowrap; } +.cs-row-problem td:first-child { box-shadow: inset 3px 0 0 var(--status-critical, #e0312b); } +.cs-flash { animation: cs-flash 2.4s ease-out 1; } +@keyframes cs-flash { from { background: rgba(240,160,32,.28); } to { background: transparent; } } +@media (prefers-reduced-motion: reduce) { .cs-flash { animation: none; outline: 2px solid var(--accent-amber, #f0a020); } } +.cs-panel { margin-bottom: 16px; } +.cs-link-btn { background: none; border: 0; padding: 0 0 0 6px; color: var(--accent-cyan, #3cc); cursor: pointer; + font: inherit; font-size: .78rem; text-decoration: underline; } +.cs-label-grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(220px, 1fr)); gap: 8px; } +.cs-label-card { display: grid; grid-template-columns: auto 1fr auto; gap: 2px 8px; align-items: center; + padding: 8px 10px; border: 1px dashed var(--border-color, #234); border-radius: 3px; } +.cs-label-text { font-family: var(--font-mono, monospace); font-size: 1.05rem; font-weight: 700; letter-spacing: .05em; } +.cs-label-copies { font-size: .8rem; color: var(--text-muted, #8aa); } +.cs-label-where { grid-column: 1 / -1; font-size: .74rem; color: var(--text-muted, #8aa); } +.cs-rack { display: flex; flex-direction: column; gap: 3px; max-width: 560px; } +.cs-rack-unit { display: grid; grid-template-columns: 70px 1fr auto; gap: 10px; align-items: center; + min-height: calc(var(--u, 1) * 26px); padding: 4px 10px; border: 1px solid var(--border-color, #234); + border-radius: 2px; background: rgba(255,255,255,.02); } +.cs-rack-u { font-family: var(--font-mono, monospace); font-size: .78rem; color: var(--text-muted, #8aa); } +.cs-rack-kind { font-size: .74rem; color: var(--text-muted, #8aa); } +.cs-end { border: 1px solid var(--border-color, #234); border-radius: 3px; padding: 4px 12px 0; margin: 0 0 12px; } +.cs-end legend { font-size: .78rem; padding: 0 6px; color: var(--text-muted, #8aa); text-transform: uppercase; } +.cs-side[hidden] { display: none; } +.cs-modal-wide { max-width: 720px; width: calc(100vw - 32px); } +.cs-mono { font-family: var(--font-mono, monospace); font-size: .8rem; } +.cs-errors ul { margin: 6px 0 0 18px; font-size: .8rem; } +@media (max-width: 640px) { + /* Cables become cards: one labeled field per line */ + .cs-table thead { display: none; } + .cs-table, .cs-table tbody, .cs-table tr, .cs-table td { display: block; width: 100%; } + .cs-table tr { border: 1px solid var(--border-color, #234); border-radius: 3px; margin-bottom: 10px; padding: 6px 10px; } + .cs-table td { border: 0 !important; padding: 3px 0 !important; overflow-wrap: anywhere; } + .cs-table td[data-label]::before { content: attr(data-label); display: block; font-size: .68rem; + text-transform: uppercase; letter-spacing: .06em; color: var(--text-muted, #8aa); } + .cs-table td.cs-id { font-size: 1rem; } + .cs-row-problem { box-shadow: inset 3px 0 0 var(--status-critical, #e0312b); } + .cs-row-problem td:first-child { box-shadow: none; } + .cs-rack-unit { grid-template-columns: 56px 1fr; } + .cs-rack-kind { display: none; } +} +@media print { + .no-print, .lt-header, .lt-nav-drawer, .lt-nav-drawer-overlay { display: none !important; } + body.cs-print-labels .g-section:not(#cs-labels-section), + body.cs-print-labels .lt-page-header, body.cs-print-labels .cs-summary { display: none !important; } + .cs-label-card { break-inside: avoid; border-color: #000; color: #000; } + .lt-table, .lt-table td, .lt-table th { color: #000 !important; } +} diff --git a/templates/base.html b/templates/base.html index f54f182..43802af 100644 --- a/templates/base.html +++ b/templates/base.html @@ -37,6 +37,9 @@ Dashboard + Cutsheet {% if user.groups and 'admin' in user.groups %} Dashboard + + Cutsheet + {% if user.groups and 'admin' in user.groups %} +
+

Cutsheet

+

Where every cable is meant to go, checked against what Gandalf sees on the network.

+
+
+ {% if can_edit %} + + + + {% endif %} + Export CSV +
+ + +
+ +
+
+

Cables

+ 0 +
+
+ + + +
+
+
+ +
+
+

Undocumented links

+ 0 +
+

Switch ports with a live link that no cable record mentions. Adding one pre-fills what Gandalf already knows; insert the patch panel hop if the cable runs through one.

+
+
+ +
+
+

Patch panels

+
+
+
+ +
+
+

Labels to make

+ 0 + +
+

Type these on the label maker. Cable IDs go on both ends of the cord; panel labels go on the port. Tick each one once it's stuck on.

+
+
+ +
+
+

Rack

+
+
+
+ +
+
+

Inventory

+ 0 +
+

Things Gandalf can't discover: patch panels, wall jacks, shelves, and rack positions for devices.

+
+
+ +
+
+

History

+
+
+
+ + + + + +{% if can_edit %} + + + + + + + + + + + +{% endif %} +{% endblock %} + +{% block scripts %} + + +{% endblock %} diff --git a/tests/test_app.py b/tests/test_app.py index 0c24774..4f29c14 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -171,3 +171,109 @@ class TestDiagnoseWithResolvedNeighbor: 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'] == [] diff --git a/tests/test_cutsheet.py b/tests/test_cutsheet.py new file mode 100644 index 0000000..49bbd4b --- /dev/null +++ b/tests/test_cutsheet.py @@ -0,0 +1,227 @@ +"""Tests for cutsheet.py: validation, path tracing and plan-vs-actual checks.""" +import os +import sys + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) + +import cutsheet as cs # noqa: E402 + +PANEL = {'id': 1, 'name': 'PP1', 'kind': 'patch_panel', 'port_count': 24} +JACK = {'id': 2, 'name': 'Bedroom jack', 'kind': 'wall_jack', 'port_count': 2} +SHELF = {'id': 3, 'name': 'Shelf', 'kind': 'shelf', 'port_count': None} +ITEMS = {i['name']: i for i in (PANEL, JACK, SHELF)} + + +def cable(cid, a, b=None, status='installed', label=None, **kw): + a_dev, a_port, a_side = (a + ('',) * 3)[:3] + b_dev, b_port, b_side = ((b or ('', '', '')) + ('',) * 3)[:3] + c = {'id': cid, 'label': label, 'a_device': a_dev, 'a_port': a_port, 'a_side': a_side, + 'b_device': b_dev, 'b_port': b_port, 'b_side': b_side, 'status': status, + 'cable_type': '', 'length_m': None, 'color': '', 'notes': '', 'label_applied': False} + c.update(kw) + return c + + +def switch(ports): + return {'sw': {'ports': {f'Port {p["port_idx"]}': p for p in ports}}} + + +def port(idx, up=True, seen='', **kw): + p = {'port_idx': idx, 'up': up} + if seen: + p['lldp'] = {'system_name': seen} + p.update(kw) + return p + + +class TestValidation: + def test_minimal_cable(self): + c = cs.validate_cable({'a_device': 'sw', 'a_port': '3'}, ITEMS) + assert c['status'] == 'installed' and c['label'] is None and c['b_device'] == '' + + def test_panel_needs_side_and_valid_port(self): + with pytest.raises(cs.ValidationError, match='front or rear'): + cs.validate_cable({'a_device': 'PP1', 'a_port': '3'}, ITEMS) + with pytest.raises(cs.ValidationError, match='ports 1 to 24'): + cs.validate_cable({'a_device': 'PP1', 'a_port': '25', 'a_side': 'front'}, ITEMS) + c = cs.validate_cable({'a_device': 'PP1', 'a_port': '3', 'a_side': 'rear'}, ITEMS) + assert c['a_side'] == 'rear' + + def test_side_is_dropped_for_active_devices(self): + c = cs.validate_cable({'a_device': 'sw', 'a_port': '3', 'a_side': 'front'}, ITEMS) + assert c['a_side'] == '' + + @pytest.mark.parametrize('data, message', [ + ({}, 'From device is required'), + ({'a_device': 'sw', 'label': 'bad label!'}, 'Label may use'), + ({'a_device': 'sw', 'status': 'gone'}, 'status must be'), + ({'a_device': 'sw', 'cable_type': 'Cat9'}, 'Type must be'), + ({'a_device': 'sw', 'length_m': 'long'}, 'Length must be a number'), + ({'a_device': 'sw', 'length_m': '0'}, 'between 0 and 999'), + ({'a_device': 'sw', 'b_port': '3'}, 'Choose a device before a port'), + ({'a_device': 'sw', 'a_port': '1', 'b_device': 'sw', 'b_port': '1'}, 'same port'), + ({'a_device': 'x' * 101}, '100 characters'), + ]) + def test_rejects(self, data, message): + with pytest.raises(cs.ValidationError, match=message): + cs.validate_cable(data, ITEMS) + + def test_item_rules(self): + assert cs.validate_item({'name': 'S', 'kind': 'shelf', 'rack_u': '5'})['rack_height'] == 1 + with pytest.raises(cs.ValidationError, match='port count'): + cs.validate_item({'name': 'P', 'kind': 'patch_panel'}) + with pytest.raises(cs.ValidationError, match='rack unit to go with'): + cs.validate_item({'name': 'S', 'rack_height': '2'}) + with pytest.raises(cs.ValidationError, match='kind must be'): + cs.validate_item({'name': 'S', 'kind': 'rack'}) + + def test_next_label_fills_gaps(self): + assert cs.next_label([]) == 'C01' + assert cs.next_label([{'label': 'C01'}, {'label': 'c03'}, {'label': 'X9'}]) == 'C02' + + +class TestTrace: + def test_switch_through_panel_to_jack(self): + cables = [ + cable(1, ('sw', '13'), ('PP1', '5', 'front')), + cable(2, ('PP1', '5', 'rear'), ('Bedroom jack', '1', 'rear')), + cable(3, ('Bedroom jack', '1', 'front'), ('tv', 'eth0')), + ] + for start in cables: + hops = cs.trace(start, cables, ITEMS)['hops'] + ends = {hops[0][0], hops[-1][0]} + assert ends == {'sw', 'tv'} + assert cs.trace(start, cables, ITEMS)['cable_ids'] in ([1, 2, 3], [3, 2, 1]) + + def test_retired_cables_are_not_followed(self): + cables = [cable(1, ('sw', '1'), ('PP1', '1', 'front')), + cable(2, ('PP1', '1', 'rear'), ('Bedroom jack', '1', 'rear'), status='retired')] + assert cs.trace(cables[0], cables, ITEMS)['cable_ids'] == [1] + + def test_unterminated_end(self): + hops = cs.trace(cable(1, ('PP1', '9', 'rear')), [cable(1, ('PP1', '9', 'rear'))], ITEMS)['hops'] + assert hops[-1] is None + + +class TestEvaluate: + def verdict(self, cables, switches, cid=1): + return cs.evaluate(cables, ITEMS, switches)[cid]['verdict'] + + def test_match_through_a_panel(self): + cables = [cable(1, ('sw', '13'), ('PP1', '5', 'front')), + cable(2, ('PP1', '5', 'rear'), ('tv', 'eth0'))] + assert self.verdict(cables, switch([port(13, seen='tv')])) == 'match' + assert self.verdict(cables, switch([port(13, seen='TV ')]), cid=2) == 'match' + + def test_wrong_neighbor(self): + cables = [cable(1, ('sw', '13'), ('tv', 'eth0'))] + result = cs.evaluate(cables, ITEMS, switch([port(13, seen='printer')]))[1] + assert result['verdict'] == 'wrong_neighbor' and 'printer' in result['detail'] + + def test_no_link_vs_idle(self): + assert self.verdict([cable(1, ('sw', '13'), ('tv', 'eth0'))], switch([port(13, up=False)])) == 'no_link' + # A run that ends at a jack with nothing plugged in is normal + jack = [cable(1, ('sw', '13'), ('Bedroom jack', '1', 'rear'))] + assert self.verdict(jack, switch([port(13, up=False)])) == 'idle' + + def test_unverified_when_nothing_identifies_the_far_end(self): + assert self.verdict([cable(1, ('sw', '13'), ('tv', 'eth0'))], switch([port(13)])) == 'unverified' + jack = [cable(1, ('sw', '13'), ('Bedroom jack', '1', 'rear'))] + result = cs.evaluate(jack, ITEMS, switch([port(13, seen='camera')]))[1] + assert result['verdict'] == 'unverified' and 'camera' in result['detail'] + + def test_missing_port_and_unknown_devices(self): + assert self.verdict([cable(1, ('sw', '99'), ('tv', 'eth0'))], switch([port(13)])) == 'missing_port' + assert self.verdict([cable(1, ('host-a', 'eth0'), ('host-b', 'eth0'))], switch([])) == 'not_checked' + + def test_statuses_skip_the_check(self): + for status in ('planned', 'dark', 'retired'): + cables = [cable(1, ('sw', '13'), ('tv', 'eth0'), status=status)] + assert self.verdict(cables, switch([port(13, seen='printer')])) == status + + def test_a_planned_hop_makes_the_run_planned(self): + cables = [cable(1, ('sw', '13'), ('PP1', '5', 'front')), + cable(2, ('PP1', '5', 'rear'), ('tv', 'eth0'), status='planned')] + assert self.verdict(cables, switch([port(13, seen='printer')])) == 'planned' + + def test_dac_serial_confirms_both_ends(self): + switches = { + 'gw': {'ports': {'p': port(11, sfp={'kind': 'dac', 'cable_peer': {'switch': 'agg', 'port_idx': 7}})}}, + 'agg': {'ports': {'p': port(7, sfp={'kind': 'dac', 'cable_peer': {'switch': 'gw', 'port_idx': 11}})}}, + } + result = cs.evaluate([cable(1, ('gw', '11'), ('agg', '7'))], ITEMS, switches)[1] + assert result['verdict'] == 'match' and 'DAC' in result['detail'] + + def test_switch_to_switch_reports_the_worse_end(self): + switches = {'a': {'ports': {'p': port(1, seen='b')}}, 'b': {'ports': {'p': port(2, seen='c')}}} + assert self.verdict([cable(1, ('a', '1'), ('b', '2'))], switches) == 'wrong_neighbor' + + def test_two_installed_cables_on_one_port_conflict(self): + cables = [cable(1, ('sw', '13'), ('tv', 'eth0'), label='C01'), + cable(2, ('sw', '13'), ('printer', 'eth0'), label='C02'), + cable(3, ('sw', '13'), ('tv', 'eth0'), status='planned')] + result = cs.evaluate(cables, ITEMS, switch([port(13, seen='tv')])) + assert result[1]['verdict'] == 'conflict' and 'C02' in result[1]['detail'] + assert result[3]['verdict'] == 'planned' + + +class TestUndocumented: + def test_lists_live_ports_without_a_cable(self): + switches = switch([port(1, seen='tv'), port(2), port(3, up=False)]) + found = cs.undocumented([cable(1, ('sw', '2'), ('x', ''))], switches) + assert [(s['port_idx'], s['neighbor']) for s in found] == [(1, 'tv')] + + def test_retired_cables_dont_count(self): + found = cs.undocumented([cable(1, ('sw', '1'), ('x', ''), status='retired')], switch([port(1)])) + assert len(found) == 1 + + def test_dac_offered_once_from_one_end(self): + switches = { + 'agg': {'ports': {'p': port(7, sfp={'kind': 'dac', 'cable_peer': {'switch': 'gw', 'port_idx': 11}})}}, + 'gw': {'ports': {'p': port(11, sfp={'kind': 'dac', 'cable_peer': {'switch': 'agg', 'port_idx': 7}})}}, + } + found = cs.undocumented([], switches) + assert len(found) == 1 and found[0]['cable_type'] == 'DAC' and found[0]['via'] == 'DAC serial' + + +class TestLabels: + def test_only_unapplied_labels_of_live_cables(self): + cables = [cable(1, ('sw', '1'), ('tv', ''), label='C10'), cable(2, ('sw', '2'), ('tv', ''), label='C2'), + cable(3, ('sw', '3'), ('tv', ''), label='C3', label_applied=True), + cable(4, ('sw', '4'), label='C4', status='dark'), + cable(5, ('sw', '5'), ('tv', ''), label='C5', status='planned'), + cable(6, ('sw', '6'), ('tv', ''))] + ports = [{'item_name': 'PP1', 'port': '10', 'label': 'OFFICE', 'applied': False}, + {'item_name': 'PP1', 'port': '2', 'label': 'GARAGE', 'applied': False}, + {'item_name': 'PP1', 'port': '3', 'label': 'DONE', 'applied': True}] + out = cs.labels_to_make(cables, ITEMS, ports) + assert [c['text'] for c in out['cables']] == ['C2', 'C4', 'C10'] + assert [c['copies'] for c in out['cables']] == [2, 1, 2] + assert [p['text'] for p in out['ports']] == ['GARAGE', 'OFFICE'] + assert out['unlabeled_cable_ids'] == [6] + + +class TestCsv: + def test_round_trip(self): + original = [cable(1, ('sw', '13'), ('PP1', '5', 'front'), label='C01', cable_type='Cat6', + length_m=1.5, color='white', label_applied=True, notes='to office, via panel')] + rows, errors = cs.parse_csv(cs.cables_to_csv(original), ITEMS) + assert errors == [] + assert rows[0]['label'] == 'C01' and rows[0]['b_side'] == 'front' and rows[0]['length_m'] == 1.5 + assert rows[0]['label_applied'] is True and rows[0]['notes'] == 'to office, via panel' + + def test_formulas_are_neutralised_and_restored(self): + text = cs.cables_to_csv([cable(1, ('sw', '1'), label='C01', notes='=HYPERLINK("x")')]) + assert "'=HYPERLINK" in text + rows, _ = cs.parse_csv(text, ITEMS) + assert rows[0]['notes'] == '=HYPERLINK("x")' + + def test_any_bad_row_is_reported(self): + text = 'label,a_device,a_port,a_side\nC01,sw,1,\nC01,sw,2,\n,PP1,40,front\n' + rows, errors = cs.parse_csv(text, ITEMS) + assert any('appears twice' in e for e in errors) and any('Line 4' in e for e in errors) + + def test_missing_required_column(self): + assert cs.parse_csv('label\nC1\n', ITEMS) == ([], ['Missing column: a_device'])