252 lines
15 KiB
Python
252 lines
15 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent.parent
|
|
SHELLS = [
|
|
('powershell', shutil.which('powershell') or shutil.which('pwsh')),
|
|
('bash', str(Path(os.environ.get('ProgramFiles', 'C:/Program Files')) / 'Git/bin/bash.exe')
|
|
if os.name == 'nt' else shutil.which('bash')),
|
|
]
|
|
|
|
|
|
class SessionStateHookTests(unittest.TestCase):
|
|
def test_git_identity_and_stale_verification(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
workspace = Path(directory)
|
|
env = dict(os.environ, GIT_AUTHOR_NAME='Test', GIT_AUTHOR_EMAIL='test@example.invalid', GIT_COMMITTER_NAME='Test', GIT_COMMITTER_EMAIL='test@example.invalid')
|
|
subprocess.run(['git', '-C', str(workspace), 'init', '-q'], check=True)
|
|
(workspace / 'file').write_text('one')
|
|
subprocess.run(['git', '-C', str(workspace), 'add', 'file'], check=True)
|
|
subprocess.run(['git', '-C', str(workspace), 'commit', '-qm', 'one'], check=True, env=env)
|
|
command = [sys.executable, str(ROOT / 'shared/hooks/scripts/session-state.py'), '--workspace', str(workspace)]
|
|
self.assertEqual(subprocess.run(command + ['--verified', 'check'], capture_output=True).returncode, 0)
|
|
state = json.loads((workspace / '.ai-session/state.json').read_text())
|
|
self.assertTrue(state['head'])
|
|
self.assertEqual(state['verification'][0]['head'], state['head'])
|
|
(workspace / 'file').write_text('two')
|
|
subprocess.run(['git', '-C', str(workspace), 'add', 'file'], check=True)
|
|
subprocess.run(['git', '-C', str(workspace), 'commit', '-qm', 'two'], check=True, env=env)
|
|
self.assertEqual(subprocess.run(command, capture_output=True).returncode, 0)
|
|
state = json.loads((workspace / '.ai-session/state.json').read_text())
|
|
self.assertTrue(state['verificationStale'])
|
|
self.assertTrue(state['verification'][0]['stale'])
|
|
|
|
def test_writer_migrates_legacy_state_and_requires_reset_for_new_goal(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
workspace = Path(directory)
|
|
state_file = workspace / '.ai-session/state.json'
|
|
state_file.parent.mkdir()
|
|
state_file.write_text(json.dumps({'goal': 'Original', 'changedFiles': ['a'],
|
|
'pending': ['b'], 'importantFindings': ['c']}))
|
|
command = [sys.executable, str(ROOT / 'shared/hooks/scripts/session-state.py'),
|
|
'--workspace', str(workspace)]
|
|
result = subprocess.run(command + ['--verified', 'passed'], capture_output=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
state = json.loads(state_file.read_text())
|
|
self.assertEqual(state['changed'], ['a'])
|
|
self.assertEqual(state['open'], ['b'])
|
|
self.assertEqual(state['decisions'], ['c'])
|
|
self.assertFalse({'changedFiles', 'pending', 'importantFindings'} & state.keys())
|
|
before = state_file.read_bytes()
|
|
result = subprocess.run(command + ['--goal', 'New'], capture_output=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(state_file.read_bytes(), before)
|
|
result = subprocess.run(command + ['--reset', '--goal', 'New'], capture_output=True)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
state = json.loads(state_file.read_text())
|
|
self.assertEqual(state['goal'], 'New')
|
|
self.assertEqual(state['verified'], [])
|
|
self.assertEqual(state['changed'], [])
|
|
state_file.write_text('{"goal":"New","verified":"invalid"}')
|
|
before = state_file.read_bytes()
|
|
result = subprocess.run(command + ['--open', 'next'], capture_output=True)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(state_file.read_bytes(), before)
|
|
|
|
def test_writers_preserve_partial_updates_and_escape_values(self):
|
|
for shell, executable in SHELLS:
|
|
if not executable or not Path(executable).exists():
|
|
continue
|
|
with self.subTest(shell=shell), tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
repository = base / 'repository'
|
|
shutil.copytree(ROOT / 'shared/hooks/scripts', repository / 'shared/hooks/scripts')
|
|
(repository / 'scripts/commands').mkdir(parents=True)
|
|
for filename in ('session-state.ps1', 'session-state.sh'):
|
|
shutil.copy(ROOT / 'scripts/commands' / filename, repository / 'scripts/commands' / filename)
|
|
workspace = base / 'workspace ΓΌ'
|
|
workspace.mkdir()
|
|
environment = dict(os.environ, TASK_WORKSPACE=str(workspace),
|
|
TASK_GOAL='Repair "quotes"\n' + 'x' * 300)
|
|
if shell == 'powershell':
|
|
prefix = [executable, '-NoProfile', '-ExecutionPolicy', 'Bypass', '-Command']
|
|
commands = [
|
|
'& ./scripts/commands/session-state.ps1 -Workspace $env:TASK_WORKSPACE -Goal $env:TASK_GOAL -Decisions @("keep", "line two") -Pending @("check")',
|
|
'& ./scripts/commands/session-state.ps1 -Workspace $env:TASK_WORKSPACE -Verified @("passed")',
|
|
'& ./scripts/commands/session-state.ps1 -Workspace $env:TASK_WORKSPACE -Pointer',
|
|
]
|
|
else:
|
|
prefix = [executable, '-c']
|
|
commands = [
|
|
'bash ./scripts/commands/session-state.sh --workspace "$TASK_WORKSPACE" --goal "$TASK_GOAL" --decisions keep "line two" --open check',
|
|
'bash ./scripts/commands/session-state.sh --workspace "$TASK_WORKSPACE" --verified passed',
|
|
'bash ./scripts/commands/session-state.sh --workspace "$TASK_WORKSPACE" --pointer',
|
|
]
|
|
for command in commands:
|
|
result = subprocess.run(prefix + [command], cwd=repository, capture_output=True,
|
|
encoding='utf-8', timeout=20, env=environment)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
state_file = workspace / '.ai-session/state.json'
|
|
self.assertIn(str(state_file), result.stdout)
|
|
state = json.loads(state_file.read_text(encoding='utf-8-sig'))
|
|
self.assertEqual(state['goal'], environment['TASK_GOAL'])
|
|
self.assertEqual(state['decisions'], ['keep', 'line two'])
|
|
self.assertEqual(state['open'], ['check'])
|
|
self.assertEqual(state['verified'], ['passed'])
|
|
self.assertEqual(state['workspace'], str(workspace))
|
|
self.assertFalse((repository / '.ai-session').exists())
|
|
self.assertEqual([p.name for p in state_file.parent.iterdir()], ['state.json'])
|
|
|
|
def run_hook(self, shell, executable, hook, config, workspace, payload):
|
|
names = {'pointer': ('Show-SessionStatePointer.ps1', 'show-session-state-pointer.sh'),
|
|
'compact': ('Record-Compact.ps1', 'record-compact.sh')}
|
|
script = config / 'hooks/scripts' / names[hook][shell == 'bash']
|
|
command = ([executable, '-NoProfile', '-ExecutionPolicy', 'Bypass', '-File', str(script)]
|
|
if shell == 'powershell' else [executable, str(script)])
|
|
result = subprocess.run(command, cwd=workspace, input=payload, capture_output=True,
|
|
encoding='utf-8', timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual(result.stderr, '')
|
|
return result.stdout
|
|
|
|
def test_workspace_selection_and_read_only_state(self):
|
|
for shell, executable in SHELLS:
|
|
if not executable or not Path(executable).exists():
|
|
continue
|
|
with self.subTest(shell=shell), tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
config = base / 'installed config'
|
|
shutil.copytree(ROOT / 'shared/hooks/scripts', config / 'hooks/scripts')
|
|
workspace = base / 'working project'
|
|
selected = base / 'selected project ΓΌ'
|
|
empty = base / 'empty project'
|
|
for folder in (config, workspace, selected, empty):
|
|
folder.mkdir(exist_ok=True)
|
|
for folder in (config, workspace, selected):
|
|
(folder / '.ai-session').mkdir()
|
|
(folder / '.ai-session/state.json').write_text('private state content')
|
|
for hook in ('pointer', 'compact'):
|
|
for payload, expected in (
|
|
('', workspace),
|
|
('{}', workspace),
|
|
(json.dumps({'cwd': str(selected)}, ensure_ascii=False), selected),
|
|
(json.dumps({'cwd': str(empty)}), None),
|
|
('{"cwd":"relative/path"}', None),
|
|
('{"cwd":null}', None),
|
|
('{"cwd":12}', None),
|
|
('{"cwd":""}', None),
|
|
('{bad json', None),
|
|
('[]', None),
|
|
):
|
|
with self.subTest(hook=hook, payload=payload):
|
|
output = self.run_hook(shell, executable, hook, config, workspace, payload)
|
|
if expected is None:
|
|
self.assertEqual(output, '')
|
|
else:
|
|
self.assertIn(str(expected / '.ai-session/state.json'), output)
|
|
self.assertNotIn(str(config), output)
|
|
self.assertNotIn('private state content', output)
|
|
for folder in (config, workspace, selected):
|
|
self.assertEqual((folder / '.ai-session/state.json').read_text(), 'private state content')
|
|
self.assertFalse((empty / '.ai-session').exists())
|
|
events = (config / '.ai-session/telemetry.jsonl').read_text(encoding='utf-8-sig').splitlines()
|
|
self.assertEqual(len(events), 10)
|
|
self.assertTrue(all(json.loads(line)['event'] == 'compact' for line in events))
|
|
|
|
def test_missing_workspace_state_succeeds_without_config_fallback(self):
|
|
for shell, executable in SHELLS:
|
|
if not executable or not Path(executable).exists():
|
|
continue
|
|
with self.subTest(shell=shell), tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
config = base / 'config'
|
|
shutil.copytree(ROOT / 'shared/hooks/scripts', config / 'hooks/scripts')
|
|
(config / '.ai-session').mkdir()
|
|
(config / '.ai-session/state.json').write_text('{}')
|
|
workspace = base / 'workspace'
|
|
workspace.mkdir()
|
|
for hook in ('pointer', 'compact'):
|
|
self.assertEqual(self.run_hook(shell, executable, hook, config, workspace, ''), '')
|
|
self.assertEqual(list(workspace.iterdir()), [])
|
|
|
|
def test_hooks_ignore_missing_helper(self):
|
|
for shell, executable in SHELLS:
|
|
if not executable or not Path(executable).exists():
|
|
continue
|
|
with self.subTest(shell=shell), tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
config = base / 'config'
|
|
shutil.copytree(ROOT / 'shared/hooks/scripts', config / 'hooks/scripts')
|
|
(config / 'hooks/scripts/session-state.py').unlink()
|
|
workspace = base / 'workspace'
|
|
workspace.mkdir()
|
|
for hook in ('pointer', 'compact'):
|
|
self.assertEqual(self.run_hook(shell, executable, hook, config, workspace, ''), '')
|
|
|
|
def test_flashbang_ignores_unavailable_runtime(self):
|
|
for shell, executable in SHELLS:
|
|
if not executable or not Path(executable).exists():
|
|
continue
|
|
with self.subTest(shell=shell):
|
|
environment = dict(os.environ)
|
|
if shell == 'powershell':
|
|
environment['FLASHBANG_TEST_SCRIPT'] = str(ROOT / 'shared/hooks/flashbang.ps1')
|
|
command = [executable, '-NoProfile', '-ExecutionPolicy', 'Bypass', '-Command',
|
|
"function Add-Type { throw 'GUI runtime unavailable' }; & $env:FLASHBANG_TEST_SCRIPT"]
|
|
else:
|
|
environment['PATH'] = ''
|
|
command = [executable, str(ROOT / 'shared/hooks/flashbang.sh')]
|
|
result = subprocess.run(command, env=environment, capture_output=True, text=True, timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
@unittest.skipUnless(os.name == 'nt', 'Windows GUI runtime required')
|
|
def test_flashbang_cache_is_optional(self):
|
|
executable = shutil.which('powershell') or shutil.which('pwsh')
|
|
if not executable:
|
|
self.skipTest('PowerShell unavailable')
|
|
# Exercise native initialization, stopping at the UI boundary without flashing the screen.
|
|
probe = """
|
|
function New-Object {
|
|
param([string]$TypeName)
|
|
if ($TypeName -eq 'System.Windows.Forms.Form' -and ('Flashbang.Native' -as [type])) {
|
|
[Console]::WriteLine('OVERLAY_READY')
|
|
}
|
|
throw 'Stop at UI boundary'
|
|
}
|
|
& $env:FLASHBANG_TEST_SCRIPT
|
|
"""
|
|
with tempfile.TemporaryDirectory(prefix='.flashbang-test-', dir=ROOT) as directory:
|
|
blocked = Path(directory) / 'not-a-directory'
|
|
blocked.write_text('keep')
|
|
for cache in ('', str(blocked), directory):
|
|
with self.subTest(cache=cache):
|
|
environment = dict(os.environ, LOCALAPPDATA=cache,
|
|
FLASHBANG_TEST_SCRIPT=str(ROOT / 'shared/hooks/flashbang.ps1'))
|
|
result = subprocess.run([executable, '-NoProfile', '-ExecutionPolicy', 'Bypass',
|
|
'-Command', probe], env=environment, capture_output=True,
|
|
text=True, timeout=20)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn('OVERLAY_READY', result.stdout, 'Flashbang exited before creating the overlay')
|
|
self.assertEqual(blocked.read_text(), 'keep')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|