MisakaNet/tests/test_dashboard.py at main · ghzhost/MisakaNet · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
import sqlite3
import tempfile
import threading
import time
import unittest
from http.server import ThreadingHTTPServer
from pathlib import Path
from urllib.request import urlopen

from misakanet.tools.dashboard import create_server


class TestTelemetryDashboard(unittest.TestCase):
    def test_dashboard_serves_telemetry_html(self):
        with tempfile.TemporaryDirectory() as tmp:
            telemetry_path = Path(tmp) / "telemetry.db"
            conn = sqlite3.connect(telemetry_path)
            try:
                conn.execute(
                    """
                    CREATE TABLE search_telemetry (
                        query TEXT,
                        timestamp REAL,
                        latency_ms REAL,
                        cache_hit INTEGER
                    )
                    """
                )
                conn.executemany(
                    """
                    INSERT INTO search_telemetry
                        (query, timestamp, latency_ms, cache_hit)
                    VALUES (?, ?, ?, ?)
                    """,
                    [
                        ("alpha query", time.time() - 2, 120.0, 0),
                        ("alpha query", time.time() - 1, 20.0, 1),
                        ("<script>unsafe</script>", time.time(), 30.0, 1),
                    ],
                )
                conn.commit()
            finally:
                conn.close()

            server = create_server(port=0, telemetry_path=telemetry_path)
            self.assertIsInstance(server, ThreadingHTTPServer)
            thread = threading.Thread(target=server.handle_request)
            thread.start()
            try:
                host, port = server.server_address
                with urlopen(f"http://{host}:{port}/", timeout=5) as response:
                    html = response.read().decode("utf-8")
            finally:
                thread.join(timeout=5)
                server.server_close()

            self.assertIn("<!doctype html>", html)
            self.assertIn("Total searches", html)
            self.assertIn("Cache hit rate", html)
            self.assertIn("Average latency", html)
            self.assertIn("Saved time", html)
            self.assertIn("alpha query", html)
            self.assertIn("&lt;script&gt;unsafe&lt;/script&gt;", html)
            self.assertIn('http-equiv="refresh" content="10"', html)


if __name__ == "__main__":
    unittest.main()