157 lines
8.2 KiB
Python
157 lines
8.2 KiB
Python
import json
|
|
import time
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from requests import Response
|
|
|
|
from codexpi.database import Database, utcnow
|
|
from codexpi.push import b64, deliver_pending, format_elapsed, notification_description, validate_subscription, vapid_keys
|
|
from codexpi.session_worker import finish_and_claim_next
|
|
|
|
|
|
def subscription(endpoint='https://web.push.apple.com/device/example'):
|
|
key = ec.generate_private_key(ec.SECP256R1()).public_key()
|
|
public = key.public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
|
|
return {'endpoint': endpoint, 'keys': {'auth': b64(b'0123456789abcdef'), 'p256dh': b64(public)}}
|
|
|
|
|
|
def test_subscription_validation_and_key_persistence(tmp_path):
|
|
db = Database(tmp_path / 'db.sqlite')
|
|
db.initialize()
|
|
assert vapid_keys(db) == vapid_keys(db)
|
|
validate_subscription(subscription())
|
|
for endpoint in ['http://web.push.apple.com/x', 'https://127.0.0.1/x', 'https://web.push.apple.com.evil.test/x', 'https://web.push.apple.com:9000/x', 'https://user@web.push.apple.com/x']:
|
|
with pytest.raises(ValueError):
|
|
validate_subscription(subscription(endpoint))
|
|
malformed = subscription()
|
|
malformed['keys']['auth'] = 'bad'
|
|
with pytest.raises(ValueError):
|
|
validate_subscription(malformed)
|
|
|
|
|
|
def test_subscriptions_require_auth_csrf_and_can_be_removed(logged_in):
|
|
client, app, _, _, headers = logged_in
|
|
data = subscription()
|
|
assert client.get('/api/push/key').json().keys() == {'public_key'}
|
|
assert client.post('/api/push/subscriptions', json=data).status_code == 403
|
|
assert client.post('/api/push/subscriptions', headers=headers, json=data).status_code == 200
|
|
assert client.post('/api/push/subscriptions', headers=headers, json=data).status_code == 200
|
|
assert len(app.state.db.all('SELECT * FROM push_subscriptions')) == 1
|
|
assert client.request('DELETE', '/api/push/subscriptions', headers=headers, json={'endpoint': data['endpoint']}).status_code == 200
|
|
assert app.state.db.all('SELECT * FROM push_subscriptions') == []
|
|
|
|
|
|
def prepare(tmp_path, result='Le résultat final de Codex'):
|
|
db = Database(tmp_path / 'db.sqlite')
|
|
db.initialize()
|
|
stamp = utcnow()
|
|
db.execute('INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)', ('p', 'Project', str(tmp_path), stamp))
|
|
db.execute('INSERT INTO conversations(id,project_id,title,status,created_at,updated_at) VALUES(?,?,?,?,?,?)', ('c', 'p', 'Task', 'running', stamp, stamp))
|
|
db.execute('INSERT INTO prompt_queue(id,conversation_id,content,status,created_at,started_at,requested_by) VALUES(?,?,?,?,?,?,?)', ('q', 'c', 'Private prompt', 'running', stamp, '2026-09-13T20:00:00+00:00', 'owner'))
|
|
for user in ['owner', 'other']:
|
|
sub = subscription('https://web.push.apple.com/device/' + user)
|
|
db.execute('INSERT INTO push_subscriptions(endpoint,user_id,subscription,created_at) VALUES(?,?,?,?)', (sub['endpoint'], user, json.dumps(sub), stamp))
|
|
finish_and_claim_next(str(db.path), 'c', 'q', 'done', None, None, result)
|
|
return db
|
|
|
|
|
|
def test_cancelled_running_task_stays_cancelled_and_next_task_is_claimed(tmp_path):
|
|
db = Database(tmp_path / 'db.sqlite')
|
|
db.initialize()
|
|
stamp = utcnow()
|
|
db.execute('INSERT INTO projects(id,name,path,created_at) VALUES(?,?,?,?)', ('p', 'Project', str(tmp_path), stamp))
|
|
db.execute('INSERT INTO conversations(id,project_id,title,status,worker_pid,created_at,updated_at) VALUES(?,?,?,?,?,?,?)', ('c', 'p', 'Task', 'running', 123, stamp, stamp))
|
|
db.execute('INSERT INTO prompt_queue(id,conversation_id,content,status,created_at,started_at) VALUES(?,?,?,?,?,?)', ('current', 'c', 'Stop', 'cancelled', stamp, stamp))
|
|
db.execute('INSERT INTO prompt_queue(id,conversation_id,content,status,created_at) VALUES(?,?,?,?,?)', ('next', 'c', 'Continue', 'queued', stamp))
|
|
|
|
claimed = finish_and_claim_next(str(db.path), 'c', 'current', 'failed', 'terminated', None)
|
|
|
|
assert claimed == 'next'
|
|
assert db.one("SELECT status,error FROM prompt_queue WHERE id='current'") == {'status': 'cancelled', 'error': None}
|
|
assert db.one("SELECT status FROM prompt_queue WHERE id='next'")['status'] == 'running'
|
|
assert db.one("SELECT status FROM conversations WHERE id='c'")['status'] == 'running'
|
|
|
|
|
|
def test_task_completion_targets_only_requester_and_encrypts(tmp_path, monkeypatch):
|
|
db = prepare(tmp_path)
|
|
rows = db.all('SELECT * FROM push_deliveries')
|
|
assert len(rows) == 1
|
|
assert rows[0]['endpoint'].endswith('/owner')
|
|
payload = json.loads(rows[0]['payload'])
|
|
assert payload['url'] == '/?project=p&conversation=c'
|
|
assert payload['title'].startswith('Project terminé • ')
|
|
assert payload['body'] == 'Le résultat final de Codex'
|
|
assert 'Private prompt' not in rows[0]['payload']
|
|
sent = []
|
|
def request(self, method, url, **kwargs):
|
|
sent.append(kwargs)
|
|
response = Response()
|
|
response.status_code = 201
|
|
return response
|
|
monkeypatch.setattr('requests.Session.request', request)
|
|
deliver_pending(db, 'https://codexpi.example.com')
|
|
assert len(sent) == 1
|
|
assert sent[0]['allow_redirects'] is False
|
|
assert sent[0]['timeout'] == 10
|
|
assert b'Task' not in sent[0]['data']
|
|
assert db.one('SELECT sent_at FROM push_deliveries')['sent_at']
|
|
deliver_pending(db, 'https://codexpi.example.com')
|
|
assert len(sent) == 1
|
|
|
|
|
|
def test_expired_subscription_is_removed(tmp_path, monkeypatch):
|
|
db = prepare(tmp_path)
|
|
response = Response()
|
|
response.status_code = 410
|
|
monkeypatch.setattr('codexpi.push.webpush', lambda **kwargs: response)
|
|
deliver_pending(db, 'https://codexpi.example.com')
|
|
assert len(db.all('SELECT * FROM push_subscriptions')) == 1
|
|
assert db.all('SELECT * FROM push_deliveries') == []
|
|
|
|
|
|
def test_failed_push_retries_without_failing_the_task(tmp_path, monkeypatch):
|
|
db = prepare(tmp_path)
|
|
def fail(**kwargs):
|
|
raise TimeoutError()
|
|
monkeypatch.setattr('codexpi.push.webpush', fail)
|
|
deliver_pending(db, 'https://codexpi.example.com')
|
|
row = db.one('SELECT * FROM push_deliveries')
|
|
assert row['attempts'] == 1 and row['next_attempt'] > time.time()
|
|
assert db.one('SELECT status FROM prompt_queue')['status'] == 'done'
|
|
|
|
|
|
def test_notification_duration_and_result_formatting():
|
|
assert format_elapsed('2026-09-13T20:00:00+00:00', '2026-09-13T20:02:05+00:00') == '2 min 05 s'
|
|
assert format_elapsed('2026-09-13T20:00:00+00:00', '2026-09-13T21:03:00+00:00') == '1 h 03 min'
|
|
assert notification_description('Ligne 1\n\nLigne 2', 'done') == 'Ligne 1 Ligne 2'
|
|
assert notification_description('', 'failed') == 'La tâche a échoué avant de produire un résultat.'
|
|
assert notification_description('', 'cancelled') == 'La tâche a été arrêtée.'
|
|
assert len(notification_description('x' * 900, 'done')) == 700
|
|
|
|
|
|
def test_service_worker_displays_push_and_opens_task():
|
|
import subprocess
|
|
from pathlib import Path
|
|
worker = Path(__file__).resolve().parents[1] / 'frontend/public/sw.js'
|
|
script = r'''
|
|
const vm=require('node:vm'),fs=require('node:fs'),assert=require('node:assert/strict');
|
|
const events={},shown=[],opened=[];
|
|
const self={location:{origin:'https://codexpi.example.com'},addEventListener:(name,handler)=>events[name]=handler,
|
|
registration:{showNotification:async(title,options)=>shown.push({title,...options})},
|
|
clients:{matchAll:async()=>[],openWindow:async(url)=>opened.push(url)}};
|
|
vm.runInNewContext(fs.readFileSync(process.argv[1],'utf8'),{self,URL});
|
|
(async()=>{
|
|
let pending;
|
|
events.push({data:{json:()=>({title:'Tâche terminée',body:'Demo',tag:'task-q',url:'/?project=p&conversation=c'})},waitUntil:p=>pending=p});
|
|
await pending;
|
|
assert.equal(shown.length,1);assert.equal(shown[0].tag,'task-q');
|
|
events.notificationclick({notification:{data:shown[0].data,close(){}},waitUntil:p=>pending=p});
|
|
await pending;assert.equal(opened[0],'https://codexpi.example.com/?project=p&conversation=c');
|
|
events.notificationclick({notification:{data:{url:'https://evil.example'},close(){}},waitUntil:p=>pending=p});
|
|
await pending;assert.equal(opened.length,1);
|
|
})().catch(error=>{console.error(error);process.exit(1)});
|
|
'''
|
|
subprocess.run(['node', '-e', script, str(worker)], check=True, capture_output=True, text=True, timeout=10)
|