Mia.Wu/test_dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

remove live updates, fix save results

Browse Source
This commit is contained in:
Mia Wu
2026-06-01 15:07:49 -04:00
parent 6fb1650cb9
commit f6bced78f3
10 changed files with 39 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/app.py
+13 -57
View File
@@ -1,14 +1,11 @@
import os
import threading
from pathlib import Path
from flask import Flask, Response, jsonify, request, send_from_directory
from flask import Flask, jsonify, request, send_from_directory
from flask_cors import CORS
from db_py import count_tests, del_config, get_all_tests, get_config, set_config
from scanner import full_scan, is_scan_in_progress, resolve_runtime_path, scan_results_only
from sse_py import broadcast, stream_events
from watcher import start_watching
PORT = int(os.getenv("PORT", "3001"))
ALLOWED_KEYS = {
@@ -45,39 +42,6 @@ def _apply_smb_env_from_config():
os.environ[env_key] = str(value)
def _start_full_scan_background(target_dir, results_dir, results_dir_ref, source_label):
def _job():
try:
full_scan(target_dir, results_dir, results_dir_ref)
start_watching(target_dir, results_dir, results_dir_ref)
tests = get_all_tests()
completed = len([t for t in tests if t.get("completed")])
print(f"[{source_label}] Scan complete -> {len(tests)} tests, {completed} completed")
except Exception as exc:
print(f"[{source_label}] background fullScan error: {exc}")
finally:
broadcast({"type": "update"})
worker = threading.Thread(target=_job, daemon=True)
worker.start()
def _start_results_scan_background(results_dir, results_dir_ref, source_label):
def _job():
try:
scan_results_only(results_dir, results_dir_ref)
tests = get_all_tests()
completed = len([t for t in tests if t.get("completed")])
print(f"[{source_label}] Results scan complete -> {len(tests)} tests, {completed} completed")
except Exception as exc:
print(f"[{source_label}] background rescan-results error: {exc}")
finally:
broadcast({"type": "update"})
worker = threading.Thread(target=_job, daemon=True)
worker.start()
@app.get("/api/tests")
def get_tests_route():
completed = request.args.get("completed")
@@ -276,10 +240,11 @@ def set_config_route():
if is_scan_in_progress():
return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})
_start_full_scan_background(target_dir, results_dir, results_dir_ref, "config")
return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})
full_scan(target_dir, results_dir, results_dir_ref)
tests = get_all_tests()
completed = len([t for t in tests if t.get("completed")])
return jsonify({"ok": True, "scanning": False, "testCount": len(tests), "completedCount": completed})
broadcast({"type": "update"})
return jsonify({"ok": True, "testCount": None, "completedCount": None})
@@ -295,8 +260,10 @@ def rescan_route():
if is_scan_in_progress():
return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})
_start_full_scan_background(target_dir, results_dir, results_dir_ref, "config")
return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})
full_scan(target_dir, results_dir, results_dir_ref)
tests = get_all_tests()
completed = len([t for t in tests if t.get("completed")])
return jsonify({"ok": True, "scanning": False, "testCount": len(tests), "completedCount": completed})
@app.post("/api/config/rescan-results")
@@ -310,18 +277,10 @@ def rescan_results_route():
if is_scan_in_progress():
return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})
_start_results_scan_background(results_dir, results_dir_ref, "config")
return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})
@app.get("/api/events")
def events_route():
headers = {
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
}
return Response(stream_events(), mimetype="text/event-stream", headers=headers)
scan_results_only(results_dir, results_dir_ref)
tests = get_all_tests()
completed = len([t for t in tests if t.get("completed")])
return jsonify({"ok": True, "scanning": False, "testCount": len(tests), "completedCount": completed})
@app.get("/")
@@ -351,9 +310,6 @@ def bootstrap():
tests = get_all_tests()
completed = len([t for t in tests if t.get("completed")])
print(f"[server] Scan complete -> {len(tests)} tests found, {completed} completed")
start_watching(target_dir, results_dir)
print("[server] Watching for changes.")
else:
print("[server] No directories configured -> open the dashboard settings to get started.")