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