251 lines
9.3 KiB
Python
251 lines
9.3 KiB
Python
"""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()
|