153 lines
5.7 KiB
Python
153 lines
5.7 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from codexpi.browser_service import BrowserError, BrowserService, screenshot_attachments
|
|
from codexpi.database import Database, utcnow
|
|
|
|
|
|
class FakeBrowser(BrowserService):
|
|
def __init__(self, db: Database, root: Path):
|
|
super().__init__(db, "http://browser.invalid", root)
|
|
self.url = "about:blank"
|
|
self.session_payload = None
|
|
self.interactive_stopped = False
|
|
|
|
def _ensure_interactive_chromium(self):
|
|
return None
|
|
|
|
def _stop_interactive_chromium(self):
|
|
self.interactive_stopped = True
|
|
|
|
def _request(self, method, path, payload=None, timeout=60):
|
|
if path == "/status":
|
|
return {"ready": True}
|
|
if method == "POST" and path == "/session":
|
|
self.session_payload = payload
|
|
return {"sessionId": "session-1"}
|
|
if path.endswith("/url") and method == "POST":
|
|
self.url = payload["url"]
|
|
return None
|
|
if path.endswith("/url"):
|
|
return self.url
|
|
if path.endswith("/screenshot"):
|
|
return base64.b64encode(b"\x89PNG\r\n\x1a\nimage").decode()
|
|
return None
|
|
|
|
|
|
class RecoveringBrowser(FakeBrowser):
|
|
def __init__(self, db: Database, root: Path):
|
|
super().__init__(db, root)
|
|
self.first_session = True
|
|
self.restarted = False
|
|
|
|
def _request(self, method, path, payload=None, timeout=60):
|
|
if method == "POST" and path == "/session" and self.first_session:
|
|
self.first_session = False
|
|
raise BrowserError("session not created: DevToolsActivePort file doesn't exist")
|
|
return super()._request(method, path, payload, timeout)
|
|
|
|
def _restart_browser_container(self):
|
|
self.restarted = True
|
|
|
|
|
|
class UnavailableBrowser(RecoveringBrowser):
|
|
def _request(self, method, path, payload=None, timeout=60):
|
|
if method == "POST" and path == "/session" and self.first_session:
|
|
self.first_session = False
|
|
raise BrowserError("Le service du navigateur intégré est indisponible")
|
|
return FakeBrowser._request(self, method, path, payload, timeout)
|
|
|
|
|
|
def browser(tmp_path: Path) -> tuple[FakeBrowser, Database, str]:
|
|
db = Database(tmp_path / "db.sqlite")
|
|
db.initialize()
|
|
project_id = "4c4c95da-7404-41db-aa9f-31d6bf13c7fc"
|
|
db.execute("INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)", (project_id, "Demo", str(tmp_path), utcnow()))
|
|
return FakeBrowser(db, tmp_path / "screenshots"), db, project_id
|
|
|
|
|
|
def add_project(db: Database, root: Path, name: str = "Second") -> str:
|
|
project_id = "d57866c2-b8e4-4fc3-a897-d67d61a8ba33"
|
|
db.execute("INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)", (project_id, name, str(root / name), utcnow()))
|
|
return project_id
|
|
|
|
|
|
def test_url_validation_rejects_privileged_schemes():
|
|
assert BrowserService.validate_url("example.com") == "https://example.com"
|
|
with pytest.raises(BrowserError):
|
|
BrowserService.validate_url("file:///etc/passwd")
|
|
|
|
|
|
def test_session_navigation_and_screenshot_are_persisted(tmp_path):
|
|
service, db, project_id = browser(tmp_path)
|
|
assert service.navigate(project_id, "example.com")["url"] == "https://example.com"
|
|
shot = service.screenshot(project_id, "Résultat", "queue-1")
|
|
assert Path(shot["path"]).is_file()
|
|
assert screenshot_attachments(str(db.path), "queue-1")[0]["url"].endswith(shot["id"])
|
|
state = service.status(project_id)
|
|
assert state["running"] is True
|
|
assert state["current_url"] == "https://example.com"
|
|
|
|
|
|
def test_webdriver_attaches_to_interactive_chromium(tmp_path):
|
|
service, _, project_id = browser(tmp_path)
|
|
|
|
service.ensure(project_id)
|
|
|
|
options = service.session_payload["capabilities"]["alwaysMatch"]["goog:chromeOptions"]
|
|
assert options == {"debuggerAddress": "127.0.0.1:9222"}
|
|
|
|
|
|
def test_handoff_blocks_codex_navigation(tmp_path):
|
|
service, _, project_id = browser(tmp_path)
|
|
service.ensure(project_id)
|
|
service.set_control(project_id, "user", "Connectez-vous")
|
|
with pytest.raises(BrowserError):
|
|
service.navigate(project_id, "https://example.com")
|
|
assert service.navigate(project_id, "https://example.com", require_codex=False)["url"] == "https://example.com"
|
|
|
|
|
|
def test_browser_session_and_handoff_are_shared_between_projects(tmp_path):
|
|
service, db, first_project = browser(tmp_path)
|
|
second_project = add_project(db, tmp_path)
|
|
|
|
assert service.navigate(first_project, "https://example.com")["url"] == "https://example.com"
|
|
assert service.status(second_project)["running"] is True
|
|
assert service.status(second_project)["current_url"] == "https://example.com"
|
|
|
|
service.set_control(first_project, "user", "Action sensible")
|
|
|
|
assert service.status(second_project)["control_mode"] == "user"
|
|
assert service.status(second_project)["handoff_message"] == "Action sensible"
|
|
with pytest.raises(BrowserError):
|
|
service.navigate(second_project, "https://openai.com")
|
|
|
|
|
|
def test_reset_closes_the_shared_interactive_browser(tmp_path):
|
|
service, _, project_id = browser(tmp_path)
|
|
service.ensure(project_id)
|
|
|
|
service.reset(project_id)
|
|
|
|
assert service.interactive_stopped is True
|
|
|
|
|
|
def test_stale_chromium_profile_is_recovered_once(tmp_path):
|
|
_, db, project_id = browser(tmp_path)
|
|
service = RecoveringBrowser(db, tmp_path / "recovery-screenshots")
|
|
|
|
assert service.ensure(project_id) == "session-1"
|
|
assert service.restarted is True
|
|
|
|
|
|
def test_unavailable_selenium_is_recovered_once(tmp_path):
|
|
_, db, project_id = browser(tmp_path)
|
|
service = UnavailableBrowser(db, tmp_path / "unavailable-screenshots")
|
|
|
|
assert service.ensure(project_id) == "session-1"
|
|
assert service.restarted is True
|