328 lines
12 KiB
Python
328 lines
12 KiB
Python
"""Unit tests for the Canvas bridge route."""
|
|
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from types import SimpleNamespace
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from routes import canvas
|
|
|
|
|
|
class _Handler:
|
|
command = "GET"
|
|
|
|
def __init__(self, body=b""):
|
|
self.status = None
|
|
self.headers_out = {}
|
|
self.body = b""
|
|
self.headers = {
|
|
"Content-Type": "application/json",
|
|
"Content-Length": str(len(body)),
|
|
}
|
|
self.rfile = _Reader(body)
|
|
|
|
def send_response(self, status):
|
|
self.status = status
|
|
|
|
def send_header(self, key, value):
|
|
self.headers_out[key] = value
|
|
|
|
def end_headers(self):
|
|
pass
|
|
|
|
@property
|
|
def wfile(self):
|
|
outer = self
|
|
|
|
class _W:
|
|
def write(self, body):
|
|
outer.body += body
|
|
|
|
return _W()
|
|
|
|
|
|
class _Reader:
|
|
def __init__(self, body):
|
|
self._body = body
|
|
|
|
def read(self, _length):
|
|
return self._body
|
|
|
|
|
|
def _parsed(query):
|
|
return SimpleNamespace(query=query)
|
|
|
|
|
|
def test_register_wires_canvas_routes():
|
|
api = MagicMock()
|
|
canvas.register(api)
|
|
calls = {(c.args[1], c.args[0]) for c in api.register_route.call_args_list}
|
|
assert ("GET", "/api/canvas/status") in calls
|
|
assert ("GET", "/api/canvas/proxy") in calls
|
|
assert ("POST", "/api/canvas/proxy") in calls
|
|
assert ("PUT", "/api/canvas/proxy") in calls
|
|
assert ("POST", "/api/canvas/command") in calls
|
|
assert ("GET", "/api/canvas/tools") in calls
|
|
assert ("GET", "/api/canvas/design-context") in calls
|
|
assert ("GET", "/api/canvas/events") in calls
|
|
|
|
|
|
def test_proxy_blocks_unlisted_path():
|
|
h = _Handler()
|
|
assert canvas._handle_proxy(h, _parsed("path=/api/v1/admin")) is True
|
|
assert h.status == 403
|
|
assert json.loads(h.body)["error"] == "path not allowed: /api/v1/admin"
|
|
|
|
|
|
def test_proxy_forwards_allowed_projects_path():
|
|
h = _Handler()
|
|
response = MagicMock()
|
|
response.__enter__.return_value.status = 200
|
|
response.__enter__.return_value.headers = {"Content-Type": "application/json"}
|
|
response.__enter__.return_value.read.return_value = b"[]"
|
|
with patch("routes.canvas.urllib.request.urlopen", return_value=response) as urlopen:
|
|
assert canvas._handle_proxy(h, _parsed("path=/api/v1/projects")) is True
|
|
assert h.status == 200
|
|
assert h.body == b"[]"
|
|
assert urlopen.call_args.args[0].full_url.endswith("/api/v1/projects")
|
|
|
|
|
|
def test_proxy_forwards_capabilities_path():
|
|
h = _Handler()
|
|
response = MagicMock()
|
|
response.__enter__.return_value.status = 200
|
|
response.__enter__.return_value.headers = {"Content-Type": "application/json"}
|
|
response.__enter__.return_value.read.return_value = b'{"projects":true}'
|
|
with patch("routes.canvas.urllib.request.urlopen", return_value=response) as urlopen:
|
|
assert canvas._handle_proxy(h, _parsed("path=/api/v1/capabilities")) is True
|
|
assert h.status == 200
|
|
assert h.body == b'{"projects":true}'
|
|
assert urlopen.call_args.args[0].full_url.endswith("/api/v1/capabilities")
|
|
|
|
|
|
def test_proxy_post_method_override_forwards_put_without_upstream_put_route():
|
|
h = _Handler(b'{"spec":{"schemaVersion":1}}')
|
|
h.command = "POST"
|
|
response = MagicMock()
|
|
response.__enter__.return_value.status = 200
|
|
response.__enter__.return_value.headers = {"Content-Type": "application/json"}
|
|
response.__enter__.return_value.read.return_value = b'{"id":"screen-1"}'
|
|
with patch("routes.canvas.urllib.request.urlopen", return_value=response) as urlopen:
|
|
assert canvas._handle_proxy(
|
|
h,
|
|
_parsed("path=/api/v1/projects/project-1/screens/screen-1&method=PUT"),
|
|
) is True
|
|
assert h.status == 200
|
|
req = urlopen.call_args.args[0]
|
|
assert req.full_url.endswith("/api/v1/projects/project-1/screens/screen-1")
|
|
assert req.get_method() == "PUT"
|
|
|
|
|
|
def test_proxy_allows_variant_and_prototype_contract_paths():
|
|
assert canvas._is_allowed("/api/v1/projects/project-1/variants")
|
|
assert canvas._is_allowed("/api/v1/projects/project-1/prototype-edges")
|
|
assert canvas._is_allowed("/api/v1/projects/project-1/export")
|
|
|
|
|
|
def test_status_includes_canva_editor_capabilities():
|
|
h = _Handler()
|
|
health = MagicMock()
|
|
health.__enter__.return_value.status = 200
|
|
health.__enter__.return_value.headers = {"Content-Type": "application/json"}
|
|
health.__enter__.return_value.read.return_value = b'{"status":"ok"}'
|
|
caps = MagicMock()
|
|
caps.__enter__.return_value.status = 200
|
|
caps.__enter__.return_value.headers = {"Content-Type": "application/json"}
|
|
caps.__enter__.return_value.read.return_value = b'{"projects":true,"variants":false}'
|
|
|
|
with patch("routes.canvas.urllib.request.urlopen", side_effect=[health, caps]):
|
|
assert canvas._handle_status(h, None) is True
|
|
|
|
payload = json.loads(h.body)
|
|
assert h.status == 200
|
|
assert payload["ok"] is True
|
|
assert payload["capabilities"] == {"projects": True, "variants": False}
|
|
|
|
|
|
def test_design_context_returns_seed_contract():
|
|
h = _Handler()
|
|
with patch("routes.canvas._load_secondbrain_hints", return_value={"status": "unconfigured", "hints": []}):
|
|
assert canvas._handle_design_context(h, _parsed("brand_id=planb")) is True
|
|
assert h.status == 200
|
|
payload = json.loads(h.body)
|
|
assert payload["brand_id"] == "planb"
|
|
assert payload["resolved_brand_id"] == "607740e8-8b76-4455-9c4f-eadbe9b4168c"
|
|
|
|
|
|
def test_design_context_reads_cmo_brand_profile_cache(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "cmo.db"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute(
|
|
"""CREATE TABLE brand_profile_cache (
|
|
brand_id TEXT PRIMARY KEY,
|
|
json TEXT,
|
|
version TEXT,
|
|
fetched_at TEXT
|
|
)"""
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO brand_profile_cache (brand_id, json, version, fetched_at) VALUES (?, ?, ?, ?)",
|
|
(
|
|
"607740e8-8b76-4455-9c4f-eadbe9b4168c",
|
|
json.dumps({
|
|
"brand": {"name": "planb", "displayName": "Plan B Brand", "description": "Dark mode brand."},
|
|
"brand_guideline": "Use confident fr-CA food copy. Never imply industrial food.",
|
|
"design_md": "# DESIGN\nUse Montserrat.",
|
|
"palette": [
|
|
{"path": "color.primary", "type": "color", "valueJson": "{\"dark\":\"#FD5000\"}"},
|
|
],
|
|
"typography": [{"path": "font.family.body", "valueJson": "\"Montserrat\""}],
|
|
"spacing": [{"path": "spacing.4", "valueJson": "4"}],
|
|
"voice": {"dos": ["Be specific"], "donts": ["No scarcity"]},
|
|
"tokens_imported": True,
|
|
}),
|
|
"v1",
|
|
"2026-05-27T12:00:00Z",
|
|
),
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
monkeypatch.setenv("CANVAS_CMO_DB", str(db_path))
|
|
|
|
h = _Handler()
|
|
with patch("routes.canvas._load_secondbrain_hints", return_value={
|
|
"status": "ready",
|
|
"hints": ["Secondbrain capsule 42: Use calm operator UI patterns."],
|
|
}):
|
|
assert canvas._handle_design_context(h, _parsed("brand_id=planb")) is True
|
|
|
|
payload = json.loads(h.body)
|
|
assert h.status == 200
|
|
assert payload["status"] == "ready"
|
|
assert payload["source"] == "cmo.brand_profile_cache"
|
|
assert payload["cache_brand_id"] == "607740e8-8b76-4455-9c4f-eadbe9b4168c"
|
|
assert payload["source_version"] == "v1"
|
|
assert payload["brand"]["displayName"] == "Plan B Brand"
|
|
assert payload["brand_guideline_text"].startswith("Use confident")
|
|
assert payload["palette_tokens"][0]["value"]["dark"] == "#FD5000"
|
|
assert "Be specific" in payload["voice_rules"]
|
|
assert any("BTE brand guideline" in hint for hint in payload["memory_hints"])
|
|
assert any("Secondbrain capsule 42" in hint for hint in payload["memory_hints"])
|
|
assert payload["secondbrain"]["status"] == "ready"
|
|
|
|
|
|
def test_design_context_reports_missing_cache_without_inventing_context(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "cmo.db"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute(
|
|
"""CREATE TABLE brand_profile_cache (
|
|
brand_id TEXT PRIMARY KEY,
|
|
json TEXT,
|
|
version TEXT,
|
|
fetched_at TEXT
|
|
)"""
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
monkeypatch.setenv("CANVAS_CMO_DB", str(db_path))
|
|
|
|
h = _Handler()
|
|
assert canvas._handle_design_context(h, _parsed("brand_id=missing-brand")) is True
|
|
|
|
payload = json.loads(h.body)
|
|
assert payload["status"] == "missing"
|
|
assert payload["brand_guideline_text"] == ""
|
|
assert payload["memory_hints"] == []
|
|
assert "missing-brand" in payload["error"]
|
|
|
|
|
|
def test_events_returns_sse_seed():
|
|
h = _Handler()
|
|
assert canvas._handle_events(h, None) is True
|
|
assert h.status == 200
|
|
assert h.headers_out["Content-Type"].startswith("text/event-stream")
|
|
assert b"canvas.connected" in h.body
|
|
|
|
|
|
def test_command_generates_persists_and_replays_events():
|
|
body = json.dumps({"prompt": "Plan B dashboard", "variants": 2}).encode("utf-8")
|
|
h = _Handler(body)
|
|
h.command = "POST"
|
|
|
|
with patch("routes.canvas._ensure_project", return_value={"id": "project-1", "name": "Hermes Canvas"}), \
|
|
patch("routes.canvas._load_design_context", return_value={
|
|
"brand_id": "planb",
|
|
"resolved_brand_id": "607740e8-8b76-4455-9c4f-eadbe9b4168c",
|
|
"status": "ready",
|
|
"source": "cmo.brand_profile_cache",
|
|
"source_version": "v1",
|
|
"fetched_at": "2026-05-27T12:00:00Z",
|
|
"brand": {"displayName": "Plan B Brand"},
|
|
"memory_hints": ["Use BTE palette tokens: #FD5000"],
|
|
"design_md_excerpt": "Use Montserrat.",
|
|
"tokens_imported": True,
|
|
}), \
|
|
patch("routes.canvas._canva_json") as canva_json:
|
|
canva_json.side_effect = [
|
|
{"screenSpec": {"root": {"type": "text", "text": "A"}}},
|
|
{"id": "screen-1", "name": "Plan B dashboard", "spec": {"root": {"type": "text", "text": "A"}}},
|
|
{"screenSpec": {"root": {"type": "text", "text": "B"}}},
|
|
{"id": "screen-2", "name": "Plan B dashboard", "spec": {"root": {"type": "text", "text": "B"}}},
|
|
{"id": "variant-1"},
|
|
]
|
|
assert canvas._handle_command(h, None) is True
|
|
|
|
payload = json.loads(h.body)
|
|
assert h.status == 200
|
|
assert payload["ok"] is True
|
|
assert payload["design_context"]["status"] == "ready"
|
|
assert len(payload["screens"]) == 2
|
|
assert any(event["type"] == "canvas.screen.persisted" for event in payload["events"])
|
|
first_generate_payload = canva_json.call_args_list[0].args[2]
|
|
assert "BTE/CMO design constraints" in first_generate_payload["prompt"]
|
|
assert "#FD5000" in first_generate_payload["prompt"]
|
|
|
|
events = _Handler()
|
|
assert canvas._handle_events(events, _parsed("format=json&since=0")) is True
|
|
replay = json.loads(events.body)
|
|
assert replay["ok"] is True
|
|
assert any(event["type"] == "canvas.command.completed" for event in replay["events"])
|
|
|
|
|
|
def test_compose_generation_prompt_keeps_prompt_plain_when_context_missing():
|
|
assert canvas._compose_generation_prompt("Make a landing page", {"status": "missing"}) == "Make a landing page"
|
|
|
|
|
|
def test_secondbrain_hints_query_uses_local_psql(monkeypatch):
|
|
monkeypatch.delenv("SECONDBRAIN_DATABASE_URL", raising=False)
|
|
monkeypatch.setenv("SECONDBRAIN_DB_PASSWORD", "test-password")
|
|
completed = SimpleNamespace(
|
|
returncode=0,
|
|
stdout=json.dumps([
|
|
{
|
|
"id": 42,
|
|
"title": "Headless everything for personal AI",
|
|
"summary": "Keep the command center sovereign and web-native.",
|
|
"sector": "svrnty",
|
|
"created_at": "2026-05-27T12:00:00Z",
|
|
}
|
|
]),
|
|
stderr="",
|
|
)
|
|
|
|
with patch("routes.canvas.subprocess.run", return_value=completed) as run:
|
|
result = canvas._load_secondbrain_hints({"displayName": "Plan B"}, "brand-1")
|
|
|
|
assert result["status"] == "ready"
|
|
assert result["source"] == "secondbrain.knowledge_capsules"
|
|
assert result["capsules"][0]["id"] == 42
|
|
assert "command center" in result["hints"][0]
|
|
cmd = run.call_args.args[0]
|
|
assert cmd[0] == "psql"
|
|
assert "-tA" in cmd
|