mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-07-08 11:56:59 +00:00
41420c59fc
Slice 2c of the route-domain reorganization (#4082/#4071, per specs/architecture-runtime-inventory.md §6.3). Moves memory_routes.py into routes/memory/, leaving a backward-compat sys.modules shim at the old path. Pure file reorganization, no behavior change. The shim uses sys.modules replacement (same pattern as the merged gallery #4903 and research #4975 slices) so that `import routes.memory_routes`, `from routes.memory_routes import X`, `importlib.import_module(...)`, and the `import ... as mr` + `monkeypatch.setattr(mr, ...)` pattern used by test_memory_routes_session_owner.py / test_memory_owner_isolation.py all operate on the same module object the application uses. The canonical module does NOT depend on the shim — routes/memory/ memory_routes.py imports only from services/, core/, src/, and stdlib (zero internal routes/ coupling). Four source-introspection test sites repointed to the new canonical path: - test_direct_upload_limits.py - test_upload_limits_centralized.py (two dict keys) - test_vision_owner_scope.py Adds tests/test_memory_routes_shim.py to pin the sys.modules shim contract (legacy and canonical paths resolve to the same module object; monkeypatch via legacy alias reaches the canonical module). Verified: compileall clean; full suite 4219 passed, 3 skipped.
102 lines
3.7 KiB
Python
102 lines
3.7 KiB
Python
from pathlib import Path
|
|
|
|
from src import ai_interaction
|
|
from src import document_processor as dp
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def test_configured_vision_model_resolution_passes_owner(monkeypatch):
|
|
seen = []
|
|
|
|
def fake_resolve_model(spec, owner=None):
|
|
seen.append((spec, owner))
|
|
return ("http://example.test/chat/completions", spec, {"Authorization": "Bearer token"})
|
|
|
|
monkeypatch.setattr(ai_interaction, "_resolve_model", fake_resolve_model)
|
|
|
|
assert dp._resolve_vl_model("gpt-4o", owner="alice") == (
|
|
"http://example.test/chat/completions",
|
|
"gpt-4o",
|
|
{"Authorization": "Bearer token"},
|
|
)
|
|
assert seen == [("gpt-4o", "alice")]
|
|
|
|
|
|
def test_auto_detected_vision_model_resolution_passes_owner(monkeypatch):
|
|
seen = []
|
|
|
|
def fake_resolve_model(spec, owner=None):
|
|
seen.append((spec, owner))
|
|
if spec == "llava":
|
|
return ("http://example.test/chat/completions", spec, {})
|
|
raise ValueError("not available")
|
|
|
|
monkeypatch.setattr(ai_interaction, "_resolve_model", fake_resolve_model)
|
|
|
|
assert dp._resolve_vl_model("", owner="alice") == (
|
|
"http://example.test/chat/completions",
|
|
"llava",
|
|
{},
|
|
)
|
|
assert seen
|
|
assert all(owner == "alice" for _spec, owner in seen)
|
|
|
|
|
|
def test_vision_analysis_uses_owner_scoped_primary_and_fallback(monkeypatch, tmp_path):
|
|
seen = {}
|
|
|
|
def fake_resolve_vl_model(configured, owner=None):
|
|
seen["primary"] = (configured, owner)
|
|
return ("http://primary.test/chat/completions", "vision-primary", {"X-Test": "1"})
|
|
|
|
def fake_fallbacks(owner=None):
|
|
seen["fallback_owner"] = owner
|
|
return []
|
|
|
|
def fake_llm_call(url, model, messages, headers=None, timeout=None):
|
|
seen["llm"] = (url, model, headers, timeout, messages)
|
|
return "description"
|
|
|
|
monkeypatch.setattr(dp, "_load_vl_settings", lambda: {"vision_enabled": True, "vision_model": "gpt-4o"})
|
|
monkeypatch.setattr(dp, "_resolve_vl_model", fake_resolve_vl_model)
|
|
monkeypatch.setattr(dp, "llm_call", fake_llm_call)
|
|
|
|
from src import endpoint_resolver
|
|
|
|
monkeypatch.setattr(endpoint_resolver, "resolve_vision_fallback_candidates", fake_fallbacks)
|
|
|
|
image = tmp_path / "image.png"
|
|
image.write_bytes(b"not-a-real-png-but-base64-is-enough")
|
|
|
|
assert dp.analyze_image_with_vl_result(str(image), owner="alice") == {
|
|
"text": "description",
|
|
"model": "vision-primary",
|
|
}
|
|
assert seen["primary"] == ("gpt-4o", "alice")
|
|
assert seen["fallback_owner"] == "alice"
|
|
assert seen["llm"][:4] == (
|
|
"http://primary.test/chat/completions",
|
|
"vision-primary",
|
|
{"X-Test": "1"},
|
|
120,
|
|
)
|
|
|
|
|
|
def test_request_vision_call_sites_pass_owner():
|
|
chat_source = (ROOT / "src" / "chat_handler.py").read_text()
|
|
processor_source = (ROOT / "src" / "document_processor.py").read_text()
|
|
upload_source = (ROOT / "routes" / "upload_routes.py").read_text()
|
|
document_source = (ROOT / "routes" / "document_routes.py").read_text()
|
|
gallery_source = (ROOT / "routes" / "gallery" / "gallery_routes.py").read_text()
|
|
memory_source = (ROOT / "routes" / "memory" / "memory_routes.py").read_text()
|
|
|
|
assert 'analyze_image_with_vl_result(file_info["path"], owner=owner)' in chat_source
|
|
assert "analyze_image_with_vl(path, owner=current_user)" in upload_source
|
|
assert "_process_pdf(path, owner=owner)" in processor_source
|
|
assert "_process_pdf(pdf_path, owner=user)" in document_source
|
|
assert "_resolve_vl_model(vl_model, owner=user)" in document_source
|
|
assert "_resolve_vl_model(configured, owner=user)" in gallery_source
|
|
assert "_process_pdf(tmp_path, owner=_owner(request))" in memory_source
|