import contextlib import csv import importlib.util import io import json import os from pathlib import Path import shutil import subprocess import tempfile import unittest from unittest.mock import patch root = Path(__file__).parents[2] spec = importlib.util.spec_from_file_location('extensions', root / 'scripts/lib/managed-extensions.py') extensions = importlib.util.module_from_spec(spec) spec.loader.exec_module(extensions) def plugin(name='Example'): return dict(name=name, codex_method='plugin', codex_plugin=name.lower() + '@test', codex_marketplace='-', claude_plugin=name.lower() + '@test', claude_marketplace='-') def skill(): return dict(name='Humanizer', codex_method='skill', codex_source='blader/humanizer', codex_skill='humanizer', codex_plugin='-') def mcporter(): return dict(name='MCPorter', codex_method='cli', codex_source='mcporter', codex_plugin='-', codex_marketplace='-', claude_plugin='-', claude_marketplace='-') def mcp(): return dict(name='Context7', codex_method='plugin', codex_plugin='context7@test', codex_marketplace='-', claude_plugin='context7@test', claude_marketplace='-', mcporter_name='context7', mcporter_url='https://mcp.context7.com/mcp') class ManagedExtensionTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.addCleanup(self.temporary.cleanup) self.home = Path(self.temporary.name) self.installed = {'codex': {}, 'claude': {}} self.commands = [] self.addCleanup(patch.stopall) patch.object(extensions, 'installed_plugins', side_effect=lambda client, home: dict(self.installed[client])).start() patch.object(extensions, 'run_command', side_effect=self.command).start() patch.object(extensions, 'fetch_skill', return_value=('humanizer', {'SKILL.md': (b'original', False), 'references/old.md': (b'old', False)})).start() self.output = contextlib.redirect_stdout(io.StringIO()) self.output.__enter__() self.addCleanup(self.output.__exit__, None, None, None) def command(self, args, summary, home=None): if args[:2] == ['npm', 'list']: return '{"dependencies":{}}' self.commands.append(args) if len(args) > 3 and args[2] in ('add', 'install') and args[1] == 'plugin': self.installed[args[0]][args[3]] = {'scope': 'user'} if args[2] in ('remove', 'uninstall'): self.installed[args[0]].pop(args[3], None) return '' def manager(self, **kwargs): return extensions.Manager(self.home, **kwargs) def test_foreign_plugin_is_never_claimed_or_removed(self): self.installed['codex']['example@test'] = {} self.manager(update=True).sync([plugin()], ['codex'], {'Example'}) self.manager().sync([], ['codex'], set()) self.assertEqual([], self.commands) self.assertEqual([], self.manager().state['resources']) def test_owned_plugin_removed_after_manifest_entry_disappears(self): self.manager().sync([plugin()], ['codex'], {'Example'}) self.manager().sync([], ['codex'], set()) self.assertEqual(['codex', 'plugin', 'remove', 'example@test'], self.commands[-1]) self.assertEqual([], self.manager().state['resources']) def test_client_scope_is_preserved(self): self.manager().sync([plugin()], ['claude', 'codex'], {'Example'}) self.manager().sync([], ['codex'], set()) self.assertEqual(['claude'], [r['client'] for r in self.manager().state['resources']]) def test_partial_plugin_failure_retains_successful_ownership(self): with patch.object(extensions, 'run_command', side_effect=[None, ValueError('failure')]): with self.assertRaises(ValueError): self.manager().sync([plugin(), plugin('Second')], ['codex'], {'Example', 'Second'}) self.assertEqual(['Example'], [r['name'] for r in self.manager().state['resources']]) def test_failed_removal_retains_ownership(self): self.manager().sync([plugin()], ['codex'], {'Example'}) with patch.object(extensions, 'run_command', side_effect=ValueError('failure')): with self.assertRaises(ValueError): self.manager().sync([], ['codex'], set()) self.assertEqual(1, len(self.manager().state['resources'])) def test_foreign_skill_directory_is_preserved(self): target = self.home / '.agents/skills/humanizer/SKILL.md' target.parent.mkdir(parents=True) target.write_text('foreign') self.manager(update=True).sync([skill()], ['codex'], {'Humanizer'}) self.manager().sync([], ['codex'], set()) self.assertEqual('foreign', target.read_text()) self.assertEqual([], self.manager().state['resources']) def test_skill_removal_preserves_modified_and_foreign_files(self): self.manager().sync([skill()], ['codex'], {'Humanizer'}) directory = self.home / '.agents/skills/humanizer' (directory / 'SKILL.md').write_text('modified') (directory / 'foreign.txt').write_text('foreign') self.manager().sync([], ['codex'], set()) self.assertEqual('modified', (directory / 'SKILL.md').read_text()) self.assertTrue((directory / 'foreign.txt').exists()) self.assertFalse((directory / 'references/old.md').exists()) self.assertEqual(1, len(self.manager().state['resources'][0]['files'])) def test_skill_update_cleans_only_owned_stale_files(self): self.manager().sync([skill()], ['codex'], {'Humanizer'}) directory = self.home / '.agents/skills/humanizer' (directory / 'foreign.txt').write_text('foreign') with patch.object(extensions, 'fetch_skill', return_value=('humanizer', {'SKILL.md': (b'new', False)})): self.manager(update=True).sync([skill()], ['codex'], {'Humanizer'}) self.assertEqual('new', (directory / 'SKILL.md').read_text()) self.assertTrue((directory / 'foreign.txt').exists()) self.assertFalse((directory / 'references/old.md').exists()) def test_skill_records_resolved_revision(self): with patch.object(extensions, 'fetch_skill', return_value=('humanizer', {'SKILL.md': (b'original', False)}, 'a' * 40)): self.manager().sync([skill()], ['codex'], {'Humanizer'}) record = self.manager().state['resources'][0] self.assertEqual(record['source'], 'blader/humanizer') self.assertEqual(record['resolvedRevision'], 'a' * 40) def test_skill_partial_failure_records_each_successful_file(self): original_write = extensions.atomic_write def write(path, content): if path.name == 'old.md': raise OSError('simulated disk failure') original_write(path, content) with patch.object(extensions, 'atomic_write', side_effect=write): with self.assertRaises(OSError): self.manager().sync([skill()], ['codex'], {'Humanizer'}) self.assertEqual(['.agents/skills/humanizer/SKILL.md'], list(self.manager().state['resources'][0]['files'])) self.manager().sync([skill()], ['codex'], {'Humanizer'}) self.assertEqual(2, len(self.manager().state['resources'][0]['files'])) self.assertTrue(self.manager().state['resources'][0]['complete']) def test_dry_run_has_no_commands_network_or_writes(self): self.manager(dry_run=True).sync([plugin(), skill()], ['codex'], {'Example', 'Humanizer'}) self.assertEqual([], list(self.home.iterdir())) self.assertEqual([], self.commands) extensions.fetch_skill.assert_not_called() def test_traversal_ledger_is_rejected(self): path = self.home / '.my-ai-configuration/extensions.json' path.parent.mkdir() path.write_text(json.dumps({'version': 1, 'resources': [dict(client='codex', name='Bad', kind='skill', directory='.agents/skills/bad', files={'.agents/skills/bad/../../../other': '0' * 64})]})) with self.assertRaises(ValueError): self.manager() def test_marketplace_failure_never_removes_existing_source(self): entry = plugin() entry['codex_marketplace'] = 'owner/repository' with patch.object(extensions, 'run_command', side_effect=ValueError('different source')) as command: with self.assertRaises(ValueError): self.manager().sync([entry], ['codex'], {'Example'}) self.assertEqual(1, command.call_count) self.assertEqual('add', command.call_args.args[0][3]) def test_owned_codex_update_upgrades_marketplace(self): entry = plugin() entry['codex_marketplace'] = 'owner/repository' self.manager().sync([entry], ['codex'], {'Example'}) self.commands.clear() self.manager(update=True).sync([entry], ['codex'], {'Example'}) self.assertEqual([['codex', 'plugin', 'marketplace', 'upgrade', 'test'], ['codex', 'plugin', 'add', 'example@test']], self.commands) def test_owned_disabled_claude_is_enabled(self): self.manager().sync([plugin()], ['claude'], {'Example'}) self.installed['claude']['example@test']['enabled'] = False self.commands.clear() self.manager(update=True).sync([plugin()], ['claude'], {'Example'}) self.assertEqual([ ['claude', 'plugin', 'marketplace', 'update', 'test'], ['claude', 'plugin', 'update', 'example@test'], ['claude', 'plugin', 'enable', 'example@test'], ], self.commands) def test_superpowers_adds_official_claude_marketplace_before_install(self): with (root / 'adapters/plugins.tsv').open(encoding='utf-8-sig', newline='') as stream: entry = next(entry for entry in csv.DictReader(stream, delimiter='\t') if entry['name'] == 'Superpowers') self.manager().sync([entry], ['claude'], {'Superpowers'}) self.assertEqual([ ['claude', 'plugin', 'marketplace', 'add', 'anthropics/claude-plugins-official'], ['claude', 'plugin', 'marketplace', 'update', 'claude-plugins-official'], ['claude', 'plugin', 'install', 'superpowers@claude-plugins-official', '--scope', 'user'], ], self.commands) def test_claude_failed_marketplace_refresh_does_not_install_or_claim(self): with patch.object(extensions, 'run_command', side_effect=ValueError('refresh failed')) as command: with self.assertRaisesRegex(ValueError, 'refresh failed'): self.manager().sync([plugin()], ['claude'], {'Example'}) self.assertEqual(['claude', 'plugin', 'marketplace', 'update', 'test'], command.call_args.args[0]) self.assertEqual([], self.manager().state['resources']) @unittest.skipUnless(os.name == 'nt', 'Windows hook compatibility') def test_owned_caveman_hook_repair_runs_in_powershell_and_is_idempotent(self): entry = dict(plugin('Caveman'), codex_plugin='caveman@thinkhome-caveman') root = self.home / '.codex/plugins/cache/thinkhome-caveman/caveman/1.0.0' path = root / 'hooks/hooks.json' path.parent.mkdir(parents=True) hook = {'type': 'command', 'command': 'unchanged', 'commandWindows': 'node "%PLUGIN_ROOT%\\hooks\\caveman-hook.mjs"'} path.write_text(json.dumps({'hooks': {event: [{'hooks': [dict(hook)]}] for event in ['SessionStart', 'UserPromptSubmit', 'SubagentStart']}})) upstream = path.read_bytes() self.manager().sync([entry], ['codex'], {'Caveman'}) repaired = path.read_bytes() for groups in json.loads(repaired)['hooks'].values(): self.assertNotIn('%PLUGIN_ROOT%', groups[0]['hooks'][0]['commandWindows']) self.assertEqual('unchanged', groups[0]['hooks'][0]['command']) self.manager().sync([entry], ['codex'], {'Caveman'}) self.assertEqual(repaired, path.read_bytes()) path.write_bytes(upstream) self.manager(update=True).sync([entry], ['codex'], {'Caveman'}) self.assertEqual(repaired, path.read_bytes()) if not shutil.which('node'): self.skipTest('Node is unavailable for launcher execution') (root / 'hooks/caveman-hook.mjs').write_text("process.stdout.write('hook-ok');") command = json.loads(repaired)['hooks']['SessionStart'][0]['hooks'][0]['commandWindows'] result = subprocess.run(['powershell', '-NoProfile', '-Command', command], env=dict(os.environ, PLUGIN_ROOT=str(root)), capture_output=True, text=True, timeout=10) self.assertEqual(0, result.returncode, result.stderr) self.assertEqual('hook-ok', result.stdout) @unittest.skipUnless(os.name == 'nt', 'Windows hook compatibility') def test_caveman_repair_preserves_foreign_and_custom_hooks_and_dry_runs(self): entry = dict(plugin('Caveman'), codex_plugin='caveman@thinkhome-caveman') path = self.home / '.codex/plugins/cache/thinkhome-caveman/caveman/1.0.0/hooks/hooks.json' path.parent.mkdir(parents=True) original = json.dumps({'hooks': {'SessionStart': [{'hooks': [{'commandWindows': 'node "%PLUGIN_ROOT%\\hooks\\caveman-hook.mjs"'}]}]}}).encode() path.write_bytes(original) self.installed['codex'][entry['codex_plugin']] = {} self.manager().sync([entry], ['codex'], {'Caveman'}) self.assertEqual([], self.manager().state['resources']) self.assertEqual(original, path.read_bytes()) self.installed['codex'].clear() self.manager(dry_run=True).sync([entry], ['codex'], {'Caveman'}) self.assertEqual(original, path.read_bytes()) original = b'{"hooks":{"SessionStart":[{"hooks":[{"commandWindows":"custom"}]}]}}' path.write_bytes(original) self.manager().sync([entry], ['codex'], {'Caveman'}) self.assertEqual(original, path.read_bytes()) def test_dry_run_summary_has_one_line(self): with contextlib.redirect_stdout(io.StringIO()) as output: self.manager(dry_run=True, summary=True).sync([plugin(), skill()], ['codex'], {'Example', 'Humanizer'}) self.assertEqual('EXTENSIONS: reconciliation complete (dry run)\n', output.getvalue()) def test_manifest_has_no_qmd_extension(self): rows = (root / 'adapters/plugins.tsv').read_text(encoding='utf-8').splitlines() self.assertFalse(any(row.split(' ')[0] == 'QMD' for row in rows[1:])) def test_legacy_qmd_record_still_validates(self): (self.home / '.my-ai-configuration').mkdir() (self.home / '.my-ai-configuration/extensions.json').write_text(json.dumps({'version': 1, 'resources': [ {'client': 'shared', 'name': 'QMD', 'kind': 'qmd', 'package': '@tobilu/qmd', 'clients': ['codex']}]})) self.manager() # must not raise def test_managed_mcporter_updates_when_requested(self): self.manager().sync([mcporter()], ['codex'], {'MCPorter'}) self.commands.clear() with patch.object(extensions.shutil, 'which', return_value='mcporter'): with patch.object(extensions.Manager, 'global_packages', return_value={'mcporter'}): self.manager(update=True).sync([mcporter()], ['codex'], {'MCPorter'}) self.assertEqual([['npm', 'update', '--global', 'mcporter']], self.commands) def test_managed_mcporter_reinstalls_when_command_is_missing(self): self.manager().sync([mcporter()], ['codex'], {'MCPorter'}) self.commands.clear() with patch.object(extensions.shutil, 'which', return_value=None): self.manager().sync([mcporter()], ['codex'], {'MCPorter'}) self.assertEqual([['npm', 'install', '--global', 'mcporter']], self.commands) def test_mcporter_replaces_direct_context7_plugin(self): self.manager().sync([mcp(), mcporter()], ['codex'], {'Context7', 'MCPorter'}) config = str(self.home / '.mcporter/mcporter.json') self.assertIn(['mcporter', '--config', config, 'config', 'add', 'context7', 'https://mcp.context7.com/mcp'], self.commands) self.assertNotIn(['codex', 'plugin', 'add', 'context7@test'], self.commands) def test_removes_mcporter_managed_mcp_before_mcporter(self): self.manager().sync([mcp(), mcporter()], ['codex'], {'Context7', 'MCPorter'}) self.commands.clear() self.manager().sync([], ['codex'], set()) config = str(self.home / '.mcporter/mcporter.json') self.assertEqual([ ['mcporter', '--config', config, 'config', 'remove', 'context7'], ['npm', 'uninstall', '--global', 'mcporter'], ], self.commands) def test_foreign_empty_directory_survives_skill_removal(self): self.manager().sync([skill()], ['codex'], {'Humanizer'}) foreign = self.home / '.agents/skills/humanizer/foreign-empty' foreign.mkdir() self.manager().sync([], ['codex'], set()) self.assertTrue(foreign.is_dir()) def test_foreign_skill_link_is_preserved(self): foreign = self.home / 'foreign' foreign.mkdir() target = self.home / '.agents/skills/humanizer' target.parent.mkdir(parents=True) try: target.symlink_to(foreign, target_is_directory=True) except OSError: self.skipTest('Symbolic links are unavailable on this host.') self.manager().sync([skill()], ['codex'], {'Humanizer'}) self.assertTrue(target.is_symlink()) extensions.fetch_skill.assert_not_called() class CommandTests(unittest.TestCase): def test_subprocess_uses_intended_home_and_cmd_launcher(self): with tempfile.TemporaryDirectory(prefix='managed extensions ') as temporary: home = Path(temporary) executable = home / ('mock.cmd' if os.name == 'nt' else 'mock') if os.name == 'nt': executable.write_text('@echo off\necho %CODEX_HOME%\necho %CLAUDE_CONFIG_DIR%\n') else: executable.write_text('#!/bin/sh\nprintf "%s\\n" "$CODEX_HOME" "$CLAUDE_CONFIG_DIR"\n') executable.chmod(0o700) with contextlib.redirect_stdout(io.StringIO()): result = extensions.run_command([str(executable), 'plugin', 'list', '--json'], True, home) self.assertEqual([str(home / '.codex'), str(home / '.claude')], result.splitlines()) if __name__ == '__main__': unittest.main()