214 lines
12 KiB
Python
214 lines
12 KiB
Python
"""Studio integration tests use temporary repositories and synthetic engine events."""
|
|
import json
|
|
import subprocess
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
|
|
def repository(logged_in):
|
|
client, app, projects, _, headers = logged_in
|
|
root = projects / 'studio-demo'
|
|
root.mkdir()
|
|
for args in (["init", "-b", "main"], ["-c", "user.name=Test", "-c", "user.email=test@example.invalid", "commit", "--allow-empty", "-m", "Fixture"]):
|
|
subprocess.run(['git', *args], cwd=root, check=True, capture_output=True)
|
|
owner = app.state.db.one('SELECT id FROM users LIMIT 1')['id']
|
|
project_id = str(uuid.uuid4())
|
|
app.state.db.execute('INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)', (project_id, 'Fixture', str(root), owner, '2026-01-01'))
|
|
response = client.post('/api/dev-studio/requests', headers=headers, json={'project_id': project_id, 'name': 'Demo', 'description': 'Créer une application de démonstration avec tests essentiels.'})
|
|
assert response.status_code == 201, response.text
|
|
return client, app, headers, root, response.json()
|
|
|
|
|
|
def test_plan_does_not_launch_generation(logged_in):
|
|
client, app, _, root, request = repository(logged_in)
|
|
assert request['status'] == 'PLANNED'
|
|
assert len(request['plan_hash']) == 64
|
|
assert request['base_sha'] == subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=root, text=True).strip()
|
|
assert app.state.db.all('SELECT * FROM prompt_queue') == []
|
|
assert not list((app.state.codex.data_dir / 'worktrees').glob('*'))
|
|
assert client.get(f"/api/dev-studio/requests/{request['id']}").status_code == 200
|
|
|
|
|
|
def test_studio_project_picker_lists_authorized_local_projects_in_all_modes(logged_in):
|
|
client, app, projects, _, _ = logged_in
|
|
owner = app.state.db.one('SELECT id FROM users LIMIT 1')['id']
|
|
safe_id, full_id, remote_id = (str(uuid.uuid4()) for _ in range(3))
|
|
for project_id, name, kind, full_access in (
|
|
(safe_id, 'Safe project', 'local', 0),
|
|
(full_id, 'Full access project', 'local', 1),
|
|
(remote_id, 'Remote project', 'remote', 0),
|
|
):
|
|
root = projects / project_id
|
|
root.mkdir()
|
|
app.state.db.execute(
|
|
'INSERT INTO projects(id,name,path,owner_id,kind,full_access,created_at) VALUES(?,?,?,?,?,?,?)',
|
|
(project_id, name, str(root), owner, kind, full_access, '2026-01-01'),
|
|
)
|
|
|
|
response = client.get('/api/dev-studio/projects')
|
|
|
|
assert response.status_code == 200
|
|
rows = response.json()
|
|
assert {item['id'] for item in rows} == {safe_id, full_id}
|
|
assert {item['name'] for item in rows} == {'Safe project', 'Full access project'}
|
|
|
|
|
|
def test_approval_checks_hash_csrf_and_engine(logged_in):
|
|
client, app, headers, root, request = repository(logged_in)
|
|
url = f"/api/dev-studio/requests/{request['id']}/approve"
|
|
assert client.post(url, json={'plan_hash': request['plan_hash']}).status_code == 403
|
|
assert client.post(url, headers=headers, json={'plan_hash': '0' * 64}).status_code == 409
|
|
result = client.post(url, headers=headers, json={'plan_hash': request['plan_hash']})
|
|
assert result.status_code == 409
|
|
assert 'Codex' in result.text
|
|
assert not list((app.state.codex.data_dir / 'worktrees').glob('*'))
|
|
|
|
|
|
def test_generation_uses_dedicated_worktree_and_project_mode(logged_in, monkeypatch):
|
|
client, app, headers, root, request = repository(logged_in)
|
|
monkeypatch.setattr(app.state.codex, 'detect', lambda *args: '/synthetic-engine')
|
|
monkeypatch.setattr(app.state.codex, '_spawn', lambda *args: 123456789)
|
|
url = f"/api/dev-studio/requests/{request['id']}/approve"
|
|
response = client.post(url, headers=headers, json={'plan_hash': request['plan_hash']})
|
|
assert response.status_code == 200, response.text
|
|
result = response.json()
|
|
conversation = app.state.db.one('SELECT * FROM conversations WHERE id=?', (result['conversation_id'],))
|
|
worktree = Path(conversation['worktree_path'])
|
|
assert worktree != root and (worktree / '.git').is_file()
|
|
assert conversation['branch_name'].startswith('studio/')
|
|
assert client.post(url, headers=headers, json={'plan_hash': request['plan_hash']}).status_code == 409
|
|
queue = app.state.db.one('SELECT * FROM prompt_queue WHERE id=?', (result['queue_id'],))
|
|
assert 'production' in queue['content']
|
|
(worktree / 'demo.txt').write_text('Synthetic fixture, not generated by AI')
|
|
app.state.db.execute("UPDATE prompt_queue SET status='done' WHERE id=?", (queue['id'],))
|
|
detail = client.get(f"/api/dev-studio/requests/{request['id']}").json()
|
|
assert detail['status'] == 'BLOCKED' # No evidence cannot mean success.
|
|
metadata = {'queue_id': queue['id'], 'actions': [{'command': 'python3 -m pytest -q', 'status': 'done', 'exit_code': 0, 'output': '1 passed (synthetic fixture)'}]}
|
|
app.state.db.execute('INSERT INTO messages(id,conversation_id,role,content,metadata,created_at) VALUES(?,?,?,?,?,?)', (str(uuid.uuid4()), conversation['id'], 'assistant', 'Synthetic V0 report', json.dumps(metadata), '2026-01-01'))
|
|
detail = client.get(f"/api/dev-studio/requests/{request['id']}").json()
|
|
assert detail['status'] == 'READY'
|
|
assert detail['tests'][0]['output'].startswith('1 passed')
|
|
assert detail['git']['files'][0]['path'] == 'demo.txt'
|
|
assert not (root / 'demo.txt').exists()
|
|
assert 'Synthetic V0 report' in detail['report']
|
|
|
|
|
|
def test_private_requests_and_project_access(logged_in):
|
|
client, app, headers, root, request = repository(logged_in)
|
|
app.state.auth.create_user('other', 'a-different-long-test-password', is_admin=False)
|
|
client.post('/api/auth/logout', headers=headers)
|
|
login = client.post('/api/auth/login', json={'username': 'other', 'password': 'a-different-long-test-password'}).json()
|
|
other_headers = {'X-CSRF-Token': login['csrf_token']}
|
|
assert client.get('/api/dev-studio/requests').json() == []
|
|
assert client.get(f"/api/dev-studio/requests/{request['id']}").status_code == 404
|
|
assert client.post(f"/api/dev-studio/requests/{request['id']}/approve", headers=other_headers, json={'plan_hash': request['plan_hash']}).status_code == 404
|
|
assert client.post('/api/dev-studio/requests', headers=other_headers, json={'project_id': request['project_id'], 'name': 'Forbidden', 'description': 'Tentative sur un autre projet utilisateur.'}).status_code == 404
|
|
|
|
|
|
def test_full_access_project_can_generate_in_its_dedicated_worktree(logged_in, monkeypatch):
|
|
client, app, headers, root, request = repository(logged_in)
|
|
app.state.db.execute('UPDATE projects SET full_access=1 WHERE id=?', (request['project_id'],))
|
|
monkeypatch.setattr(app.state.codex, 'detect', lambda *args: '/synthetic-engine')
|
|
monkeypatch.setattr(app.state.codex, '_spawn', lambda *args: 123456789)
|
|
result = client.post(f"/api/dev-studio/requests/{request['id']}/approve", headers=headers, json={'plan_hash': request['plan_hash']})
|
|
assert result.status_code == 200, result.text
|
|
conversation = app.state.db.one('SELECT * FROM conversations WHERE id=?', (result.json()['conversation_id'],))
|
|
assert Path(conversation['worktree_path']) != root
|
|
|
|
|
|
def test_unborn_repository_is_not_initialized_by_studio(logged_in):
|
|
client, app, projects, _, headers = logged_in
|
|
root = projects / 'unborn'
|
|
root.mkdir()
|
|
subprocess.run(['git', 'init', '-b', 'main'], cwd=root, check=True, capture_output=True)
|
|
project_id = str(uuid.uuid4())
|
|
owner = app.state.db.one('SELECT id FROM users LIMIT 1')['id']
|
|
app.state.db.execute('INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)', (project_id, 'Unborn', str(root), owner, '2026-01-01'))
|
|
result = client.post('/api/dev-studio/requests', headers=headers, json={'project_id': project_id, 'name': 'Demo', 'description': 'Une application de démonstration testable.'})
|
|
assert result.status_code == 409
|
|
assert subprocess.run(['git', 'rev-parse', '--verify', 'HEAD'], cwd=root, capture_output=True).returncode != 0
|
|
|
|
|
|
def test_private_preview_is_limited_to_worktree_loopback(logged_in, monkeypatch):
|
|
import socket
|
|
import sys
|
|
import time
|
|
client, app, headers, root, request = repository(logged_in)
|
|
monkeypatch.setattr(app.state.codex, 'detect', lambda *args: '/synthetic-engine')
|
|
monkeypatch.setattr(app.state.codex, '_spawn', lambda *args: 123456789)
|
|
response = client.post(f"/api/dev-studio/requests/{request['id']}/approve", headers=headers, json={'plan_hash': request['plan_hash']}).json()
|
|
worktree = Path(response['worktree_path'])
|
|
(worktree / 'index.html').write_text('<h1>Synthetic preview fixture</h1>')
|
|
with socket.socket() as sock:
|
|
sock.bind(('127.0.0.1', 0))
|
|
port = sock.getsockname()[1]
|
|
process = subprocess.Popen([sys.executable, '-m', 'http.server', str(port), '--bind', '127.0.0.1'], cwd=worktree, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
try:
|
|
for _ in range(50):
|
|
try:
|
|
with socket.create_connection(('127.0.0.1', port), timeout=.1):
|
|
break
|
|
except OSError:
|
|
time.sleep(.05)
|
|
result = client.get(f"/api/dev-studio/requests/{request['id']}/preview/{port}/")
|
|
assert result.status_code == 200, result.text
|
|
assert 'Synthetic preview fixture' in result.text
|
|
assert 'sandbox allow-scripts' in result.headers['content-security-policy']
|
|
assert client.get(f"/api/dev-studio/requests/{request['id']}/preview/8000/").status_code == 404
|
|
assert client.get(f"/api/dev-studio/requests/{request['id']}/preview/{port}/missing").status_code == 409
|
|
finally:
|
|
process.terminate()
|
|
process.wait(timeout=5)
|
|
|
|
|
|
def test_real_worker_with_synthetic_engine_executes_real_test(logged_in):
|
|
import sys
|
|
import time
|
|
client, app, headers, root, request = repository(logged_in)
|
|
engine = root.parent / 'synthetic-codex'
|
|
engine.write_text('''#!/usr/bin/python3
|
|
import json, pathlib, subprocess, sys
|
|
assert '--sandbox' in sys.argv and 'workspace-write' in sys.argv
|
|
assert '--dangerously-bypass-approvals-and-sandbox' not in sys.argv
|
|
pathlib.Path('test_fixture.py').write_text('import unittest\\nclass Example(unittest.TestCase):\\n def test_sum(self): self.assertEqual(2 + 2, 4)\\n')
|
|
result = subprocess.run([sys.executable, '-m', 'unittest', '-q'], capture_output=True, text=True)
|
|
print(json.dumps({'type': 'item.completed', 'item': {'id': 'real-test', 'type': 'command_execution', 'command': 'python3 -m unittest -q', 'exit_code': result.returncode, 'aggregated_output': result.stdout + result.stderr}}))
|
|
print(json.dumps({'type': 'item.completed', 'item': {'type': 'agent_message', 'text': 'Synthetic engine fixture: one real unittest executed. This is not an AI generation validation.'}}))
|
|
''')
|
|
engine.chmod(0o755)
|
|
app.state.codex.codex_bin = str(engine)
|
|
result = client.post(f"/api/dev-studio/requests/{request['id']}/approve", headers=headers, json={'plan_hash': request['plan_hash']})
|
|
assert result.status_code == 200, result.text
|
|
queue_id = result.json()['queue_id']
|
|
for _ in range(100):
|
|
queue = app.state.db.one('SELECT * FROM prompt_queue WHERE id=?', (queue_id,))
|
|
if queue['status'] not in ('running', 'queued'):
|
|
break
|
|
time.sleep(.1)
|
|
assert queue['status'] == 'done', queue.get('error')
|
|
detail = client.get(f"/api/dev-studio/requests/{request['id']}").json()
|
|
assert detail['status'] == 'READY', detail
|
|
assert detail['tests'][0]['exit_code'] == 0
|
|
assert 'Ran 1 test' in detail['tests'][0]['output']
|
|
assert not (root / 'test_fixture.py').exists()
|
|
|
|
|
|
def test_verification_evidence_rejects_echo_and_masked_failures():
|
|
from codexpi.dev_studio_service import is_verification_command
|
|
assert is_verification_command("/bin/bash -lc 'pytest -q'")
|
|
assert is_verification_command('cd frontend && npm run test')
|
|
assert is_verification_command('npm --prefix frontend run build')
|
|
assert not is_verification_command('echo pytest')
|
|
assert not is_verification_command('pytest || true')
|
|
assert not is_verification_command('pytest -q||true')
|
|
assert not is_verification_command("/bin/bash -lc 'pytest; echo passed'")
|
|
|
|
|
|
def test_nested_directory_cannot_use_parent_repository(logged_in):
|
|
client, app, headers, root, request = repository(logged_in)
|
|
nested = root / 'nested'
|
|
nested.mkdir()
|
|
app.state.db.execute('UPDATE projects SET path=? WHERE id=?', (str(nested), request['project_id']))
|
|
response = client.post('/api/dev-studio/requests', headers=headers, json={'project_id': request['project_id'], 'name': 'Nested', 'description': 'Cette demande ne doit pas utiliser le dépôt parent.'})
|
|
assert response.status_code == 409
|