commit b7b25062723650ebc744851a51dd8b0fc2ab475a Author: Mikei386 <44135113+Mikei386@users.noreply.github.com> Date: Mon Sep 28 14:27:55 2026 +0200 Add isolated Athena Deck prototype with demo control and live hardware telemetry diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7a60b85 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/README.md b/README.md new file mode 100644 index 0000000..1079234 --- /dev/null +++ b/README.md @@ -0,0 +1,89 @@ +# Athena Deck · Prototyp 0.1 + +Eigenständige neue Oberfläche, Python-Standardbibliothek und HTML/CSS/JavaScript. +Keine Installation von Python-Paketen oder Frontend-Builds nötig. Python >= 3.10. + +## Start und Stopp auf dem Mac + +```sh +cd /Users/mike_i386/Documents/ChatGPT/Athena/AI-Profile-Router/experiments/athena-control-ui +python3 server.py --port 8108 +``` + +GUI: http://127.0.0.1:8108 · Beenden mit Ctrl+C im Startterminal. +SIGTERM beendet ebenfalls den Server und seinen Demo-Kindprozess sauber. +Ein belegter UI-Port führt zum Abbruch, nicht zur Übernahme eines Dienstes. +Der Demo-Port wird vom Betriebssystem vergeben und als bereits gebundener +Socket an den Kindprozess weitergereicht (keine Port-Prüfungs-Race). +Kein SSH-Tunnel erforderlich. Die Anwendung bindet ausschließlich Loopback. +Im separaten Git-Repository liegen diese Dateien im Wurzelverzeichnis; +dort entsprechend `cd` in den Checkout und denselben Startbefehl verwenden. + +Hardware-Abfragen verwenden den vorhandenen SSH-Zugang: +`ssh -i ~/.ssh/athena_key -o BatchMode=yes root@192.168.1.212`. +Der bekannte Hostschlüssel muss bereits hinterlegt sein. Schlüssel werden +weder kopiert noch ausgegeben. Athena erhält nur den Collector über stdin; +auf dem Host werden keine Dateien angelegt und keine Dienste gesteuert. + +## Interne API v1 + +Alle Antworten JSON, keine persistenten Zustandsdaten. GUI verwendet nur diese API. + +| Methode | Pfad | Ergebnis | +|---|---|---| +| GET | `/api/v1/status` | Name, Version, Laufzeit, Modus, Demo-Zustand | +| GET | `/api/v1/hardware` | Verfügbarkeit, Messzeit, CPU, RAM, GPU-Liste, Fehler | +| GET | `/api/v1/demo` | state, reachable, pid, port, location | +| POST | `/api/v1/demo/start` | Idempotent starten und Health prüfen | +| POST | `/api/v1/demo/stop` | Idempotent eigenen Kindprozess stoppen | + +POST benötigt `X-Athena-Deck: 1`; Browser-Origin muss zum Host passen. +Host-Allowlist: localhost oder 127.0.0.1 mit tatsächlichem UI-Port. +Keine CORS-Freigabe. 403 bei ungültigem Zugriff, 404 bei unbekannten Pfaden, +503 bei fehlgeschlagener Demo-Bereitschaft. Keine frei übergebbaren Befehle, +Prozess-IDs, Service-Namen oder Remote-Ziele. + +```sh +curl http://127.0.0.1:8108/api/v1/status +curl -X POST -H 'X-Athena-Deck: 1' http://127.0.0.1:8108/api/v1/demo/start +curl -X POST -H 'X-Athena-Deck: 1' http://127.0.0.1:8108/api/v1/demo/stop +``` + +Hardware: `available` bezeichnet Erfolg der SSH-Collector-Abfrage, einzelne +Messwerte können trotzdem fehlen (`null`). SSH-Ausfall liefert available=false, +leere Daten und eine Fehlermeldung, keine alten Werte als vermeintliches Livebild. +CPU-Auslastung: 250-ms-Differenz aus `/proc/stat` ohne doppelte Guest-Zählung. +RAM: MemTotal minus MemAvailable aus `/proc/meminfo`, Einheit Bytes. +CPU-Temperatur: k10temp/coretemp temp1_input, Grad Celsius. +GPU: nvidia-smi CSV mit Index, UUID, Name, VRAM in MiB, GPU-Auslastung in Prozent, +Temperatur in Grad Celsius. Keine Zuordnung nach vermuteter GPU-Reihenfolge. +Die Anzeige rechnet Speicher in GiB um. Fehlende Werte heißen „Nicht verfügbar“. +Abfrage bedarfsgesteuert mit fünf Sekunden Cache, GUI-Polling alle fünf Sekunden. +Keine Historie und kein Hintergrund-Collector bei geschlossener Hardware-Seite. + +## Struktur und Grenzen + +- `server.py`: API, separater HardwareProvider, eigenständiger DemoService. +- `demo.py`: Health-only HTTP-Kindprozess auf dem Mac, ohne Modellabhängigkeiten. +- `collect_hardware.py`: fest begrenzte lesende Linux-Hardware-Abfragen. +- `index.html`, `app.js`, `style.css`: neue responsive Oberfläche. +- `test_server.py`: Prozess-Lebenszyklus, Kontrollgrenzen, Hardware-Ausfall. + +Sprachmodelle/Chat, Bildgenerierung, Audio und Weitere Dienste sind ausdrücklich +Platzhalter. Keine Downloads, Installation, Presets, Chats oder Modellwechsel. +Spätere Modellprofile und native llama.cpp-Worker sollten eigene Service-Adapter +mit derselben Status-/Start-/Stopp-Trennung erhalten. Noch kein Worker-Registry, +Scheduler, Authentifizierung für Netzwerkbetrieb oder produktionsreifer Supervisor. +SIGKILL/Absturz-Cleanup ist nicht implementiert; regulär Ctrl+C/SIGTERM verwenden. +SSH verwendet den vorgegebenen Root-Zugang; ein eingeschränktes Telemetrie-Konto +wäre ein späterer eigener Auftrag. Keinerlei Host-Konfiguration wurde geändert. + +## Verifikation am 28.09.2026 + +`python3 -m unittest discover -s . -v`: drei Tests erfolgreich. +Zusätzlich GUI-Start → Läuft/Erreichbar → Stopp → Gestoppt/Nicht erreichbar geprüft. +Hardware im Browser mit Ryzen 5 5600, 46,9 GiB RAM, RTX 3060 und RTX 5080 geprüft; +CPU-/GPU-Temperaturen, VRAM und Auslastung vorhanden. Hardware-Ausfall im Test +simuliert, ohne Athena oder SSH zu unterbrechen. +Produktive Router-/Modell-/Bild-/Audio-/Netzwerkdienste nicht verändert. +Vorhandene ungesicherte Änderungen im übergeordneten Repository unangetastet. diff --git a/app.js b/app.js new file mode 100644 index 0000000..b8fa9a1 --- /dev/null +++ b/app.js @@ -0,0 +1,17 @@ +const view=document.querySelector('#view'), error=document.querySelector('#error'); +let busy=false, refreshing=false; +const esc=v=>String(v??'Nicht verfügbar').replace(/[&<>"']/g,c=>({'&':'&','<':'<','>':'>','"':'"',"'":'''}[c])); +const fmt=(v,unit='')=>v==null?'Nicht verfügbar':`${Number(v).toLocaleString('de-DE',{maximumFractionDigits:1})}${unit}`; +const heading=(tag,title,desc)=>`
${tag}

