Coverage for agentos/tests/test_startup_accelerator.py: 0%
317 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-08 10:59 +0800
1"""Tests for agentos.tools.startup_accelerator — LazyLoader, ModulePreloader, StartupOptimizer."""
3import pytest
4import threading
5import time
6from agentos.tools.startup_accelerator import (
7 LazyLoader,
8 ModulePreloader,
9 StartupOptimizer,
10 _LazyModule,
11 _StartupPhase,
12 create_lazy_loader,
13 create_module_preloader,
14 create_startup_optimizer,
15 quick_start,
16)
19# ============================================================================
20# _LazyModule
21# ============================================================================
23class TestLazyModule:
24 def test_initial_state(self):
25 lm = _LazyModule("os")
26 assert lm._module_name == "os"
27 assert lm._module is None
29 def test_repr_unloaded(self):
30 lm = _LazyModule("nonexistent")
31 r = repr(lm)
32 assert "nonexistent" in r
33 assert "unloaded" in r
35 def test_repr_loaded(self):
36 lm = _LazyModule("os")
37 lm._load()
38 r = repr(lm)
39 assert "unloaded" not in r
41 def test_deferred_load(self):
42 lm = _LazyModule("os")
43 assert lm._module is None
44 # Access an attribute triggers load
45 _ = lm.path
46 assert lm._module is not None
48 def test_getattr_attribute(self):
49 lm = _LazyModule("json")
50 assert lm.dumps({"k": 1}) == '{"k": 1}'
52 def test_load_caches(self):
53 lm = _LazyModule("os")
54 m1 = lm._load()
55 m2 = lm._load()
56 assert m1 is m2
59# ============================================================================
60# LazyLoader
61# ============================================================================
63class TestLazyLoader:
64 def test_register_new(self):
65 ll = LazyLoader()
66 proxy = ll.register("os")
67 assert isinstance(proxy, _LazyModule)
68 assert proxy._module is None
70 def test_register_idempotent(self):
71 ll = LazyLoader()
72 p1 = ll.register("json")
73 p2 = ll.register("json")
74 assert p1 is p2
76 def test_getitem(self):
77 ll = LazyLoader()
78 proxy = ll["os"]
79 assert proxy._module_name == "os"
81 def test_load_now(self):
82 ll = LazyLoader()
83 ll.register("json")
84 mod = ll.load_now("json")
85 assert hasattr(mod, "dumps")
87 def test_load_now_returns_module(self):
88 ll = LazyLoader()
89 mod = ll.load_now("os")
90 assert hasattr(mod, "path")
92 def test_load_all(self):
93 ll = LazyLoader()
94 ll.register("json")
95 ll.register("os")
96 results = ll.load_all()
97 assert len(results) == 2
98 for name, elapsed in results:
99 assert elapsed >= 0
101 def test_preload(self):
102 ll = LazyLoader()
103 modules = ["json", "os"]
104 results = ll.preload(modules)
105 assert len(results) == 2
106 for name, elapsed in results:
107 assert elapsed >= 0
109 def test_stats(self):
110 ll = LazyLoader()
111 ll.register("json")
112 ll.register("os")
113 ll.load_now("json")
114 s = ll.stats
115 assert s["registered"] == 2
116 assert s["loaded"] == 1
117 assert s["unloaded"] == 1
118 assert "json" in s["load_times"]
119 assert isinstance(s["total_load_time"], float)
121 def test_stats_all_unloaded(self):
122 ll = LazyLoader()
123 assert ll.stats["registered"] == 0
124 assert ll.stats["loaded"] == 0
126 def test_thread_safety(self):
127 ll = LazyLoader()
128 errors = []
130 def worker(name):
131 try:
132 ll.register(name)
133 ll.load_now(name)
134 _ = ll.stats
135 except Exception as e:
136 errors.append(e)
138 threads = [threading.Thread(target=worker, args=(f"os",)) for _ in range(10)]
139 for t in threads:
140 t.start()
141 for t in threads:
142 t.join()
143 assert len(errors) == 0
146# ============================================================================
147# ModulePreloader
148# ============================================================================
150class TestModulePreloader:
151 def test_defaults(self):
152 mp = ModulePreloader()
153 assert mp._max_concurrent == 4
155 def test_custom_concurrent(self):
156 mp = ModulePreloader(max_concurrent=2)
157 assert mp._max_concurrent == 2
159 def test_precompile_serial(self):
160 mp = ModulePreloader(max_concurrent=1)
161 results = mp.precompile(["json", "os"], parallel=False)
162 assert len(results) == 2
163 assert "json" in results
164 assert results["json"] >= 0
166 def test_precompile_parallel(self):
167 mp = ModulePreloader()
168 results = mp.precompile(["json", "os", "sys"], parallel=True)
169 assert len(results) == 3
171 def test_precompile_single(self):
172 mp = ModulePreloader()
173 results = mp.precompile(["json"], parallel=True)
174 assert len(results) == 1
176 def test_get(self):
177 mp = ModulePreloader()
178 mp.precompile(["json"], parallel=False)
179 mod = mp.get("json")
180 assert mod is not None
181 assert hasattr(mod, "dumps")
183 def test_get_missing(self):
184 mp = ModulePreloader()
185 assert mp.get("nonexistent_mod_12345") is None
187 def test_warm_cache(self):
188 mp = ModulePreloader()
189 count = mp.warm_cache(["json", "os"])
190 assert count == 2
191 # Second warm should not re-cache
192 count2 = mp.warm_cache(["json", "os"])
193 assert count2 == 0
195 def test_warm_cache_handles_errors(self):
196 mp = ModulePreloader()
197 count = mp.warm_cache(["nonexistent_mod_12345"])
198 assert count == 0
200 def test_clear(self):
201 mp = ModulePreloader()
202 mp.precompile(["json"], parallel=False)
203 assert mp.get("json") is not None
204 mp.clear()
205 assert mp.get("json") is None
207 def test_stats(self):
208 mp = ModulePreloader()
209 mp.precompile(["json", "os"], parallel=True)
210 s = mp.stats
211 assert s["cached_modules"] == 2
212 assert isinstance(s["total_preload_time"], float)
213 assert len(s["module_times"]) == 2
215 def test_stats_empty(self):
216 mp = ModulePreloader()
217 s = mp.stats
218 assert s["cached_modules"] == 0
220 def test_precompile_parallel_batching(self):
221 mp = ModulePreloader(max_concurrent=2)
222 results = mp.precompile(["json", "os", "sys", "time"], parallel=True)
223 assert len(results) == 4
225 def test_precompile_invalid_module(self):
226 mp = ModulePreloader()
227 # parallel requires >1 modules to actually use thread pool
228 results = mp.precompile(["json", "this_module_does_not_exist_123"], parallel=True)
229 assert "json" in results
230 assert "this_module_does_not_exist_123" not in results
232 def test_precompile_invalid_module_serial_raises(self):
233 mp = ModulePreloader()
234 with pytest.raises(ModuleNotFoundError):
235 mp.precompile(["this_module_does_not_exist_123"], parallel=False)
238# ============================================================================
239# StartupOptimizer
240# ============================================================================
242class TestStartupOptimizer:
243 def test_start_and_end(self):
244 so = StartupOptimizer()
245 so.start()
246 time.sleep(0.01)
247 so.end()
248 assert so.total_duration_ms() > 0
250 def test_begin_end_phase(self):
251 so = StartupOptimizer()
252 so.start()
253 so.begin_phase("init")
254 time.sleep(0.01)
255 duration = so.end_phase("init")
256 so.end()
257 assert duration is not None
258 assert duration > 0
260 def test_end_phase_missing(self):
261 so = StartupOptimizer()
262 assert so.end_phase("nonexistent") is None
264 def test_begin_phase_with_metadata(self):
265 so = StartupOptimizer()
266 so.start()
267 so.begin_phase("db", target="postgres")
268 so.end_phase("db")
269 so.end()
270 report = so.report()
271 phases = report["phases"]
272 db_phase = [p for p in phases if p["name"] == "db"][0]
273 assert db_phase["target"] == "postgres"
275 def test_report_structure(self):
276 so = StartupOptimizer()
277 so.start()
278 so.begin_phase("a")
279 so.end_phase("a")
280 so.begin_phase("b")
281 so.end_phase("b")
282 so.end()
283 r = so.report()
284 assert "total_duration_ms" in r
285 assert "phase_count" in r
286 assert "phases" in r
287 assert "bottleneck" in r
288 assert r["phase_count"] == 2
289 assert isinstance(r["total_duration_ms"], float)
291 def test_report_bottleneck(self):
292 so = StartupOptimizer()
293 so.start()
294 so.begin_phase("fast")
295 so.end_phase("fast")
296 so.begin_phase("slow")
297 time.sleep(0.02)
298 so.end_phase("slow")
299 so.end()
300 r = so.report()
301 assert r["bottleneck"] == "slow"
303 def test_report_empty(self):
304 so = StartupOptimizer()
305 so.start()
306 so.end()
307 r = so.report()
308 assert r["phase_count"] == 0
309 assert r["bottleneck"] is None
311 def test_phases_sorted_by_duration(self):
312 so = StartupOptimizer()
313 so.start()
314 so.begin_phase("medium")
315 time.sleep(0.01)
316 so.end_phase("medium")
317 so.begin_phase("fast")
318 so.end_phase("fast")
319 so.end()
320 r = so.report()
321 durations = [p["duration_ms"] for p in r["phases"]]
322 assert durations == sorted(durations, reverse=True)
324 def test_pct_of_total(self):
325 so = StartupOptimizer()
326 so.start()
327 so.begin_phase("a")
328 time.sleep(0.01)
329 so.end_phase("a")
330 so.end()
331 r = so.report()
332 for p in r["phases"]:
333 assert 0 <= p["pct_of_total"] <= 100
335 def test_total_duration_ms(self):
336 so = StartupOptimizer()
337 so.start()
338 time.sleep(0.01)
339 so.end()
340 assert so.total_duration_ms() > 0
342 def test_total_duration_ms_before_end(self):
343 so = StartupOptimizer()
344 so.start()
345 # Accessing before end returns some value (may be negative)
346 d = so.total_duration_ms()
347 assert isinstance(d, float)
349 def test_thread_safety(self):
350 so = StartupOptimizer()
351 so.start()
352 errors = []
354 def worker(idx):
355 try:
356 so.begin_phase(f"p{idx}")
357 so.end_phase(f"p{idx}")
358 except Exception as e:
359 errors.append(e)
361 threads = [threading.Thread(target=worker, args=(i,)) for i in range(20)]
362 for t in threads:
363 t.start()
364 for t in threads:
365 t.join()
366 so.end()
367 assert len(errors) == 0
370# ============================================================================
371# _StartupPhase
372# ============================================================================
374class TestStartupPhase:
375 def test_duration(self):
376 sp = _StartupPhase(name="test", start_time=0.0, end_time=5.0)
377 assert sp.duration == 5.0
379 def test_name(self):
380 sp = _StartupPhase(name="init")
381 assert sp.name == "init"
383 def test_default_metadata(self):
384 sp = _StartupPhase(name="x")
385 assert sp.metadata == {}
388# ============================================================================
389# Convenience Functions
390# ============================================================================
392class TestConvenience:
393 def test_create_lazy_loader(self):
394 ll = create_lazy_loader()
395 assert isinstance(ll, LazyLoader)
397 def test_create_module_preloader(self):
398 mp = create_module_preloader(max_concurrent=8)
399 assert isinstance(mp, ModulePreloader)
400 assert mp._max_concurrent == 8
402 def test_create_module_preloader_default(self):
403 mp = create_module_preloader()
404 assert mp._max_concurrent == 4
406 def test_create_startup_optimizer(self):
407 so = create_startup_optimizer()
408 assert isinstance(so, StartupOptimizer)
410 def test_quick_start(self):
411 result = quick_start(
412 essential_modules=["json"],
413 lazy_modules=["os"],
414 hot_modules=["sys"],
415 )
416 assert "essential" in result
417 assert result["lazy_count"] == 1
418 assert result["cached_total"] == 2
419 assert "total_essential_ms" in result
421 def test_quick_start_no_hot_modules(self):
422 result = quick_start(
423 essential_modules=["json"],
424 lazy_modules=["os"],
425 )
426 assert result["cached_total"] == 1