156 lines
8.0 KiB
Python
156 lines
8.0 KiB
Python
|
|
from codexpi.database import Database
|
||
|
|
|
||
|
|
|
||
|
|
def register(logged_in):
|
||
|
|
client, app, projects, _, headers = logged_in
|
||
|
|
root = projects / 'effort'
|
||
|
|
root.mkdir()
|
||
|
|
project = client.post('/api/projects', headers=headers, json={'name': 'Effort', 'path': str(root)}).json()
|
||
|
|
return client, app, headers, project['id']
|
||
|
|
|
||
|
|
|
||
|
|
def test_effort_updates_existing_sessions_and_new_defaults(logged_in):
|
||
|
|
client, app, headers, project_id = register(logged_in)
|
||
|
|
url = f'/api/projects/{project_id}'
|
||
|
|
session = client.post(url + '/conversations', headers=headers, json={'title': 'Existing'}).json()
|
||
|
|
result = client.put(url + '/settings', headers=headers, json={'reasoning_effort': 'high'})
|
||
|
|
assert result.status_code == 200
|
||
|
|
assert client.get(url + '/settings').json()['reasoning_effort'] == 'high'
|
||
|
|
assert client.get(url + '/conversations').json()[0]['reasoning_effort'] == 'high'
|
||
|
|
created = client.post(url + '/conversations', headers=headers, json={'title': 'New'}).json()
|
||
|
|
assert created['reasoning_effort'] == 'high'
|
||
|
|
explicit = client.post(url + '/conversations', headers=headers, json={'title': 'Override', 'reasoning_effort': 'low'}).json()
|
||
|
|
assert explicit['reasoning_effort'] == 'low'
|
||
|
|
assert app.state.db.one('SELECT reasoning_effort FROM conversations WHERE id=?', (session['id'],))['reasoning_effort'] == 'high'
|
||
|
|
|
||
|
|
|
||
|
|
def test_project_effort_validates_models_and_csrf(logged_in):
|
||
|
|
client, app, headers, project_id = register(logged_in)
|
||
|
|
url = f'/api/projects/{project_id}'
|
||
|
|
client.post(url + '/conversations', headers=headers, json={'model': 'gpt-5.5'})
|
||
|
|
assert 'ultra' not in client.get(url + '/settings').json()['efforts']
|
||
|
|
for effort in ('invalid', 'ultra'):
|
||
|
|
assert client.put(url + '/settings', headers=headers, json={'reasoning_effort': effort}).status_code == 400
|
||
|
|
assert client.put(url + '/settings', json={'reasoning_effort': 'high'}).status_code == 403
|
||
|
|
assert client.get(url + '/settings').json()['reasoning_effort'] == 'medium'
|
||
|
|
|
||
|
|
|
||
|
|
def test_project_effort_migrates_existing_database(tmp_path):
|
||
|
|
import sqlite3
|
||
|
|
path = tmp_path / 'old.db'
|
||
|
|
with sqlite3.connect(path) as conn:
|
||
|
|
conn.execute('CREATE TABLE projects (id TEXT PRIMARY KEY, name TEXT, path TEXT, full_access INTEGER, created_at TEXT)')
|
||
|
|
conn.execute("INSERT INTO projects VALUES ('p','Demo','/tmp',0,'2026-01-01')")
|
||
|
|
db = Database(path)
|
||
|
|
db.initialize()
|
||
|
|
db.initialize()
|
||
|
|
assert db.one("SELECT reasoning_effort FROM projects WHERE id='p'")['reasoning_effort'] == 'medium'
|
||
|
|
|
||
|
|
|
||
|
|
def test_project_model_changes_existing_sessions_and_defaults(logged_in):
|
||
|
|
client, app, headers, project_id = register(logged_in)
|
||
|
|
url = f'/api/projects/{project_id}'
|
||
|
|
client.post(url + '/conversations', headers=headers, json={'model': 'gpt-5.5'})
|
||
|
|
result = client.put(url + '/settings', headers=headers, json={'model': 'gpt-5.6-sol', 'reasoning_effort': 'high'})
|
||
|
|
assert result.status_code == 200
|
||
|
|
assert client.get(url + '/settings').json()['model'] == 'gpt-5.6-sol'
|
||
|
|
assert client.get(url + '/conversations').json()[0]['model'] == 'gpt-5.6-sol'
|
||
|
|
assert client.post(url + '/conversations', headers=headers, json={}).json()['model'] == 'gpt-5.6-sol'
|
||
|
|
assert client.post(url + '/conversations', headers=headers, json={'model': ''}).json()['model'] is None
|
||
|
|
for model, effort in [('invalid', 'high'), ('gpt-5.5', 'ultra')]:
|
||
|
|
assert client.put(url + '/settings', headers=headers, json={'model': model, 'reasoning_effort': effort}).status_code == 400
|
||
|
|
assert client.get(url + '/settings').json()['model'] == 'gpt-5.6-sol'
|
||
|
|
assert client.put(url + '/settings', headers=headers, json={'model': '', 'reasoning_effort': 'medium'}).status_code == 200
|
||
|
|
assert all(row['model'] is None for row in client.get(url + '/conversations').json())
|
||
|
|
|
||
|
|
|
||
|
|
def test_model_default_limits_effort_before_first_session(logged_in):
|
||
|
|
client, app, headers, project_id = register(logged_in)
|
||
|
|
url = f'/api/projects/{project_id}/settings'
|
||
|
|
assert client.put(url, headers=headers, json={'model':'gpt-5.5', 'reasoning_effort':'high'}).status_code == 200
|
||
|
|
assert client.put(url, headers=headers, json={'reasoning_effort':'ultra'}).status_code == 400
|
||
|
|
assert 'ultra' not in client.get(url).json()['efforts']
|
||
|
|
|
||
|
|
|
||
|
|
def test_project_settings_update_name_and_source_files(logged_in):
|
||
|
|
client, app, headers, project_id = register(logged_in)
|
||
|
|
project = app.state.db.one('SELECT path FROM projects WHERE id=?', (project_id,))
|
||
|
|
root = __import__('pathlib').Path(project['path'])
|
||
|
|
(root / 'README.md').write_text('# Demo', encoding='utf-8')
|
||
|
|
(root / 'src').mkdir()
|
||
|
|
(root / 'src' / 'main.py').write_text('print("ok")', encoding='utf-8')
|
||
|
|
(root / '.git').mkdir()
|
||
|
|
(root / '.git' / 'config').write_text('secret', encoding='utf-8')
|
||
|
|
|
||
|
|
files = client.get(f'/api/projects/{project_id}/source-files').json()['files']
|
||
|
|
assert files == ['README.md', 'src/main.py']
|
||
|
|
result = client.put(
|
||
|
|
f'/api/projects/{project_id}/settings', headers=headers,
|
||
|
|
json={
|
||
|
|
'name': 'Projet renommé', 'reasoning_effort': 'medium',
|
||
|
|
'execution_group': 'tv',
|
||
|
|
'source_file': 'README.md', 'additional_files': ['src/main.py'],
|
||
|
|
},
|
||
|
|
)
|
||
|
|
assert result.status_code == 200
|
||
|
|
saved = client.get(f'/api/projects/{project_id}/settings').json()
|
||
|
|
assert saved['name'] == 'Projet renommé'
|
||
|
|
assert saved['execution_group'] == 'tv'
|
||
|
|
assert saved['source_file'] == 'README.md'
|
||
|
|
assert saved['additional_files'] == ['src/main.py']
|
||
|
|
listed = client.get('/api/projects').json()
|
||
|
|
assert next(item for item in listed if item['id'] == project_id)['additional_files'] == ['src/main.py']
|
||
|
|
|
||
|
|
|
||
|
|
def test_project_settings_reject_source_outside_project(logged_in):
|
||
|
|
client, _, headers, project_id = register(logged_in)
|
||
|
|
result = client.put(
|
||
|
|
f'/api/projects/{project_id}/settings', headers=headers,
|
||
|
|
json={'reasoning_effort': 'medium', 'source_file': '../secret', 'additional_files': []},
|
||
|
|
)
|
||
|
|
assert result.status_code == 400
|
||
|
|
|
||
|
|
|
||
|
|
def test_standby_api_does_not_recover_workers(tmp_path, monkeypatch):
|
||
|
|
from codexpi.main import should_recover_workers
|
||
|
|
|
||
|
|
monkeypatch.setenv('CODEXPI_SKIP_WORKER_RECOVERY', 'true')
|
||
|
|
assert should_recover_workers() is False
|
||
|
|
monkeypatch.delenv('CODEXPI_SKIP_WORKER_RECOVERY')
|
||
|
|
assert should_recover_workers() is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_remote_project_accepts_local_reference_from_file_browser(logged_in):
|
||
|
|
client, app, headers, project_id = register(logged_in)
|
||
|
|
project = app.state.db.one('SELECT path FROM projects WHERE id=?', (project_id,))
|
||
|
|
root = __import__('pathlib').Path(project['path'])
|
||
|
|
reference = root / 'instructions.md'
|
||
|
|
reference.write_text('Contexte local', encoding='utf-8')
|
||
|
|
app.state.db.execute("UPDATE projects SET kind='remote' WHERE id=?", (project_id,))
|
||
|
|
|
||
|
|
listing = client.get(f'/api/filesystem/files?path={root}').json()
|
||
|
|
assert listing['files'] == [{'name': 'instructions.md', 'path': str(reference), 'size': 14}]
|
||
|
|
saved = client.put(
|
||
|
|
f'/api/projects/{project_id}/settings', headers=headers,
|
||
|
|
json={'reasoning_effort': 'medium', 'source_file': '', 'additional_files': [str(reference)]},
|
||
|
|
)
|
||
|
|
assert saved.status_code == 200
|
||
|
|
assert client.get(f'/api/projects/{project_id}/settings').json()['additional_files'] == [str(reference)]
|
||
|
|
|
||
|
|
|
||
|
|
def test_remote_project_accepts_local_directory_as_source(logged_in):
|
||
|
|
client, app, headers, project_id = register(logged_in)
|
||
|
|
project = app.state.db.one('SELECT path FROM projects WHERE id=?', (project_id,))
|
||
|
|
root = __import__('pathlib').Path(project['path'])
|
||
|
|
source_directory = root / 'documentation'
|
||
|
|
source_directory.mkdir()
|
||
|
|
(source_directory / 'guide.md').write_text('Guide local', encoding='utf-8')
|
||
|
|
app.state.db.execute("UPDATE projects SET kind='remote' WHERE id=?", (project_id,))
|
||
|
|
|
||
|
|
saved = client.put(
|
||
|
|
f'/api/projects/{project_id}/settings', headers=headers,
|
||
|
|
json={'reasoning_effort': 'medium', 'source_file': '', 'additional_files': [str(source_directory)]},
|
||
|
|
)
|
||
|
|
assert saved.status_code == 200
|
||
|
|
assert client.get(f'/api/projects/{project_id}/settings').json()['additional_files'] == [str(source_directory)]
|