${title}

${desc}

`; +async function api(path,method='GET'){const r=await fetch('/api/v1/'+path,{method,headers:method==='POST'?{'X-Athena-Deck':'1'}:{},signal:AbortSignal.timeout(15000)});if(!r.ok)throw Error(`API-Anfrage fehlgeschlagen (${r.status})`);return r.json()} +const metric=(title,value)=>`
${title}${value}
`; +const bar=v=>v==null?'':``; +const placeholders={chat:['Sprachmodelle / Chat','Modellprofile, Laufzeit und Chat','Hier werden später Modellprofile und native llama.cpp-Prozesse verwaltet.'],image:['Bildgenerierung','Worker und Aufträge','Hier entsteht später die Übersicht für Bild-Worker und deren Zustand.'],audio:['Audio','Spracheingabe und Sprachausgabe','Hier werden später getrennte Worker für Erkennung und Ausgabe eingebunden.'],services:['Weitere Dienste','Erweiterbare Dienste','Hier finden später zusätzliche Worker und ihre Schnittstellen Platz.']}; +async function refresh(){if(refreshing)return;refreshing=true;const page=location.hash.slice(1)||'home';document.querySelectorAll('nav a').forEach(a=>a.classList.toggle('active',a.hash==='#'+page));try{error.textContent='';let html; +if(page==='home'){const s=await api('status'),d=s.demo;html=heading('DEIN KONTROLLZENTRUM','Alles beginnt mit einer klaren Basis.','Eine kleine Oberfläche für Status, Hardware und den ersten steuerbaren Dienst.')+`
Anwendung
Bereit

