e3e42e53f2
- Reject null bytes and control characters (U+0000-U+001F, U+007F) in _validate_spawn_params with explicit INVALID_PROJECT error - Reject whitespace-only project names as MISSING_PROJECT - Reject non-string project names (int, list, etc.) - Add _sanitize_pane_name() to clean Zellij pane names: replaces quotes, backticks, and control chars with underscores; collapses whitespace; truncates to 64 chars - Add 44 new tests: safe chars (hyphens, spaces, dots, @, +, #), dangerous chars (null byte, newline, tab, ESC, DEL), shell metacharacters ($, ;, backtick, |), pane name sanitization, and spawn command construction with special char names Closes bd-14p
892 lines
35 KiB
Python
892 lines
35 KiB
Python
import io
|
|
import json
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import amc_server.mixins.spawn as spawn_mod
|
|
from amc_server.mixins.spawn import SpawnMixin, load_projects_cache, _projects_cache, _sanitize_pane_name
|
|
|
|
|
|
class DummySpawnHandler(SpawnMixin):
|
|
"""Minimal handler stub matching the project convention (see test_control.py)."""
|
|
|
|
def __init__(self, body=None, auth_header=''):
|
|
if body is None:
|
|
body = {}
|
|
raw = json.dumps(body).encode('utf-8')
|
|
self.headers = {
|
|
'Content-Length': str(len(raw)),
|
|
'Authorization': auth_header,
|
|
}
|
|
self.rfile = io.BytesIO(raw)
|
|
self.sent = []
|
|
self.errors = []
|
|
|
|
def _send_json(self, code, payload):
|
|
self.sent.append((code, payload))
|
|
|
|
def _json_error(self, code, message):
|
|
self.errors.append((code, message))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _validate_spawn_params
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestValidateSpawnParams(unittest.TestCase):
|
|
"""Tests for _validate_spawn_params security validation."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummySpawnHandler()
|
|
|
|
# --- missing / empty project ---
|
|
|
|
def test_empty_project_returns_error(self):
|
|
result = self.handler._validate_spawn_params('', 'claude')
|
|
self.assertEqual(result['code'], 'MISSING_PROJECT')
|
|
|
|
def test_none_project_returns_error(self):
|
|
result = self.handler._validate_spawn_params(None, 'claude')
|
|
self.assertEqual(result['code'], 'MISSING_PROJECT')
|
|
|
|
# --- path traversal: slash ---
|
|
|
|
def test_forward_slash_rejected(self):
|
|
result = self.handler._validate_spawn_params('../etc', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
def test_nested_slash_rejected(self):
|
|
result = self.handler._validate_spawn_params('foo/bar', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
# --- path traversal: dotdot ---
|
|
|
|
def test_dotdot_in_name_rejected(self):
|
|
result = self.handler._validate_spawn_params('foo..bar', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
def test_leading_dotdot_rejected(self):
|
|
result = self.handler._validate_spawn_params('..project', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
# --- backslash ---
|
|
|
|
def test_backslash_rejected(self):
|
|
result = self.handler._validate_spawn_params('foo\\bar', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
# --- symlink escape ---
|
|
|
|
def test_symlink_escape_rejected(self):
|
|
"""Project resolving outside PROJECTS_DIR is rejected."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
|
|
# Create a symlink that escapes projects dir
|
|
escape_target = Path(tmpdir) / 'outside'
|
|
escape_target.mkdir()
|
|
symlink = projects / 'evil'
|
|
symlink.symlink_to(escape_target)
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('evil', 'claude')
|
|
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
# --- nonexistent project ---
|
|
|
|
def test_nonexistent_project_rejected(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('nonexistent', 'claude')
|
|
|
|
self.assertEqual(result['code'], 'PROJECT_NOT_FOUND')
|
|
|
|
def test_file_instead_of_directory_rejected(self):
|
|
"""A regular file (not a directory) is not a valid project."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
(projects / 'notadir').write_text('oops')
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('notadir', 'claude')
|
|
|
|
self.assertEqual(result['code'], 'PROJECT_NOT_FOUND')
|
|
|
|
# --- invalid agent type ---
|
|
|
|
def test_invalid_agent_type_rejected(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
(projects / 'myproject').mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('myproject', 'gpt5')
|
|
|
|
self.assertEqual(result['code'], 'INVALID_AGENT_TYPE')
|
|
self.assertIn('gpt5', result['error'])
|
|
|
|
def test_empty_agent_type_rejected(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
(projects / 'myproject').mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('myproject', '')
|
|
|
|
self.assertEqual(result['code'], 'INVALID_AGENT_TYPE')
|
|
|
|
# --- valid project returns resolved_path ---
|
|
|
|
def test_valid_project_returns_resolved_path(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
project_dir = projects / 'amc'
|
|
project_dir.mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('amc', 'claude')
|
|
|
|
self.assertNotIn('error', result)
|
|
self.assertEqual(result['resolved_path'], project_dir.resolve())
|
|
|
|
def test_valid_project_with_codex_agent(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
(projects / 'myproject').mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('myproject', 'codex')
|
|
|
|
self.assertNotIn('error', result)
|
|
self.assertIn('resolved_path', result)
|
|
|
|
# --- unicode / special characters ---
|
|
|
|
def test_unicode_project_name_without_traversal_chars(self):
|
|
"""Project names with unicode but no traversal chars resolve normally."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
# Create a project with a non-ASCII name
|
|
project_dir = projects / 'cafe'
|
|
project_dir.mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('cafe', 'claude')
|
|
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_whitespace_only_project_name(self):
|
|
"""Whitespace-only project name should fail (falsy)."""
|
|
result = self.handler._validate_spawn_params('', 'claude')
|
|
self.assertEqual(result['code'], 'MISSING_PROJECT')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# load_projects_cache
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestProjectsCache(unittest.TestCase):
|
|
"""Tests for projects cache loading."""
|
|
|
|
def test_loads_directory_names_sorted(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir)
|
|
(projects / 'zebra').mkdir()
|
|
(projects / 'alpha').mkdir()
|
|
(projects / 'middle').mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
load_projects_cache()
|
|
|
|
self.assertEqual(spawn_mod._projects_cache, ['alpha', 'middle', 'zebra'])
|
|
|
|
def test_excludes_hidden_directories(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir)
|
|
(projects / 'visible').mkdir()
|
|
(projects / '.hidden').mkdir()
|
|
(projects / '.git').mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
load_projects_cache()
|
|
|
|
self.assertEqual(spawn_mod._projects_cache, ['visible'])
|
|
|
|
def test_excludes_files(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir)
|
|
(projects / 'real-project').mkdir()
|
|
(projects / 'README.md').write_text('hello')
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
load_projects_cache()
|
|
|
|
self.assertEqual(spawn_mod._projects_cache, ['real-project'])
|
|
|
|
def test_handles_oserror_gracefully(self):
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR') as mock_dir:
|
|
mock_dir.iterdir.side_effect = OSError('permission denied')
|
|
load_projects_cache()
|
|
|
|
self.assertEqual(spawn_mod._projects_cache, [])
|
|
|
|
def test_empty_directory(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir)
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
load_projects_cache()
|
|
|
|
self.assertEqual(spawn_mod._projects_cache, [])
|
|
|
|
def test_missing_directory(self):
|
|
"""PROJECTS_DIR that doesn't exist returns empty list, no crash."""
|
|
missing = Path('/tmp/amc-test-nonexistent-projects-dir')
|
|
assert not missing.exists(), 'test precondition: path must not exist'
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', missing):
|
|
load_projects_cache()
|
|
|
|
self.assertEqual(spawn_mod._projects_cache, [])
|
|
|
|
def test_only_hidden_dirs_returns_empty(self):
|
|
"""Directory with only hidden subdirectories returns empty list."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir)
|
|
(projects / '.config').mkdir()
|
|
(projects / '.cache').mkdir()
|
|
(projects / '.local').mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
load_projects_cache()
|
|
|
|
self.assertEqual(spawn_mod._projects_cache, [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rate limiting
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRateLimiting(unittest.TestCase):
|
|
"""Tests for per-project rate limiting in _handle_spawn."""
|
|
|
|
def _make_handler(self, project, agent_type='claude', token='test-token'):
|
|
body = {'project': project, 'agent_type': agent_type}
|
|
return DummySpawnHandler(body, auth_header=f'Bearer {token}')
|
|
|
|
def test_first_spawn_allowed(self):
|
|
"""First spawn for a project should not be rate-limited."""
|
|
from amc_server.context import _spawn_timestamps
|
|
_spawn_timestamps.clear()
|
|
|
|
handler = self._make_handler('fresh-project')
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir()
|
|
(projects / 'fresh-project').mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects), \
|
|
patch.object(spawn_mod, 'validate_auth_token', return_value=True), \
|
|
patch.object(handler, '_spawn_agent_in_project_tab', return_value={'ok': True}), \
|
|
patch.object(handler, '_validate_spawn_params', return_value={
|
|
'resolved_path': projects / 'fresh-project',
|
|
}):
|
|
handler._handle_spawn()
|
|
|
|
self.assertEqual(len(handler.sent), 1)
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(payload['ok'])
|
|
|
|
_spawn_timestamps.clear()
|
|
|
|
def test_rapid_spawn_same_project_rejected(self):
|
|
"""Spawning the same project within cooldown returns 429."""
|
|
from amc_server.context import _spawn_timestamps, SPAWN_COOLDOWN_SEC
|
|
_spawn_timestamps.clear()
|
|
# Pretend we just spawned this project
|
|
_spawn_timestamps['rapid-project'] = time.monotonic()
|
|
|
|
handler = self._make_handler('rapid-project')
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=True), \
|
|
patch.object(handler, '_validate_spawn_params', return_value={
|
|
'resolved_path': Path('/fake/rapid-project'),
|
|
}):
|
|
handler._handle_spawn()
|
|
|
|
self.assertEqual(len(handler.sent), 1)
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 429)
|
|
self.assertEqual(payload['code'], 'RATE_LIMITED')
|
|
|
|
_spawn_timestamps.clear()
|
|
|
|
def test_spawn_different_project_allowed(self):
|
|
"""Spawning a different project while one is on cooldown succeeds."""
|
|
from amc_server.context import _spawn_timestamps
|
|
_spawn_timestamps.clear()
|
|
_spawn_timestamps['project-a'] = time.monotonic()
|
|
|
|
handler = self._make_handler('project-b')
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=True), \
|
|
patch.object(handler, '_validate_spawn_params', return_value={
|
|
'resolved_path': Path('/fake/project-b'),
|
|
}), \
|
|
patch.object(handler, '_spawn_agent_in_project_tab', return_value={'ok': True}):
|
|
handler._handle_spawn()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(payload['ok'])
|
|
|
|
_spawn_timestamps.clear()
|
|
|
|
def test_spawn_after_cooldown_allowed(self):
|
|
"""Spawning the same project after cooldown expires succeeds."""
|
|
from amc_server.context import _spawn_timestamps, SPAWN_COOLDOWN_SEC
|
|
_spawn_timestamps.clear()
|
|
# Set timestamp far enough in the past
|
|
_spawn_timestamps['cooled-project'] = time.monotonic() - SPAWN_COOLDOWN_SEC - 1
|
|
|
|
handler = self._make_handler('cooled-project')
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=True), \
|
|
patch.object(handler, '_validate_spawn_params', return_value={
|
|
'resolved_path': Path('/fake/cooled-project'),
|
|
}), \
|
|
patch.object(handler, '_spawn_agent_in_project_tab', return_value={'ok': True}):
|
|
handler._handle_spawn()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(payload['ok'])
|
|
|
|
_spawn_timestamps.clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth token validation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestAuthToken(unittest.TestCase):
|
|
"""Tests for auth token validation in _handle_spawn."""
|
|
|
|
def test_valid_bearer_token_accepted(self):
|
|
handler = DummySpawnHandler(
|
|
{'project': 'test', 'agent_type': 'claude'},
|
|
auth_header='Bearer valid-token',
|
|
)
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=True), \
|
|
patch.object(handler, '_validate_spawn_params', return_value={
|
|
'resolved_path': Path('/fake/test'),
|
|
}), \
|
|
patch.object(handler, '_spawn_agent_in_project_tab', return_value={'ok': True}):
|
|
handler._handle_spawn()
|
|
|
|
code, _ = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
|
|
def test_missing_auth_header_rejected(self):
|
|
handler = DummySpawnHandler(
|
|
{'project': 'test', 'agent_type': 'claude'},
|
|
auth_header='',
|
|
)
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=False):
|
|
handler._handle_spawn()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 401)
|
|
self.assertEqual(payload['code'], 'UNAUTHORIZED')
|
|
|
|
def test_wrong_token_rejected(self):
|
|
handler = DummySpawnHandler(
|
|
{'project': 'test', 'agent_type': 'claude'},
|
|
auth_header='Bearer wrong-token',
|
|
)
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=False):
|
|
handler._handle_spawn()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 401)
|
|
self.assertEqual(payload['code'], 'UNAUTHORIZED')
|
|
|
|
def test_malformed_bearer_rejected(self):
|
|
handler = DummySpawnHandler(
|
|
{'project': 'test', 'agent_type': 'claude'},
|
|
auth_header='NotBearer sometoken',
|
|
)
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=False):
|
|
handler._handle_spawn()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 401)
|
|
self.assertEqual(payload['code'], 'UNAUTHORIZED')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _handle_spawn JSON parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHandleSpawnParsing(unittest.TestCase):
|
|
"""Tests for JSON body parsing in _handle_spawn."""
|
|
|
|
def test_invalid_json_body_returns_400(self):
|
|
handler = DummySpawnHandler.__new__(DummySpawnHandler)
|
|
handler.headers = {
|
|
'Content-Length': '11',
|
|
'Authorization': 'Bearer tok',
|
|
}
|
|
handler.rfile = io.BytesIO(b'not json!!!')
|
|
handler.sent = []
|
|
handler.errors = []
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=True):
|
|
handler._handle_spawn()
|
|
|
|
self.assertEqual(handler.errors, [(400, 'Invalid JSON body')])
|
|
|
|
def test_non_dict_body_returns_400(self):
|
|
raw = b'"just a string"'
|
|
handler = DummySpawnHandler.__new__(DummySpawnHandler)
|
|
handler.headers = {
|
|
'Content-Length': str(len(raw)),
|
|
'Authorization': 'Bearer tok',
|
|
}
|
|
handler.rfile = io.BytesIO(raw)
|
|
handler.sent = []
|
|
handler.errors = []
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=True):
|
|
handler._handle_spawn()
|
|
|
|
self.assertEqual(handler.errors, [(400, 'Invalid JSON body')])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _handle_spawn lock contention
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHandleSpawnLock(unittest.TestCase):
|
|
"""Tests for spawn lock behavior."""
|
|
|
|
def test_lock_timeout_returns_503(self):
|
|
"""When lock can't be acquired within timeout, returns 503."""
|
|
handler = DummySpawnHandler(
|
|
{'project': 'test', 'agent_type': 'claude'},
|
|
auth_header='Bearer tok',
|
|
)
|
|
|
|
mock_lock = MagicMock()
|
|
mock_lock.acquire.return_value = False # Simulate timeout
|
|
|
|
with patch.object(spawn_mod, 'validate_auth_token', return_value=True), \
|
|
patch.object(handler, '_validate_spawn_params', return_value={
|
|
'resolved_path': Path('/fake/test'),
|
|
}), \
|
|
patch.object(spawn_mod, '_spawn_lock', mock_lock):
|
|
handler._handle_spawn()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 503)
|
|
self.assertEqual(payload['code'], 'SERVER_BUSY')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _handle_projects / _handle_projects_refresh
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHandleProjects(unittest.TestCase):
|
|
|
|
def test_handle_projects_returns_cached_list(self):
|
|
handler = DummySpawnHandler()
|
|
original = spawn_mod._projects_cache
|
|
|
|
spawn_mod._projects_cache = ['alpha', 'beta']
|
|
try:
|
|
handler._handle_projects()
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(payload['ok'])
|
|
self.assertEqual(payload['projects'], ['alpha', 'beta'])
|
|
finally:
|
|
spawn_mod._projects_cache = original
|
|
|
|
def test_handle_projects_returns_empty_list_gracefully(self):
|
|
handler = DummySpawnHandler()
|
|
original = spawn_mod._projects_cache
|
|
|
|
spawn_mod._projects_cache = []
|
|
try:
|
|
handler._handle_projects()
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(payload['ok'])
|
|
self.assertEqual(payload['projects'], [])
|
|
finally:
|
|
spawn_mod._projects_cache = original
|
|
|
|
def test_handle_projects_refresh_missing_dir_returns_empty(self):
|
|
"""Refreshing with a missing PROJECTS_DIR returns empty list, no crash."""
|
|
handler = DummySpawnHandler()
|
|
missing = Path('/tmp/amc-test-nonexistent-projects-dir')
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', missing):
|
|
handler._handle_projects_refresh()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(payload['ok'])
|
|
self.assertEqual(payload['projects'], [])
|
|
|
|
def test_handle_projects_refresh_reloads_cache(self):
|
|
handler = DummySpawnHandler()
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects = Path(tmpdir)
|
|
(projects / 'new-project').mkdir()
|
|
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
handler._handle_projects_refresh()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual(payload['projects'], ['new-project'])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _handle_health
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHandleHealth(unittest.TestCase):
|
|
|
|
def test_health_with_zellij_available(self):
|
|
handler = DummySpawnHandler()
|
|
handler._check_zellij_session_exists = MagicMock(return_value=True)
|
|
|
|
handler._handle_health()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(payload['ok'])
|
|
self.assertTrue(payload['zellij_available'])
|
|
|
|
def test_health_with_zellij_unavailable(self):
|
|
handler = DummySpawnHandler()
|
|
handler._check_zellij_session_exists = MagicMock(return_value=False)
|
|
|
|
handler._handle_health()
|
|
|
|
code, payload = handler.sent[0]
|
|
self.assertEqual(code, 200)
|
|
self.assertFalse(payload['zellij_available'])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Special characters in project names (bd-14p)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSpecialCharacterValidation(unittest.TestCase):
|
|
"""Verify project names with special characters are handled correctly."""
|
|
|
|
def setUp(self):
|
|
self.handler = DummySpawnHandler()
|
|
|
|
def _make_project(self, tmpdir, name):
|
|
"""Create a project directory with the given name and patch PROJECTS_DIR."""
|
|
projects = Path(tmpdir) / 'projects'
|
|
projects.mkdir(exist_ok=True)
|
|
project_dir = projects / name
|
|
project_dir.mkdir()
|
|
return projects, project_dir
|
|
|
|
# --- safe characters that should work ---
|
|
|
|
def test_hyphenated_name(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'my-project')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('my-project', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_underscored_name(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'my_project')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('my_project', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_numeric_name(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'project2')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('project2', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_name_with_spaces(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'my project')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('my project', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_name_with_dots(self):
|
|
"""Single dots are fine, only '..' is rejected."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'my.project')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('my.project', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_name_with_parentheses(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'project (copy)')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('project (copy)', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_name_with_at_sign(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, '@scope-pkg')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('@scope-pkg', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_name_with_plus(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'c++project')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('c++project', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_name_with_hash(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'c#project')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('c#project', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
# --- control characters: should be rejected ---
|
|
|
|
def test_null_byte_rejected(self):
|
|
result = self.handler._validate_spawn_params('project\x00evil', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
def test_newline_rejected(self):
|
|
result = self.handler._validate_spawn_params('project\nevil', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
def test_tab_rejected(self):
|
|
result = self.handler._validate_spawn_params('project\tevil', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
def test_carriage_return_rejected(self):
|
|
result = self.handler._validate_spawn_params('project\revil', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
def test_escape_char_rejected(self):
|
|
result = self.handler._validate_spawn_params('project\x1bevil', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
def test_bell_char_rejected(self):
|
|
result = self.handler._validate_spawn_params('project\x07evil', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
def test_del_char_rejected(self):
|
|
result = self.handler._validate_spawn_params('project\x7fevil', 'claude')
|
|
self.assertEqual(result['code'], 'INVALID_PROJECT')
|
|
|
|
# --- whitespace edge cases ---
|
|
|
|
def test_whitespace_only_space(self):
|
|
result = self.handler._validate_spawn_params(' ', 'claude')
|
|
self.assertEqual(result['code'], 'MISSING_PROJECT')
|
|
|
|
def test_whitespace_only_multiple_spaces(self):
|
|
result = self.handler._validate_spawn_params(' ', 'claude')
|
|
self.assertEqual(result['code'], 'MISSING_PROJECT')
|
|
|
|
# --- non-string project name ---
|
|
|
|
def test_integer_project_name_rejected(self):
|
|
result = self.handler._validate_spawn_params(42, 'claude')
|
|
self.assertEqual(result['code'], 'MISSING_PROJECT')
|
|
|
|
def test_list_project_name_rejected(self):
|
|
result = self.handler._validate_spawn_params(['bad'], 'claude')
|
|
self.assertEqual(result['code'], 'MISSING_PROJECT')
|
|
|
|
# --- shell metacharacters (safe because subprocess uses list args) ---
|
|
|
|
def test_dollar_sign_in_name(self):
|
|
"""Dollar sign is safe - no shell expansion with list args."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, '$HOME')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('$HOME', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_semicolon_in_name(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'foo;bar')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('foo;bar', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_backtick_in_name(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'foo`bar')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('foo`bar', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
def test_pipe_in_name(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
projects, _ = self._make_project(tmpdir, 'foo|bar')
|
|
with patch.object(spawn_mod, 'PROJECTS_DIR', projects):
|
|
result = self.handler._validate_spawn_params('foo|bar', 'claude')
|
|
self.assertNotIn('error', result)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _sanitize_pane_name (bd-14p)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSanitizePaneName(unittest.TestCase):
|
|
"""Tests for Zellij pane name sanitization."""
|
|
|
|
def test_simple_name_unchanged(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-myproject'), 'claude-myproject')
|
|
|
|
def test_name_with_spaces_preserved(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-my project'), 'claude-my project')
|
|
|
|
def test_multiple_spaces_collapsed(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-my project'), 'claude-my project')
|
|
|
|
def test_quotes_replaced(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-"quoted"'), 'claude-_quoted_')
|
|
|
|
def test_single_quotes_replaced(self):
|
|
self.assertEqual(_sanitize_pane_name("claude-it's"), 'claude-it_s')
|
|
|
|
def test_backtick_replaced(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-`cmd`'), 'claude-_cmd_')
|
|
|
|
def test_control_chars_replaced(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-\x07bell'), 'claude-_bell')
|
|
|
|
def test_tab_replaced(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-\ttab'), 'claude-_tab')
|
|
|
|
def test_truncated_at_64_chars(self):
|
|
long_name = 'a' * 100
|
|
result = _sanitize_pane_name(long_name)
|
|
self.assertEqual(len(result), 64)
|
|
|
|
def test_empty_returns_unnamed(self):
|
|
self.assertEqual(_sanitize_pane_name(''), 'unnamed')
|
|
|
|
def test_only_control_chars_returns_underscores(self):
|
|
result = _sanitize_pane_name('\x01\x02\x03')
|
|
self.assertEqual(result, '___')
|
|
|
|
def test_unicode_preserved(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-cafe'), 'claude-cafe')
|
|
|
|
def test_hyphens_and_underscores_preserved(self):
|
|
self.assertEqual(_sanitize_pane_name('claude-my_project-v2'), 'claude-my_project-v2')
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Special chars in spawn command construction (bd-14p)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSpawnWithSpecialChars(unittest.TestCase):
|
|
"""Verify special character project names pass through to Zellij correctly."""
|
|
|
|
def test_space_in_project_name_passes_to_zellij(self):
|
|
"""Project with spaces should be passed correctly to subprocess args."""
|
|
handler = DummySpawnHandler()
|
|
|
|
with patch.object(handler, '_check_zellij_session_exists', return_value=True), \
|
|
patch.object(handler, '_wait_for_session_file', return_value=True), \
|
|
patch('amc_server.mixins.spawn.subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0, stderr='')
|
|
result = handler._spawn_agent_in_project_tab(
|
|
'my project', Path('/fake/my project'), 'claude', 'spawn-123',
|
|
)
|
|
|
|
self.assertTrue(result['ok'])
|
|
# Check the tab creation call contains the name with spaces
|
|
tab_call = mock_run.call_args_list[0]
|
|
tab_args = tab_call[0][0]
|
|
self.assertIn('my project', tab_args)
|
|
|
|
# Check pane name was sanitized (spaces preserved but other chars cleaned)
|
|
pane_call = mock_run.call_args_list[1]
|
|
pane_args = pane_call[0][0]
|
|
name_idx = pane_args.index('--name') + 1
|
|
self.assertEqual(pane_args[name_idx], 'claude-my project')
|
|
|
|
def test_quotes_in_project_name_sanitized_in_pane_name(self):
|
|
"""Quotes in project names should be stripped from pane names."""
|
|
handler = DummySpawnHandler()
|
|
|
|
with patch.object(handler, '_check_zellij_session_exists', return_value=True), \
|
|
patch.object(handler, '_wait_for_session_file', return_value=True), \
|
|
patch('amc_server.mixins.spawn.subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0, stderr='')
|
|
handler._spawn_agent_in_project_tab(
|
|
'it\'s-a-project', Path('/fake/its-a-project'), 'claude', 'spawn-123',
|
|
)
|
|
|
|
pane_call = mock_run.call_args_list[1]
|
|
pane_args = pane_call[0][0]
|
|
name_idx = pane_args.index('--name') + 1
|
|
# Single quote should be replaced with underscore
|
|
self.assertNotIn("'", pane_args[name_idx])
|
|
self.assertEqual(pane_args[name_idx], 'claude-it_s-a-project')
|
|
|
|
def test_unicode_project_name_in_pane(self):
|
|
"""Unicode project names should pass through to pane names."""
|
|
handler = DummySpawnHandler()
|
|
|
|
with patch.object(handler, '_check_zellij_session_exists', return_value=True), \
|
|
patch.object(handler, '_wait_for_session_file', return_value=True), \
|
|
patch('amc_server.mixins.spawn.subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0, stderr='')
|
|
result = handler._spawn_agent_in_project_tab(
|
|
'cafe', Path('/fake/cafe'), 'claude', 'spawn-123',
|
|
)
|
|
|
|
self.assertTrue(result['ok'])
|
|
pane_call = mock_run.call_args_list[1]
|
|
pane_args = pane_call[0][0]
|
|
name_idx = pane_args.index('--name') + 1
|
|
self.assertEqual(pane_args[name_idx], 'claude-cafe')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|