test(ai_client): TDD red phase - no top-level SDK imports allowed
Phase 3 Task T3.1 of startup_speedup_20260606 track. 9 tests assert:
- import src.ai_client does NOT trigger google.genai / anthropic /
openai / requests / google.genai.types imports (the main thread
must not load these on import; they're warmed on _io_pool)
- _require_warmed(name) helper exists and is callable
- _require_warmed returns the cached module if already in sys.modules
- _require_warmed falls back to importlib for tests/dev where
warmup didn't run
- The static audit script does not see src/ai_client.py as a
contributor of heavy-import violations
All 9 tests are currently FAILING (RED). They will turn GREEN when
T3.2 (the actual refactor of src/ai_client.py to remove top-level
imports and add _require_warmed) lands.
The implementation is held pending MCP client fix (per user instruction).
This commit is contained in:
@@ -0,0 +1,166 @@
|
||||
"""Tests that src/ai_client.py has NO top-level heavy SDK imports.
|
||||
|
||||
Per spec.md:2.2 Layer 1, the main thread's import chain must not include
|
||||
heavy modules. Heavy SDKs (google.genai, anthropic, openai, requests) are
|
||||
warmed on AppController's _io_pool and accessed via _require_warmed().
|
||||
|
||||
These tests run in a fresh subprocess to ensure no warmup state leaks
|
||||
from the test runner. We assert:
|
||||
- The four heavy SDKs are NOT imported as a side effect of
|
||||
`import src.ai_client`
|
||||
- The _require_warmed helper exists
|
||||
- Calling _require_warmed with an un-warmed module works (fallback
|
||||
import via importlib)
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def _run_in_subprocess(snippet: str) -> subprocess.CompletedProcess:
|
||||
script = textwrap.dedent(snippet)
|
||||
return subprocess.run(
|
||||
[sys.executable, "-c", script],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
cwd=str(ROOT),
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
def test_ai_client_does_not_import_google_genai_at_module_level() -> None:
|
||||
res = _run_in_subprocess("""
|
||||
import sys
|
||||
import src.ai_client
|
||||
print('google.genai' in sys.modules)
|
||||
print('google' in sys.modules)
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
lines = res.stdout.strip().splitlines()
|
||||
assert lines == ["False", "False"], f"ai_client triggered google.genai import: {res.stdout}"
|
||||
|
||||
|
||||
def test_ai_client_does_not_import_anthropic_at_module_level() -> None:
|
||||
res = _run_in_subprocess("""
|
||||
import sys
|
||||
import src.ai_client
|
||||
print('anthropic' in sys.modules)
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
assert res.stdout.strip() == "False"
|
||||
|
||||
|
||||
def test_ai_client_does_not_import_openai_at_module_level() -> None:
|
||||
res = _run_in_subprocess("""
|
||||
import sys
|
||||
import src.ai_client
|
||||
print('openai' in sys.modules)
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
assert res.stdout.strip() == "False"
|
||||
|
||||
|
||||
def test_ai_client_does_not_import_requests_at_module_level() -> None:
|
||||
res = _run_in_subprocess("""
|
||||
import sys
|
||||
import src.ai_client
|
||||
print('requests' in sys.modules)
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
assert res.stdout.strip() == "False"
|
||||
|
||||
|
||||
def test_ai_client_does_not_import_google_genai_types_at_module_level() -> None:
|
||||
res = _run_in_subprocess("""
|
||||
import sys
|
||||
import src.ai_client
|
||||
print('google.genai.types' in sys.modules)
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
assert res.stdout.strip() == "False"
|
||||
|
||||
|
||||
def test_ai_client_exposes_require_warmed_helper() -> None:
|
||||
res = _run_in_subprocess("""
|
||||
import src.ai_client
|
||||
print(hasattr(src.ai_client, '_require_warmed'))
|
||||
print(callable(src.ai_client._require_warmed))
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
assert res.stdout.strip() == "True\nTrue"
|
||||
|
||||
|
||||
def test_require_warmed_returns_module_when_already_loaded() -> None:
|
||||
res = _run_in_subprocess("""
|
||||
import sys
|
||||
import importlib
|
||||
import src.ai_client
|
||||
# Pre-load the module (simulating warmup)
|
||||
importlib.import_module('json')
|
||||
mod = src.ai_client._require_warmed('json')
|
||||
print(mod is sys.modules['json'])
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
assert res.stdout.strip() == "True"
|
||||
|
||||
|
||||
def test_require_warmed_falls_back_to_import_if_not_warmed() -> None:
|
||||
res = _run_in_subprocess("""
|
||||
import sys
|
||||
import src.ai_client
|
||||
# Pick a stdlib module that ai_client doesn't trigger
|
||||
assert 'collections' in sys.modules # stdlib
|
||||
# Pick a module that's definitely NOT imported yet
|
||||
target = 'collections.abc'
|
||||
importlib_mod = __import__('importlib')
|
||||
if target in sys.modules:
|
||||
del sys.modules[target]
|
||||
mod = src.ai_client._require_warmed(target)
|
||||
print(mod is not None)
|
||||
print(target in sys.modules)
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
lines = res.stdout.strip().splitlines()
|
||||
assert lines[0] == "True"
|
||||
assert lines[1] == "True"
|
||||
|
||||
|
||||
def test_audit_main_thread_imports_sees_no_new_violation_from_ai_client() -> None:
|
||||
"""Run the static audit and check that src/ai_client.py contributes no
|
||||
new heavy imports (any remaining violations are pre-existing in OTHER
|
||||
files; Phase 4-5 will address them).
|
||||
"""
|
||||
res = _run_in_subprocess("""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
sys.path.insert(0, str(Path.cwd()))
|
||||
from scripts.audit_main_thread_imports import audit, _walk_import_graph
|
||||
from pathlib import Path
|
||||
root = Path('.').resolve()
|
||||
entry = (root / 'sloppy.py').resolve()
|
||||
graph = _walk_import_graph(entry, root)
|
||||
ai_client_path = root / 'src' / 'ai_client.py'
|
||||
if ai_client_path in graph:
|
||||
import ast
|
||||
tree = ast.parse(ai_client_path.read_text(encoding='utf-8'))
|
||||
heavy = ['anthropic', 'google.genai', 'openai', 'requests', 'google.genai.types']
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.Import):
|
||||
for alias in node.names:
|
||||
for h in heavy:
|
||||
if alias.name == h or alias.name.startswith(h + '.'):
|
||||
print('VIOLATION:', alias.name)
|
||||
elif isinstance(node, ast.ImportFrom):
|
||||
if node.module:
|
||||
for h in heavy:
|
||||
if node.module == h or node.module.startswith(h + '.'):
|
||||
print('VIOLATION:', node.module)
|
||||
print('OK')
|
||||
""")
|
||||
assert res.returncode == 0, f"stderr: {res.stderr}"
|
||||
assert "OK" in res.stdout
|
||||
assert "VIOLATION" not in res.stdout
|
||||
Reference in New Issue
Block a user