289 lines
12 KiB
Python
289 lines
12 KiB
Python
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import importlib.util
|
|||
|
|
import json
|
|||
|
|
import threading
|
|||
|
|
import time
|
|||
|
|
import urllib.error
|
|||
|
|
import urllib.request
|
|||
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
|
|||
|
|
from codexpi.database import Database
|
|||
|
|
from codexpi.local_ai_bridge_service import LocalAIBridgeError, LocalAIBridgeService
|
|||
|
|
|
|||
|
|
|
|||
|
|
MODELS = [
|
|||
|
|
{
|
|||
|
|
"name": "qwen2.5-coder:14b",
|
|||
|
|
"size": 8_988_000_000,
|
|||
|
|
"details": {"parameter_size": "14.8B", "quantization_level": "Q4_K_M"},
|
|||
|
|
"capabilities": ["completion", "tools"],
|
|||
|
|
},
|
|||
|
|
{
|
|||
|
|
"name": "gemma3:12b",
|
|||
|
|
"size": 8_149_000_000,
|
|||
|
|
"details": {"parameter_size": "12.2B", "quantization_level": "Q4_K_M"},
|
|||
|
|
"capabilities": ["completion", "vision"],
|
|||
|
|
},
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def load_bridge_script():
|
|||
|
|
path = Path(__file__).parents[1] / "scripts" / "codexpi_ai_bridge.py"
|
|||
|
|
spec = importlib.util.spec_from_file_location("codexpi_ai_bridge_test", path)
|
|||
|
|
assert spec and spec.loader
|
|||
|
|
module = importlib.util.module_from_spec(spec)
|
|||
|
|
spec.loader.exec_module(module)
|
|||
|
|
return module
|
|||
|
|
|
|||
|
|
|
|||
|
|
class FakeOllamaHandler(BaseHTTPRequestHandler):
|
|||
|
|
protocol_version = "HTTP/1.0"
|
|||
|
|
|
|||
|
|
def log_message(self, *_args) -> None:
|
|||
|
|
return
|
|||
|
|
|
|||
|
|
def _send(self, value: dict, status: int = 200) -> None:
|
|||
|
|
body = json.dumps(value).encode("utf-8")
|
|||
|
|
self.send_response(status)
|
|||
|
|
self.send_header("Content-Type", "application/json")
|
|||
|
|
self.send_header("Content-Length", str(len(body)))
|
|||
|
|
self.end_headers()
|
|||
|
|
self.wfile.write(body)
|
|||
|
|
|
|||
|
|
def do_GET(self) -> None:
|
|||
|
|
if self.path in {"/api/tags", "/v1/models"}:
|
|||
|
|
self._send({"models": MODELS})
|
|||
|
|
elif self.path == "/api/experimental/model-recommendations":
|
|||
|
|
self._send({"recommendations": [{
|
|||
|
|
"model": "qwen3:8b",
|
|||
|
|
"description": "Local reasoning and tools",
|
|||
|
|
"vram_bytes": 6_000_000_000,
|
|||
|
|
"thinking": {"values": [False, True], "default": True},
|
|||
|
|
}]})
|
|||
|
|
else:
|
|||
|
|
self._send({"error": "missing"}, 404)
|
|||
|
|
|
|||
|
|
def do_POST(self) -> None:
|
|||
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|||
|
|
payload = json.loads(self.rfile.read(length) or b"{}")
|
|||
|
|
if self.path == "/v1/chat/completions":
|
|||
|
|
self._send({"choices": [{"message": {"role": "assistant", "content": payload.get("model")}}]})
|
|||
|
|
elif self.path == "/api/show":
|
|||
|
|
self._send({"capabilities": ["completion", "tools"]})
|
|||
|
|
elif self.path == "/api/pull":
|
|||
|
|
rows = [
|
|||
|
|
{"status": "pulling manifest"},
|
|||
|
|
{"status": "pulling layer", "digest": "sha256:one", "total": 100, "completed": 35},
|
|||
|
|
{"status": "pulling layer", "digest": "sha256:one", "total": 100, "completed": 100},
|
|||
|
|
{"status": "verifying sha256 digest"},
|
|||
|
|
{"status": "success"},
|
|||
|
|
]
|
|||
|
|
body = b"".join(json.dumps(item).encode("utf-8") + b"\n" for item in rows)
|
|||
|
|
self.send_response(200)
|
|||
|
|
self.send_header("Content-Type", "application/x-ndjson")
|
|||
|
|
self.send_header("Content-Length", str(len(body)))
|
|||
|
|
self.end_headers()
|
|||
|
|
self.wfile.write(body)
|
|||
|
|
else:
|
|||
|
|
self._send({"error": "missing"}, 404)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture
|
|||
|
|
def fake_ollama():
|
|||
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), FakeOllamaHandler)
|
|||
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|||
|
|
thread.start()
|
|||
|
|
try:
|
|||
|
|
yield server
|
|||
|
|
finally:
|
|||
|
|
server.shutdown()
|
|||
|
|
server.server_close()
|
|||
|
|
thread.join(timeout=2)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_local_ai_pairing_encrypts_token_and_writes_isolated_opencode_config(tmp_path, fake_ollama):
|
|||
|
|
db = Database(tmp_path / "codexpi.db")
|
|||
|
|
db.initialize()
|
|||
|
|
service = LocalAIBridgeService(db, "test-secret", tmp_path / "data")
|
|||
|
|
pairing = service.create_pairing("owner")
|
|||
|
|
token = "a" * 48
|
|||
|
|
bridge = service.pair(
|
|||
|
|
pairing["pairing_id"], pairing["code"], "PC bureau", "Windows",
|
|||
|
|
"127.0.0.1", fake_ollama.server_address[1], token,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
assert bridge["connected"] is True
|
|||
|
|
assert bridge["default_model"] == "qwen2.5-coder:14b"
|
|||
|
|
assert bridge["models"][0]["capabilities"] == ["completion", "tools"]
|
|||
|
|
assert "token" not in bridge and "token_encrypted" not in bridge
|
|||
|
|
stored = db.one("SELECT token_encrypted FROM local_ai_bridges WHERE id=?", (bridge["id"],))
|
|||
|
|
assert stored and stored["token_encrypted"] != token
|
|||
|
|
assert service.token_path.read_text(encoding="utf-8") == token
|
|||
|
|
config = json.loads(service.config_path.read_text(encoding="utf-8"))
|
|||
|
|
assert config["model"] == "ollama/qwen2.5-coder:14b"
|
|||
|
|
assert config["providers"]["ollama"]["settings"]["baseURL"].endswith("/v1")
|
|||
|
|
assert token not in service.config_path.read_text(encoding="utf-8")
|
|||
|
|
|
|||
|
|
with pytest.raises(LocalAIBridgeError, match="déjà utilisé"):
|
|||
|
|
service.pair(
|
|||
|
|
pairing["pairing_id"], pairing["code"], "PC bureau", "Windows",
|
|||
|
|
"127.0.0.1", fake_ollama.server_address[1], token,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
changed = service.configure(bridge["id"], "gemma3:12b")
|
|||
|
|
assert changed["default_model"] == "gemma3:12b"
|
|||
|
|
assert json.loads(service.config_path.read_text(encoding="utf-8"))["model"] == "ollama/gemma3:12b"
|
|||
|
|
service.revoke(bridge["id"])
|
|||
|
|
assert not service.config_path.exists() and not service.token_path.exists()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_local_ai_service_rejects_public_addresses(tmp_path):
|
|||
|
|
db = Database(tmp_path / "codexpi.db")
|
|||
|
|
db.initialize()
|
|||
|
|
service = LocalAIBridgeService(db, "test-secret", tmp_path / "data")
|
|||
|
|
pairing = service.create_pairing("owner")
|
|||
|
|
with pytest.raises(LocalAIBridgeError, match="réseau local"):
|
|||
|
|
service.pair(pairing["pairing_id"], pairing["code"], "PC", "Windows", "8.8.8.8", 11435, "x" * 48)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_invalid_pairing_code_never_triggers_network_probe(tmp_path, monkeypatch):
|
|||
|
|
db = Database(tmp_path / "codexpi.db")
|
|||
|
|
db.initialize()
|
|||
|
|
service = LocalAIBridgeService(db, "test-secret", tmp_path / "data")
|
|||
|
|
pairing = service.create_pairing("owner")
|
|||
|
|
probed = False
|
|||
|
|
|
|||
|
|
def fail_if_probed(*_args, **_kwargs):
|
|||
|
|
nonlocal probed
|
|||
|
|
probed = True
|
|||
|
|
raise AssertionError("the bridge must not be contacted")
|
|||
|
|
|
|||
|
|
monkeypatch.setattr(service, "probe", fail_if_probed)
|
|||
|
|
with pytest.raises(LocalAIBridgeError, match="Code d’association invalide"):
|
|||
|
|
service.pair(
|
|||
|
|
pairing["pairing_id"], "wrong-pairing-code", "PC", "Windows",
|
|||
|
|
"127.0.0.1", 11435, "x" * 48,
|
|||
|
|
)
|
|||
|
|
assert probed is False
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_pc_bridge_requires_bearer_and_only_proxies_ollama_routes(fake_ollama):
|
|||
|
|
bridge = load_bridge_script()
|
|||
|
|
upstream = f"http://127.0.0.1:{fake_ollama.server_address[1]}"
|
|||
|
|
server = bridge.create_server("127.0.0.1", 0, upstream, "private-token")
|
|||
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|||
|
|
thread.start()
|
|||
|
|
base = f"http://127.0.0.1:{server.server_address[1]}"
|
|||
|
|
try:
|
|||
|
|
with pytest.raises(urllib.error.HTTPError) as unauthorized:
|
|||
|
|
urllib.request.urlopen(base + "/api/tags", timeout=2)
|
|||
|
|
assert unauthorized.value.code == 401
|
|||
|
|
|
|||
|
|
request = urllib.request.Request(base + "/api/tags", headers={"Authorization": "Bearer private-token"})
|
|||
|
|
with urllib.request.urlopen(request, timeout=2) as response:
|
|||
|
|
assert json.loads(response.read())["models"][0]["name"] == "qwen2.5-coder:14b"
|
|||
|
|
|
|||
|
|
forbidden = urllib.request.Request(base + "/api/delete", headers={"Authorization": "Bearer private-token"})
|
|||
|
|
with pytest.raises(urllib.error.HTTPError) as missing:
|
|||
|
|
urllib.request.urlopen(forbidden, timeout=2)
|
|||
|
|
assert missing.value.code == 404
|
|||
|
|
|
|||
|
|
chat = urllib.request.Request(
|
|||
|
|
base + "/v1/chat/completions",
|
|||
|
|
data=json.dumps({"model": "qwen2.5-coder:14b", "messages": []}).encode("utf-8"),
|
|||
|
|
method="POST",
|
|||
|
|
headers={"Authorization": "Bearer private-token", "Content-Type": "application/json"},
|
|||
|
|
)
|
|||
|
|
with urllib.request.urlopen(chat, timeout=2) as response:
|
|||
|
|
assert json.loads(response.read())["choices"][0]["message"]["content"] == "qwen2.5-coder:14b"
|
|||
|
|
|
|||
|
|
pull = urllib.request.Request(
|
|||
|
|
base + "/api/pull",
|
|||
|
|
data=json.dumps({"model": "qwen3:8b", "stream": True}).encode("utf-8"),
|
|||
|
|
method="POST",
|
|||
|
|
headers={"Authorization": "Bearer private-token", "Content-Type": "application/json"},
|
|||
|
|
)
|
|||
|
|
with urllib.request.urlopen(pull, timeout=2) as response:
|
|||
|
|
events = [json.loads(line) for line in response if line.strip()]
|
|||
|
|
assert events[-1]["status"] == "success"
|
|||
|
|
finally:
|
|||
|
|
server.shutdown()
|
|||
|
|
server.server_close()
|
|||
|
|
thread.join(timeout=2)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_local_ai_settings_flow_exposes_models_without_exposing_token(logged_in, monkeypatch):
|
|||
|
|
client, app, _, _, headers = logged_in
|
|||
|
|
monkeypatch.setattr(app.state.local_ai, "probe", lambda *_args, **_kwargs: app.state.local_ai._models({"models": MODELS}))
|
|||
|
|
|
|||
|
|
created = client.post("/api/local-ai/pairings", headers=headers)
|
|||
|
|
assert created.status_code == 201
|
|||
|
|
pairing = created.json()
|
|||
|
|
paired = client.post("/api/local-ai/agent/pair", json={
|
|||
|
|
"pairing_id": pairing["pairing_id"],
|
|||
|
|
"code": pairing["code"],
|
|||
|
|
"name": "RTX 3060",
|
|||
|
|
"platform": "Windows",
|
|||
|
|
"address": "127.0.0.1",
|
|||
|
|
"port": 11435,
|
|||
|
|
"token": "z" * 48,
|
|||
|
|
})
|
|||
|
|
assert paired.status_code == 201
|
|||
|
|
bridge = paired.json()
|
|||
|
|
assert bridge["default_model"] == "qwen2.5-coder:14b"
|
|||
|
|
assert "token" not in json.dumps(bridge)
|
|||
|
|
|
|||
|
|
listed = client.get("/api/local-ai/bridges")
|
|||
|
|
assert listed.status_code == 200 and listed.json()[0]["name"] == "RTX 3060"
|
|||
|
|
options = client.get("/api/codex/options").json()
|
|||
|
|
assert options["local_default_model"] == "ollama/qwen2.5-coder:14b"
|
|||
|
|
assert options["local_models"][0]["id"] == "ollama/qwen2.5-coder:14b"
|
|||
|
|
assert next(item for item in options["engines"] if item["id"] == "opencode")["label"] == "OpenCode · PC local"
|
|||
|
|
assert client.get("/api/local-ai/connector").status_code == 200
|
|||
|
|
assert client.get("/api/local-ai/guide").status_code == 200
|
|||
|
|
|
|||
|
|
catalog = client.get(f"/api/local-ai/bridges/{bridge['id']}/catalog")
|
|||
|
|
assert catalog.status_code == 200
|
|||
|
|
assert any(item["name"] == "qwen2.5-coder:14b" and item["installed"] for item in catalog.json()["models"])
|
|||
|
|
|
|||
|
|
updated = client.put(
|
|||
|
|
f"/api/local-ai/bridges/{bridge['id']}", headers=headers,
|
|||
|
|
json={"default_model": "gemma3:12b", "active": True},
|
|||
|
|
)
|
|||
|
|
assert updated.status_code == 200 and updated.json()["default_model"] == "gemma3:12b"
|
|||
|
|
revoked = client.delete(f"/api/local-ai/bridges/{bridge['id']}", headers=headers)
|
|||
|
|
assert revoked.status_code == 200
|
|||
|
|
assert client.get("/api/local-ai/bridges").json() == []
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_local_ai_pull_tracks_real_stream_progress(tmp_path, fake_ollama):
|
|||
|
|
db = Database(tmp_path / "codexpi.db")
|
|||
|
|
db.initialize()
|
|||
|
|
service = LocalAIBridgeService(db, "test-secret", tmp_path / "data")
|
|||
|
|
pairing = service.create_pairing("owner")
|
|||
|
|
bridge = service.pair(
|
|||
|
|
pairing["pairing_id"], pairing["code"], "PC bureau", "Windows",
|
|||
|
|
"127.0.0.1", fake_ollama.server_address[1], "p" * 48,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
catalog = service.catalog(bridge["id"])
|
|||
|
|
recommended = next(item for item in catalog["models"] if item["name"] == "qwen3:8b")
|
|||
|
|
assert recommended["recommended"] is True
|
|||
|
|
assert recommended["fit"] in {"ideal", "possible", "heavy"}
|
|||
|
|
|
|||
|
|
job = service.start_pull(bridge["id"], "qwen3:8b")
|
|||
|
|
deadline = time.monotonic() + 3
|
|||
|
|
while time.monotonic() < deadline:
|
|||
|
|
current = service.list_pulls(bridge["id"])[0]
|
|||
|
|
if current["status"] not in {"queued", "downloading", "verifying"}:
|
|||
|
|
break
|
|||
|
|
time.sleep(0.02)
|
|||
|
|
assert current["status"] == "completed"
|
|||
|
|
assert current["progress"] == 100.0
|
|||
|
|
assert current["total"] == 100
|
|||
|
|
assert json.loads(service.pulls_path.read_text(encoding="utf-8"))[0]["status"] == "completed"
|