84 lines
4.7 KiB
Python
84 lines
4.7 KiB
Python
"""Exercise the shipped mobile adapter against real IDE HTTP/WS/DB code.
|
||
|
||
Only the paid inference endpoint is replaced by a local deterministic provider.
|
||
This proves wiring, not real-provider quality or Android hardware behavior.
|
||
"""
|
||
import json
|
||
import os
|
||
import shutil
|
||
import socket
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
|
||
def test_mobile_reads_and_submits_through_real_runtime(tmp_path, monkeypatch):
|
||
node = shutil.which('node')
|
||
if not node:
|
||
pytest.fail('Node is required for mobile integration')
|
||
calls = []
|
||
|
||
class Provider(BaseHTTPRequestHandler):
|
||
def log_message(self, *_):
|
||
pass
|
||
|
||
def do_POST(self):
|
||
body = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
|
||
calls.append(body)
|
||
message = {'role': 'assistant', 'content': 'MOBIL_RUNTIME_OK'}
|
||
response = {'id': 'fixture', 'object': 'chat.completion', 'created': 1,
|
||
'model': 'mobile-fixture', 'choices': [{'index': 0, 'message': message, 'finish_reason': 'stop'}],
|
||
'usage': {'prompt_tokens': 10, 'completion_tokens': 5, 'total_tokens': 15}}
|
||
if body.get('stream'):
|
||
self.send_response(200); self.send_header('Content-Type', 'text/event-stream'); self.end_headers()
|
||
for delta, finish in [(message, None), ({}, 'stop')]:
|
||
chunk = {**response, 'object': 'chat.completion.chunk', 'choices': [{'index': 0, 'delta': delta, 'finish_reason': finish}]}
|
||
self.wfile.write(('data: ' + json.dumps(chunk) + '\n\n').encode())
|
||
self.wfile.write(b'data: [DONE]\n\n'); self.wfile.flush()
|
||
else:
|
||
raw = json.dumps(response).encode()
|
||
self.send_response(200); self.send_header('Content-Type', 'application/json'); self.send_header('Content-Length', str(len(raw))); self.end_headers(); self.wfile.write(raw)
|
||
|
||
provider = ThreadingHTTPServer(('127.0.0.1', 0), Provider)
|
||
provider_thread = threading.Thread(target=provider.serve_forever, daemon=True); provider_thread.start()
|
||
home = Path(os.environ['HERMES_HOME'])
|
||
base_url = f'http://127.0.0.1:{provider.server_port}/v1'
|
||
config = {'model': {'default': 'mobile-fixture', 'provider': 'custom:mobile-fixture'},
|
||
'custom_providers': [{'name': 'mobile-fixture', 'base_url': base_url, 'api_key': 'local-fixture-only', 'api_mode': 'chat_completions'}],
|
||
'agent': {'max_iterations': 2}, 'memory': {'memory_enabled': False, 'user_profile_enabled': False}}
|
||
(home / 'config.yaml').write_text(json.dumps(config), encoding='utf-8')
|
||
monkeypatch.setenv('HERMES_DASHBOARD_SESSION_TOKEN', 'local-mobile-integration-token')
|
||
monkeypatch.setenv('HERMES_SERVE_HEADLESS', '1')
|
||
from hermes_state import SessionDB
|
||
db = SessionDB()
|
||
db.create_session('mobile-integration-session', 'desktop', model='mobile-fixture',
|
||
model_config={'model': 'mobile-fixture', 'provider': 'custom:mobile-fixture', 'base_url': base_url}, cwd=str(tmp_path))
|
||
db.append_message('mobile-integration-session', 'user', 'Yerel bağlantı sınaması.')
|
||
db.append_message('mobile-integration-session', 'assistant', 'Hazır.')
|
||
from hermes_cli import web_server
|
||
import uvicorn
|
||
bound = socket.socket(); bound.bind(('127.0.0.1', 0)); port = bound.getsockname()[1]
|
||
server = uvicorn.Server(uvicorn.Config(web_server.app, host='127.0.0.1', port=port, lifespan='off', log_level='error'))
|
||
thread = threading.Thread(target=lambda: server.run(sockets=[bound]), daemon=True); thread.start()
|
||
try:
|
||
for _ in range(100):
|
||
if server.started:
|
||
break
|
||
time.sleep(.1)
|
||
assert server.started
|
||
root = Path(__file__).resolve().parents[2]
|
||
result = subprocess.run([node, str(root / 'apps/mobile/scripts/runtime-driver.mjs'), str(port), str(tmp_path / 'journal')],
|
||
cwd=root, capture_output=True, text=True, encoding='utf-8', timeout=100)
|
||
assert result.returncode == 0, result.stdout + result.stderr
|
||
assert 'MOBIL_RUNTIME_OK' in result.stdout
|
||
delivered = [message for call in calls for message in call.get('messages', []) if message.get('role') == 'user' and 'MOBIL_GOREV_20260908' in str(message.get('content'))]
|
||
assert delivered
|
||
history = db.get_messages('mobile-integration-session')
|
||
assert sum('MOBIL_GOREV_20260908' in str(row.get('content')) for row in history if row['role'] == 'user') == 1
|
||
finally:
|
||
server.should_exit = True; thread.join(10); bound.close(); provider.shutdown(); provider.server_close(); db.close()
|