test(e2e): add autocomplete workflow tests
This commit is contained in:
teernisse
250
tests/e2e/test_skills_endpoint.py
Normal file
250
tests/e2e/test_skills_endpoint.py
Normal file
@@ -0,0 +1,250 @@
|
||||
"""E2E tests for the /api/skills endpoint.
|
||||
|
||||
Spins up a real AMC server on a random port and verifies the skills API
|
||||
returns correct data for Claude and Codex agents, including trigger
|
||||
characters, alphabetical sorting, and response format.
|
||||
"""
|
||||
|
||||
import json
|
||||
import socket
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
import urllib.request
|
||||
from http.server import ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
from amc_server.handler import AMCHandler
|
||||
|
||||
|
||||
def _find_free_port():
|
||||
"""Find an available port for the test server."""
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||||
s.bind(("127.0.0.1", 0))
|
||||
return s.getsockname()[1]
|
||||
|
||||
|
||||
def _get_json(url):
|
||||
"""Fetch JSON from a URL, returning (status_code, parsed_json)."""
|
||||
req = urllib.request.Request(url)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||||
return resp.status, json.loads(resp.read())
|
||||
except urllib.error.HTTPError as e:
|
||||
return e.code, json.loads(e.read())
|
||||
|
||||
|
||||
class TestSkillsEndpointE2E(unittest.TestCase):
|
||||
"""E2E tests: start a real server and hit /api/skills over HTTP."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
"""Start a test server on a random port with mock skill data."""
|
||||
cls.port = _find_free_port()
|
||||
cls.base_url = f"http://127.0.0.1:{cls.port}"
|
||||
|
||||
# Create temp directories for skill data
|
||||
cls.tmpdir = tempfile.mkdtemp()
|
||||
cls.home = Path(cls.tmpdir)
|
||||
|
||||
# Claude skills
|
||||
for name, desc in [
|
||||
("commit", "Create a git commit"),
|
||||
("review-pr", "Review a pull request"),
|
||||
("comment", "Add a comment"),
|
||||
]:
|
||||
skill_dir = cls.home / ".claude/skills" / name
|
||||
skill_dir.mkdir(parents=True, exist_ok=True)
|
||||
(skill_dir / "SKILL.md").write_text(desc)
|
||||
|
||||
# Codex curated skills
|
||||
cache_dir = cls.home / ".codex/vendor_imports"
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
cache = {
|
||||
"skills": [
|
||||
{"id": "lint", "shortDescription": "Lint code"},
|
||||
{"id": "deploy", "shortDescription": "Deploy to prod"},
|
||||
]
|
||||
}
|
||||
(cache_dir / "skills-curated-cache.json").write_text(json.dumps(cache))
|
||||
|
||||
# Codex user skill
|
||||
codex_skill = cls.home / ".codex/skills/my-script"
|
||||
codex_skill.mkdir(parents=True, exist_ok=True)
|
||||
(codex_skill / "SKILL.md").write_text("Run my custom script")
|
||||
|
||||
# Patch Path.home() for the skills enumeration
|
||||
cls.home_patcher = patch.object(Path, "home", return_value=cls.home)
|
||||
cls.home_patcher.start()
|
||||
|
||||
# Start server in background thread
|
||||
cls.server = ThreadingHTTPServer(("127.0.0.1", cls.port), AMCHandler)
|
||||
cls.server_thread = threading.Thread(target=cls.server.serve_forever)
|
||||
cls.server_thread.daemon = True
|
||||
cls.server_thread.start()
|
||||
|
||||
# Wait for server to be ready
|
||||
for _ in range(50):
|
||||
try:
|
||||
with socket.create_connection(("127.0.0.1", cls.port), timeout=0.1):
|
||||
break
|
||||
except OSError:
|
||||
time.sleep(0.05)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
"""Shut down the test server."""
|
||||
cls.server.shutdown()
|
||||
cls.server_thread.join(timeout=5)
|
||||
cls.home_patcher.stop()
|
||||
|
||||
# -- Core: /api/skills serves correctly --
|
||||
|
||||
def test_skills_default_is_claude(self):
|
||||
"""GET /api/skills without ?agent defaults to claude (/ trigger)."""
|
||||
status, data = _get_json(f"{self.base_url}/api/skills")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["trigger"], "/")
|
||||
self.assertIsInstance(data["skills"], list)
|
||||
|
||||
def test_claude_skills_returned(self):
|
||||
"""GET /api/skills?agent=claude returns Claude skills."""
|
||||
status, data = _get_json(f"{self.base_url}/api/skills?agent=claude")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["trigger"], "/")
|
||||
names = [s["name"] for s in data["skills"]]
|
||||
self.assertIn("commit", names)
|
||||
self.assertIn("review-pr", names)
|
||||
self.assertIn("comment", names)
|
||||
|
||||
def test_codex_skills_returned(self):
|
||||
"""GET /api/skills?agent=codex returns Codex skills with $ trigger."""
|
||||
status, data = _get_json(f"{self.base_url}/api/skills?agent=codex")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["trigger"], "$")
|
||||
names = [s["name"] for s in data["skills"]]
|
||||
self.assertIn("lint", names)
|
||||
self.assertIn("deploy", names)
|
||||
self.assertIn("my-script", names)
|
||||
|
||||
def test_unknown_agent_defaults_to_claude(self):
|
||||
"""Unknown agent type defaults to claude behavior."""
|
||||
status, data = _get_json(f"{self.base_url}/api/skills?agent=unknown-agent")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["trigger"], "/")
|
||||
|
||||
# -- Response format --
|
||||
|
||||
def test_response_has_trigger_and_skills_keys(self):
|
||||
"""Response JSON has exactly trigger and skills keys."""
|
||||
_, data = _get_json(f"{self.base_url}/api/skills?agent=claude")
|
||||
self.assertIn("trigger", data)
|
||||
self.assertIn("skills", data)
|
||||
|
||||
def test_each_skill_has_name_and_description(self):
|
||||
"""Each skill object has name and description fields."""
|
||||
_, data = _get_json(f"{self.base_url}/api/skills?agent=claude")
|
||||
for skill in data["skills"]:
|
||||
self.assertIn("name", skill)
|
||||
self.assertIn("description", skill)
|
||||
self.assertIsInstance(skill["name"], str)
|
||||
self.assertIsInstance(skill["description"], str)
|
||||
|
||||
# -- Alphabetical sorting --
|
||||
|
||||
def test_claude_skills_alphabetically_sorted(self):
|
||||
"""Claude skills are returned in alphabetical order."""
|
||||
_, data = _get_json(f"{self.base_url}/api/skills?agent=claude")
|
||||
names = [s["name"] for s in data["skills"]]
|
||||
self.assertEqual(names, sorted(names, key=str.lower))
|
||||
|
||||
def test_codex_skills_alphabetically_sorted(self):
|
||||
"""Codex skills are returned in alphabetical order."""
|
||||
_, data = _get_json(f"{self.base_url}/api/skills?agent=codex")
|
||||
names = [s["name"] for s in data["skills"]]
|
||||
self.assertEqual(names, sorted(names, key=str.lower))
|
||||
|
||||
# -- Descriptions --
|
||||
|
||||
def test_claude_skill_descriptions(self):
|
||||
"""Claude skills have correct descriptions from SKILL.md."""
|
||||
_, data = _get_json(f"{self.base_url}/api/skills?agent=claude")
|
||||
by_name = {s["name"]: s["description"] for s in data["skills"]}
|
||||
self.assertEqual(by_name["commit"], "Create a git commit")
|
||||
self.assertEqual(by_name["review-pr"], "Review a pull request")
|
||||
|
||||
def test_codex_curated_descriptions(self):
|
||||
"""Codex curated skills have correct descriptions from cache."""
|
||||
_, data = _get_json(f"{self.base_url}/api/skills?agent=codex")
|
||||
by_name = {s["name"]: s["description"] for s in data["skills"]}
|
||||
self.assertEqual(by_name["lint"], "Lint code")
|
||||
self.assertEqual(by_name["deploy"], "Deploy to prod")
|
||||
|
||||
def test_codex_user_skill_description(self):
|
||||
"""Codex user-installed skills have descriptions from SKILL.md."""
|
||||
_, data = _get_json(f"{self.base_url}/api/skills?agent=codex")
|
||||
by_name = {s["name"]: s["description"] for s in data["skills"]}
|
||||
self.assertEqual(by_name["my-script"], "Run my custom script")
|
||||
|
||||
# -- CORS --
|
||||
|
||||
def test_cors_header_present(self):
|
||||
"""Response includes Access-Control-Allow-Origin header."""
|
||||
url = f"{self.base_url}/api/skills?agent=claude"
|
||||
with urllib.request.urlopen(url, timeout=5) as resp:
|
||||
cors = resp.headers.get("Access-Control-Allow-Origin")
|
||||
self.assertEqual(cors, "*")
|
||||
|
||||
|
||||
class TestSkillsEndpointEmptyE2E(unittest.TestCase):
|
||||
"""E2E tests: server with no skills data."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.port = _find_free_port()
|
||||
cls.base_url = f"http://127.0.0.1:{cls.port}"
|
||||
|
||||
# Empty home directory - no skills at all
|
||||
cls.tmpdir = tempfile.mkdtemp()
|
||||
cls.home = Path(cls.tmpdir)
|
||||
|
||||
cls.home_patcher = patch.object(Path, "home", return_value=cls.home)
|
||||
cls.home_patcher.start()
|
||||
|
||||
cls.server = ThreadingHTTPServer(("127.0.0.1", cls.port), AMCHandler)
|
||||
cls.server_thread = threading.Thread(target=cls.server.serve_forever)
|
||||
cls.server_thread.daemon = True
|
||||
cls.server_thread.start()
|
||||
|
||||
for _ in range(50):
|
||||
try:
|
||||
with socket.create_connection(("127.0.0.1", cls.port), timeout=0.1):
|
||||
break
|
||||
except OSError:
|
||||
time.sleep(0.05)
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.server.shutdown()
|
||||
cls.server_thread.join(timeout=5)
|
||||
cls.home_patcher.stop()
|
||||
|
||||
def test_empty_claude_skills(self):
|
||||
"""Server with no Claude skills returns empty list."""
|
||||
status, data = _get_json(f"{self.base_url}/api/skills?agent=claude")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["trigger"], "/")
|
||||
self.assertEqual(data["skills"], [])
|
||||
|
||||
def test_empty_codex_skills(self):
|
||||
"""Server with no Codex skills returns empty list."""
|
||||
status, data = _get_json(f"{self.base_url}/api/skills?agent=codex")
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(data["trigger"], "$")
|
||||
self.assertEqual(data["skills"], [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user