Athena Deck läuft lokal auf deinem Mac.

Laufzeit ${fmt(s.uptime_seconds,' s')} · API v1
Betriebsmodus
Isoliert

Ein eigenständiger Prototyp mit lesendem Hardware-Zugriff auf Athena.

Keine Modell- oder Router-Steuerung angebunden

Demo-Dienst

LOKAL / CPU

Ein minimaler HTTP-Prozess zum Testen von Start, Stopp und Erreichbarkeit.

${metric('Prozess',d.state==='running'?'Läuft':'Gestoppt')}${metric('Health-Prüfung',d.reachable?'Erreichbar':'Nicht erreichbar')}${metric('Lokaler Port',d.port??'—')}
Lädt keine Modelle · endet beim Beenden von Athena Deck
`; +}else if(page==='hardware'){const h=await api('hardware');html=heading('TELEMETRIE / ATHENA','Hardware im Blick.','Live-Messwerte vom Host 192.168.1.212. Aktualisierung alle fünf Sekunden.');if(!h.available)html+=`
Hardware nicht verfügbar

${esc(h.errors.join(' '))}

`;else{const c=h.cpu,r=h.ram;html+=`
CPU

${esc(c.name)}

${metric('Auslastung',fmt(c.percent,' %'))}${metric('Temperatur',fmt(c.temperature_c,' °C'))}
${bar(c.percent)}
System-RAM

Arbeitsspeicher

${metric('Belegt',fmt(r.used_bytes==null?null:r.used_bytes/2**30,' GiB'))}${metric('Gesamt',fmt(r.total_bytes==null?null:r.total_bytes/2**30,' GiB'))}
${bar(r.total_bytes&&r.used_bytes!=null?100*r.used_bytes/r.total_bytes:null)}
`;for(const g of h.gpus)html+=`
GPU ${g.index}${fmt(g.temperature_c,' °C')}

${esc(g.name)}

${metric('GPU-Auslastung',fmt(g.percent,' %'))}${metric('VRAM belegt / gesamt',fmt(g.used_mib==null?null:g.used_mib/1024)+' / '+fmt(g.total_mib==null?null:g.total_mib/1024,' GiB'))}
${bar(g.percent)}
`;if(h.gpus.length<2)html+='
Nicht alle zwei GPUs verfügbar.
';html+=`

${esc(h.source)} · Messung ${new Date(h.sampled_at*1000).toLocaleTimeString('de-DE')}
${esc(h.errors.join(' '))}

`;} +}else{const p=placeholders[page]||placeholders.services;html=heading('ARBEITSBEREICH / VORBEREITET',p[0],p[1])+`
＋

Platz für den nächsten Schritt.

${p[2]}

Platzhalter · noch keine Funktionen

Installation, Downloads, Presets und Modellwechsel sind in dieser Version nicht enthalten.

