Files
gandalf/scripts/demo/mock_pulse.py
T
jared 4a1998046b
Lint / Python (flake8) (push) Failing after 2m9s
Lint / JS (eslint) (push) Successful in 11s
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 39s
Test / Python Tests (pytest) (push) Successful in 40s
Lint / Notify on failure (push) Successful in 3s
README: add walkthrough GIF, rationale and gallery; add demo seed, mock PULSE and capture scripts
2026-10-03 02:23:26 -04:00

116 lines
3.9 KiB
Python

"""Minimal stand-in for the PULSE internal API, returning canned diagnostics output (fictional data)."""
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
CANNED = """=== carrier ===
1
=== operstate ===
up
=== sysfs_stats ===
rx_bytes:9183029113
tx_bytes:7732918231
rx_errors:14
tx_errors:0
rx_dropped:0
tx_dropped:0
rx_crc_errors:14
rx_frame_errors:0
rx_fifo_errors:0
tx_carrier_errors:0
collisions:0
rx_missed_errors:0
=== carrier_changes ===
6
=== ethtool ===
Settings for enp5s0:
Supported ports: [ FIBRE ]
Supported link modes: 10000baseT/Full
Speed: 10000Mb/s
Duplex: Full
Port: FIBRE
PHYAD: 0
Auto-negotiation: off
Link detected: yes
=== ethtool_driver ===
driver: ixgbe
version: 6.8.12-4-pve
firmware-version: 0x800003e1, 1.3429.0
bus-info: 0000:05:00.0
=== ethtool_pause ===
Pause parameters for enp5s0:
Autonegotiate: off
RX: off
TX: off
=== ethtool_ring ===
Ring parameters for enp5s0:
Pre-set maximums:
RX: 8192
RX Mini: n/a
RX Jumbo: n/a
TX: 8192
Current hardware settings:
RX: 512
RX Mini: n/a
RX Jumbo: n/a
TX: 512
=== ethtool_stats ===
NIC statistics:
rx_packets: 183029113
tx_packets: 120918231
rx_errors: 14
tx_errors: 0
rx_crc_errors: 14
rx_missed_errors: 0
tx_timeout_count: 0
fdir_match: 41022
tx_flow_control_xon: 0
rx_flow_control_xon: 0
=== ethtool_dom ===
Identifier : 0x03 (SFP)
Connector : 0x07 (LC)
Vendor name : FS
Vendor PN : SFP-10GSR-85
Laser wavelength : 850nm
Laser bias current : 6.100 mA
Laser output power : 0.5530 mW / -2.57 dBm
Receiver signal average optical power : 0.0120 mW / -19.21 dBm
Module temperature : 47.50 degrees C / 117.50 degrees F
Module voltage : 3.3120 V
=== ip_link ===
5: enp5s0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 9000 qdisc mq master vmbr1 state UP mode DEFAULT group default qlen 1000
link/ether 02:00:5e:10:00:05 brd ff:ff:ff:ff:ff:ff
RX: bytes packets errors dropped missed mcast
9183029113 183029113 14 0 0 10234
TX: bytes packets errors dropped carrier collsns
7732918231 120918231 0 0 0 0
=== ip_addr ===
5: enp5s0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 9000 qdisc mq master vmbr1 state UP group default qlen 1000
link/ether 02:00:5e:10:00:05 brd ff:ff:ff:ff:ff:ff
=== ip_route ===
=== dmesg ===
[ 412.118233] ixgbe 0000:05:00.0 enp5s0: NIC Link is Up 10 Gbps, Flow Control: None
[1188221.402917] ixgbe 0000:05:00.0 enp5s0: NIC Link is Down
[1188224.190331] ixgbe 0000:05:00.0 enp5s0: NIC Link is Up 10 Gbps, Flow Control: None
[2230418.550112] ixgbe 0000:05:00.0 enp5s0: Reset adapter
[2230421.118904] ixgbe 0000:05:00.0 enp5s0: NIC Link is Up 10 Gbps, Flow Control: None
=== lldpctl ===
-------------------------------------------------------------------------------
LLDP neighbors:
-------------------------------------------------------------------------------
Interface: enp5s0, via: LLDP, RID: 1, Time: 12 days, 03:11:42
Chassis:
ChassisID: mac aa:bb:cc:00:00:03
SysName: USW-Aggregation
Port:
PortID: ifname Port 4
=== end ===
"""
class H(BaseHTTPRequestHandler):
def _send(self, obj, code=200):
b = json.dumps(obj).encode(); self.send_response(code); self.send_header('Content-Type', 'application/json'); self.send_header('Content-Length', str(len(b))); self.end_headers(); self.wfile.write(b)
def do_POST(self):
self.rfile.read(int(self.headers.get('Content-Length', 0) or 0)); self._send({'execution_id': 'demo-exec-0001'})
def do_GET(self):
self._send({'status': 'completed', 'logs': [{'action': 'command_result', 'stdout': CANNED}]})
def log_message(self, *a): pass
HTTPServer(('127.0.0.1', 8650), H).serve_forever()