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, _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_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()