`;} +if((location.hash.slice(1)||'home')===page){view.innerHTML=html;for(const action of ['start','stop'])document.getElementById(action)?.addEventListener('click',()=>control(action));} +}catch(e){error.textContent=e.message;view.innerHTML=heading('VERBINDUNG','Status nicht verfügbar','Die API ist gerade nicht erreichbar. Angezeigte Messwerte wurden verworfen.');}finally{refreshing=false;}} +async function control(action){busy=true;document.querySelectorAll('button').forEach(b=>b.disabled=true);try{await api('demo/'+action,'POST')}catch(e){error.textContent=e.message;busy=false;return}busy=false;await refresh()} +window.addEventListener('hashchange',refresh);refresh();setInterval(refresh,5000); diff --git a/collect_hardware.py b/collect_hardware.py new file mode 100644 index 0000000..5a373c1 --- /dev/null +++ b/collect_hardware.py @@ -0,0 +1,66 @@ +"""Read-only Linux collector, executed via SSH stdin; never imports router code.""" +import csv +import json +import time +import subprocess +from pathlib import Path + + +def read(path): + try: + return Path(path).read_text().strip() + except OSError: + return '' + + +def number(value): + try: + return float(value) + except (ValueError, TypeError): + return None + + +def ticks(): + values = list(map(int, read('/proc/stat').splitlines()[0].split()[1:9])) + return sum(values), sum(values[3:5]) + + +def collect(): + errors = [] + cpu = {'name': None, 'percent': None, 'temperature_c': None} + try: + a, ai = ticks() + time.sleep(.25) + b, bi = ticks() + cpu['percent'] = round(100 * (1 - (bi-ai)/(b-a)), 1) if b > a else None + except (ValueError, IndexError): + errors.append('CPU-Auslastung nicht verfügbar') + for line in read('/proc/cpuinfo').splitlines(): + if line.startswith('model name'): + cpu['name'] = line.split(':', 1)[1].strip() + break + for path in Path('/sys/class/hwmon').glob('hwmon*'): + if read(path / 'name') in ('k10temp', 'coretemp'): + value = number(read(path / 'temp1_input')) + cpu['temperature_c'] = value / 1000 if value is not None else None + break + mem = {} + for line in read('/proc/meminfo').splitlines(): + parts = line.split() + if parts[0] in ('MemTotal:', 'MemAvailable:'): + mem[parts[0][:-1]] = int(parts[1]) * 1024 + total, available = mem.get('MemTotal'), mem.get('MemAvailable') + ram = {'total_bytes': total, 'used_bytes': total-available if total is not None and available is not None else None} + gpus = [] + try: + result = subprocess.run(['nvidia-smi', '--query-gpu=index,name,uuid,memory.total,memory.used,utilization.gpu,temperature.gpu', '--format=csv,noheader,nounits'], capture_output=True, text=True, timeout=4, check=True) + for row in csv.reader(result.stdout.splitlines(), skipinitialspace=True): + index, name, uuid, total, used, usage, temp = row + gpus.append(dict(index=int(index), name=name, uuid=uuid, total_mib=number(total), used_mib=number(used), percent=number(usage), temperature_c=number(temp))) + except (OSError, subprocess.SubprocessError, ValueError): + errors.append('GPU-Messwerte nicht verfügbar') + return dict(source='Athena · SSH /proc + hwmon + nvidia-smi', sampled_at=time.time(), cpu=cpu, ram=ram, gpus=gpus, errors=errors) + + +if __name__ == '__main__': + print(json.dumps(collect())) diff --git a/demo.py b/demo.py new file mode 100644 index 0000000..10e0a0a --- /dev/null +++ b/demo.py @@ -0,0 +1,22 @@ +"""Owned child: health-only HTTP server, inherited bound loopback socket.""" +import json +import socket +import sys +from http.server import BaseHTTPRequestHandler, HTTPServer + +class Handler(BaseHTTPRequestHandler): + def do_GET(self): + body = json.dumps({'service': 'athena-deck-demo', 'status': 'ok'}).encode() + self.send_response(200 if self.path == '/health' else 404) + self.send_header('Content-Type', 'application/json') + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): + pass + +server = HTTPServer(('127.0.0.1', 0), Handler, bind_and_activate=False) +server.socket.close() +server.socket = socket.socket(fileno=int(sys.argv[1])) +server.server_address = server.socket.getsockname() +server.serve_forever() diff --git a/index.html b/index.html new file mode 100644 index 0000000..071fca9 --- /dev/null +++ b/index.html @@ -0,0 +1 @@ +Athena Deck
ATHENA CONTROL SURFACEv0.1 · Prototyp
diff --git a/server.py b/server.py new file mode 100644 index 0000000..7b66ebf --- /dev/null +++ b/server.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python3 +"""Athena Deck prototype: loopback API, owned demo process, read-only telemetry.""" +import argparse +import json +import signal +import socket +import subprocess +import sys +import threading +import time +import urllib.request +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +ROOT = Path(__file__).resolve().parent + +class DemoService: + def __init__(self): + self.lock = threading.RLock() + self.process = None + self.port = None + + def status(self): + with self.lock: + running = self.process is not None and self.process.poll() is None + reachable = False + if running: + try: + with urllib.request.urlopen(f'http://127.0.0.1:{self.port}/health', timeout=.5) as response: + reachable = json.load(response).get('service') == 'athena-deck-demo' + except (OSError, ValueError): + pass + return dict(state='running' if running else 'stopped', reachable=reachable, pid=self.process.pid if running else None, port=self.port if running else None, location='Mac · lokaler Demo-Prozess') + + def start(self): + with self.lock: + if self.process is None or self.process.poll() is not None: + with socket.socket() as sock: + sock.bind(('127.0.0.1', 0)) + sock.listen(8) + self.port = sock.getsockname()[1] + self.process = subprocess.Popen([sys.executable, str(ROOT/'demo.py'), str(sock.fileno())], pass_fds=(sock.fileno(),), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + for _ in range(30): + if self.status()['reachable']: + break + time.sleep(.05) + return self.status() + + def stop(self): + with self.lock: + if self.process is not None and self.process.poll() is None: + self.process.terminate() + try: + self.process.wait(timeout=3) + except subprocess.TimeoutExpired: + self.process.kill() + self.process.wait() + return self.status() + +class HardwareProvider: + def __init__(self): + self.lock = threading.Lock() + self.cached = None + self.checked = 0 + + def snapshot(self): + with self.lock: + if time.monotonic() - self.checked < 5 and self.cached is not None: + return self.cached + try: + result = subprocess.run(['ssh', '-i', str(Path.home()/'.ssh/athena_key'), '-o', 'BatchMode=yes', '-o', 'ConnectTimeout=4', '-o', 'StrictHostKeyChecking=yes', 'root@192.168.1.212', 'python3 -'], input=(ROOT/'collect_hardware.py').read_text(), capture_output=True, text=True, timeout=10, check=True) + self.cached = json.loads(result.stdout) + self.cached['available'] = True + except (OSError, subprocess.SubprocessError, ValueError): + self.cached = dict(available=False, sampled_at=None, cpu={}, ram={}, gpus=[], errors=['Athena per SSH nicht erreichbar oder Messung fehlgeschlagen.']) + self.checked = time.monotonic() + return self.cached + +class Server(ThreadingHTTPServer): + daemon_threads = True + def __init__(self, port): + super().__init__(('127.0.0.1', port), Handler) + self.demo = DemoService() + self.hardware = HardwareProvider() + self.started = time.time() + +class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def respond(self, data, status=200, mime='application/json'): + body = json.dumps(data).encode() if mime == 'application/json' else data + self.send_response(status) + self.send_header('Content-Type', mime) + self.send_header('Content-Length', str(len(body))) + self.send_header('Cache-Control', 'no-store') + self.send_header('X-Content-Type-Options', 'nosniff') + self.send_header('Content-Security-Policy', "default-src 'self'; script-src 'self'; style-src 'self'; frame-ancestors 'none'") + self.end_headers() + self.wfile.write(body) + + def valid_host(self): + return self.headers.get('Host') in (f'127.0.0.1:{self.server.server_port}', f'localhost:{self.server.server_port}') + + def do_GET(self): + if not self.valid_host(): + return self.respond({'error': 'Host rejected'}, 403) + routes = {'/': ('index.html', 'text/html; charset=utf-8'), '/app.js': ('app.js', 'text/javascript'), '/style.css': ('style.css', 'text/css')} + if self.path in routes: + name, mime = routes[self.path] + return self.respond((ROOT/name).read_bytes(), mime=mime) + if self.path == '/api/v1/status': + return self.respond(dict(name='Athena Deck', version='0.1.0', state='ready', uptime_seconds=round(time.time()-self.server.started), mode='isolated', demo=self.server.demo.status())) + if self.path == '/api/v1/hardware': + return self.respond(self.server.hardware.snapshot()) + if self.path == '/api/v1/demo': + return self.respond(self.server.demo.status()) + return self.respond({'error': 'Not found'}, 404) + + def do_POST(self): + origin = self.headers.get('Origin') + if not self.valid_host() or self.headers.get('X-Athena-Deck') != '1' or (origin and origin != 'http://' + self.headers.get('Host', '')): + return self.respond({'error': 'Same-origin control required'}, 403) + if self.path not in ('/api/v1/demo/start', '/api/v1/demo/stop'): + return self.respond({'error': 'Not found'}, 404) + action = self.server.demo.start if self.path.endswith('/start') else self.server.demo.stop + result = action() + self.respond(result, 503 if self.path.endswith('/start') and not result['reachable'] else 200) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--port', type=int, default=8108) + args = parser.parse_args() + server = Server(args.port) # Binding fails rather than taking over an occupied port. + def shutdown(*_): + threading.Thread(target=server.shutdown, daemon=True).start() + signal.signal(signal.SIGTERM, shutdown) + signal.signal(signal.SIGINT, shutdown) + print(f'Athena Deck: http://127.0.0.1:{server.server_port}', flush=True) + try: + server.serve_forever() + finally: + server.demo.stop() + server.server_close() + +if __name__ == '__main__': + main() diff --git a/style.css b/style.css new file mode 100644 index 0000000..ebab5ef --- /dev/null +++ b/style.css @@ -0,0 +1 @@ +:root{color-scheme:dark;font-family:Inter,ui-sans-serif,system-ui,sans-serif;background:#101416;color:#e9edea}*{box-sizing:border-box}body{margin:0;display:flex;min-height:100vh}aside{width:270px;background:#141a1c;border-right:1px solid #2c3435;padding:36px 23px;display:flex;flex-direction:column;position:fixed;height:100vh}.brand{font-size:17px;letter-spacing:2px;text-decoration:none;color:inherit;display:flex;align-items:center;gap:8px}.brand span{background:#c5edaf;color:#172412;font-size:28px;border-radius:10px;padding:0 9px;margin-right:6px}.brand b{color:#a5adab;font-weight:400}.eyebrow{font-size:10px;letter-spacing:1.7px;color:#798783;margin-top:46px}nav{display:grid;gap:8px;margin-top:12px}nav a{color:#aeb9b5;text-decoration:none;padding:14px 12px;border-radius:7px;font-size:13px;display:flex;justify-content:space-between}nav a:hover{background:#20292a}nav a.active{background:#28362b;color:#d3efc5}nav span{opacity:.4;font-size:10px}footer{margin-top:auto;font-size:11px;color:#b6c2bc}footer p{line-height:1.8;color:#77857f}i{display:inline-block;width:7px;height:7px;background:#bdeba4;border-radius:50%;margin-right:7px}main{margin-left:270px;padding:30px 5vw 60px;width:calc(100% - 270px);max-width:1500px}header{display:flex;justify-content:space-between;align-items:center;font-size:10px;letter-spacing:2px;color:#7f9189;padding-bottom:40px}.pill{border:1px solid #35423b;padding:7px 11px;border-radius:20px;font-size:11px;letter-spacing:0;color:#c6d6ca}h1{font-size:40px;font-weight:500;letter-spacing:-1.5px;margin:12px 0}h2{font-weight:500;font-size:20px}p{color:#97a79e;line-height:1.65;font-size:14px}.kicker{color:#c5edaf;font-size:11px;letter-spacing:2px}.grid{display:grid;grid-template-columns:repeat(2,minmax(0,1fr));gap:20px;margin-top:30px}.card{background:#181f21;border:1px solid #303a37;border-radius:13px;padding:26px}.wide{grid-column:1/-1}.label{font-size:11px;color:#95a69b;text-transform:uppercase;letter-spacing:1px}.value{font-size:30px;margin:16px 0;color:#e4f1de}.card-top{display:flex;justify-content:space-between;align-items:center;gap:10px}.tag{color:#bdeba4;font-size:12px}.muted{color:#8b9c91}.metrics{display:flex;gap:34px;flex-wrap:wrap;margin:24px 0}.metrics strong{display:block;font-size:24px;font-weight:500;margin:8px 0}.metrics small{color:#8e9c93}button{font:inherit;padding:11px 20px;border-radius:7px;border:1px solid #465348;background:#c5edaf;color:#1a2a18;cursor:pointer;margin-right:8px}button.secondary{background:transparent;color:#dce7da}button:disabled{opacity:.35;cursor:default}button:focus-visible,a:focus-visible{outline:2px solid #d9fac7;outline-offset:4px}.note{border-top:1px solid #303a37;margin-top:25px;padding-top:20px;font-size:12px}.empty{padding:46px 30px;margin-top:30px;border:1px dashed #3b4840;border-radius:12px}.empty .symbol{font-size:35px;color:#bdd4b2}#error{color:#ffc1a7}progress{width:100%;height:6px;accent-color:#bce4a6}progress::-webkit-progress-bar{background:#303a37;border-radius:10px}progress::-webkit-progress-value{background:#bce4a6;border-radius:10px}@media(max-width:850px){aside{position:static;width:100%;height:auto;padding:20px}body{display:block}nav{grid-template-columns:repeat(2,1fr)}footer{display:none}main{margin:0;width:100%;padding:28px 20px}.eyebrow{margin-top:22px}.grid{grid-template-columns:1fr}h1{font-size:32px}header{padding-bottom:20px}} diff --git a/test_server.py b/test_server.py new file mode 100644 index 0000000..73eab4b --- /dev/null +++ b/test_server.py @@ -0,0 +1,45 @@ +import json +import threading +import unittest +import urllib.request +import urllib.error +from unittest.mock import patch +from server import Server, HardwareProvider + +class Tests(unittest.TestCase): + def setUp(self): + self.server=Server(0) + self.thread=threading.Thread(target=self.server.serve_forever) + self.thread.start() + self.url=f'http://127.0.0.1:{self.server.server_port}' + def tearDown(self): + self.server.shutdown() + self.server.demo.stop() + self.server.server_close() + self.thread.join() + def request(self,path,method='GET',headers=None): + req=urllib.request.Request(self.url+'/api/v1/'+path,method=method,headers=headers or {}) + with urllib.request.urlopen(req) as r:return json.load(r) + def test_lifecycle(self): + self.assertEqual(self.request('status')['demo']['state'],'stopped') + a=self.request('demo/start','POST',{'X-Athena-Deck':'1'}) + self.assertTrue(a['reachable']) + self.assertEqual(a['pid'],self.request('demo/start','POST',{'X-Athena-Deck':'1'})['pid']) + child=self.server.demo.process + self.assertEqual(self.request('demo/stop','POST',{'X-Athena-Deck':'1'})['state'],'stopped') + self.assertIsNotNone(child.poll()) + self.assertFalse(self.request('demo/stop','POST',{'X-Athena-Deck':'1'})['reachable']) + def test_control_guard(self): + for headers in ({},{'X-Athena-Deck':'1','Origin':'http://evil.invalid'}): + with self.assertRaises(urllib.error.HTTPError) as e:self.request('demo/start','POST',headers) + self.assertEqual(e.exception.code,403) + with self.assertRaises(urllib.error.HTTPError) as e:self.request('models/start','POST',{'X-Athena-Deck':'1'}) + self.assertEqual(e.exception.code,404) + def test_unavailable_hardware(self): + with patch('server.subprocess.run',side_effect=OSError()): + result=HardwareProvider().snapshot() + self.assertFalse(result['available']) + self.assertEqual(result['gpus'],[]) + self.assertIsNone(result['sampled_at']) + +if __name__=='__main__':unittest.main()