test_process_registry.py
2.57 KB
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
69
70
71
72
73
74
75
76
77
78
from __future__ import annotations
from apps.web_api.runtime.process_registry import ProcessRuntimeRegistry, build_default_process_table
class _RunningProcess:
pid = 1234
@staticmethod
def poll():
return None
def test_snapshot_returns_a_cloned_process_entry():
registry = ProcessRuntimeRegistry(build_default_process_table())
registry.set_process("insight", _RunningProcess())
registry.set_status("insight", "starting")
registry.set_output("insight", ["booting"])
registry.set_healthcheck_started_at("insight", 123.0)
snapshot = registry.snapshot("insight")
snapshot["output"].append("mutated")
assert snapshot == {
"process": registry.get_process("insight"),
"port": 8501,
"status": "starting",
"output": ["booting", "mutated"],
"log_file": None,
"healthcheck_started_at": 123.0,
}
assert registry.snapshot("insight") == {
"process": registry.get_process("insight"),
"port": 8501,
"status": "starting",
"output": ["booting"],
"log_file": None,
"healthcheck_started_at": 123.0,
}
def test_running_apps_status_snapshot_and_active_ports_follow_registry_state():
registry = ProcessRuntimeRegistry(build_default_process_table())
registry.set_status("insight", "running")
registry.set_status("query", "running")
registry.set_output("query", ["line-1", "line-2"])
assert registry.running_apps(["insight", "media", "query"]) == ["insight", "query"]
assert registry.status_snapshot(include_output_lines=True) == {
"insight": {"status": "running", "port": 8501, "output_lines": 0},
"media": {"status": "stopped", "port": 8502, "output_lines": 0},
"query": {"status": "running", "port": 8503, "output_lines": 2},
"forum": {"status": "stopped", "port": None, "output_lines": 0},
}
assert registry.active_ports() == [
"insight:8501",
"media:8502",
"query:8503",
]
def test_reset_runtime_clears_process_and_healthcheck_without_dropping_logs():
registry = ProcessRuntimeRegistry(build_default_process_table())
registry.set_process("media", _RunningProcess())
registry.set_status("media", "running")
registry.set_output("media", ["ok"])
registry.set_healthcheck_started_at("media", 456.0)
registry.reset_runtime("media", status="stopped")
assert registry.snapshot("media") == {
"process": None,
"port": 8502,
"status": "stopped",
"output": ["ok"],
"log_file": None,
"healthcheck_started_at": None,
}