codexpi/tests/test_dev_studio.py

"""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