Live Page

1#!/usr/bin/env python3
2"""Serve the live page of a feature test run, then its report.
3
4  live_page.py --run-dir <run-dir> [--parent-pid <pid>] [--port <port>]
5
6run.py starts it in the background. It uses the server of testmax-n8n-ai
7(live_server.py): 127.0.0.1 only, the address in <run-dir>/live-url.txt, and it stops
830 minutes after the last request when the run is done. When verdicts.json changes after the
9run, the page builds the report again, and the open report reloads.
10"""
11
12import argparse
13import threading
14import time
15from pathlib import Path
16
17import common  # noqa: F401  (puts the shared scripts on the import path)
18import report
19from live_server import read_status, serve
20from report import ScenarioApp
21
22
23def rebuild_on_verdicts(run_dir):
24	verdicts, built = run_dir / 'verdicts.json', None
25	while True:
26		time.sleep(2)
27		if read_status(run_dir).get('phase') != 'done' or not verdicts.exists():
28			continue
29		mtime = verdicts.stat().st_mtime
30		if mtime == built:
31			continue
32		try:
33			report.build(run_dir)
34			built = mtime
35		except ValueError:
36			# The judge is still writing the file. The next check reads it again.
37			pass
38		except Exception as e:
39			print(f'cannot build the report: {e}', flush=True)
40			built = mtime
41
42
43def main():
44	ap = argparse.ArgumentParser()
45	ap.add_argument('--run-dir', required=True)
46	ap.add_argument('--parent-pid', type=int)
47	ap.add_argument('--port', type=int, default=0)
48	args = ap.parse_args()
49	run_dir = Path(args.run_dir).resolve()
50	threading.Thread(target=rebuild_on_verdicts, args=(run_dir,), daemon=True).start()
51	serve(ScenarioApp(run_dir), args.parent_pid, args.port)
52
53
54if __name__ == '__main__':
55	main()