162 lines
7.9 KiB
Python
162 lines
7.9 KiB
Python
"""Offline chat persistence tests; never call deployed services."""
|
|
import sqlite3
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
import pytest
|
|
|
|
from control.accounts import Accounts
|
|
from control.chat_store import ChatConflict, ChatStore
|
|
|
|
|
|
@pytest.fixture
|
|
def setup(tmp_path):
|
|
accounts = Accounts(str(tmp_path / 'accounts.db'))
|
|
a = accounts.create('alice', 'Alice', 'admin', 'correct-password-alice')['id']
|
|
b = accounts.create('bob', 'Bob', 'user', 'correct-password-bob')['id']
|
|
store = ChatStore(accounts)
|
|
yield accounts, store, a, b
|
|
accounts.close()
|
|
|
|
|
|
def test_metadata_ownership_revisions_and_tombstones(setup):
|
|
accounts, s, a, b = setup
|
|
assert s.db is accounts.db and s.lock is accounts.lock
|
|
w = s.create_workspace(a, 'Work', 'Be concise')
|
|
c = s.create_conversation(a, workspace_id=w['id'])
|
|
assert 'hermes_session_id' not in c
|
|
with pytest.raises(KeyError): s.get_conversation(b, c['id'])
|
|
with pytest.raises(KeyError): s.create_conversation(b, workspace_id=w['id'])
|
|
with pytest.raises(KeyError): s.delete_workspace(b, w['id'])
|
|
c = s.update_conversation(a, c['id'], c['revision'], pinned=True, title='Revised')
|
|
with pytest.raises(ChatConflict): s.update_conversation(a, c['id'], 1, title='Stale')
|
|
w = s.update_workspace(a, w['id'], 1, name='Renamed')
|
|
with pytest.raises(ChatConflict): s.update_workspace(a, w['id'], 1, name='Stale')
|
|
s.delete_workspace(a, w['id'])
|
|
assert s.get_conversation(a, c['id'])['conversation']['workspace_id'] is None
|
|
s.delete_conversation(a, c['id'])
|
|
changes = s.sync(a)['changes']
|
|
assert any(x['kind']=='workspace' and x['deleted'] for x in changes)
|
|
assert changes[-1]['deleted'] and changes[-1]['id']==c['id']
|
|
assert s.sync(b)['changes']==[]
|
|
assert s.sync(a, s.sync(a)['cursor'])['changes']==[]
|
|
|
|
|
|
def test_preferences_and_validation(setup):
|
|
_, s, a, b = setup
|
|
assert s.get_preferences(a)['revision']==0
|
|
assert s.set_preferences(a, 0, 'ocean', 'dark')['revision']==1
|
|
with pytest.raises(ChatConflict): s.set_preferences(a, 0, 'opal', 'light')
|
|
assert s.get_preferences(b)['palette']=='opal'
|
|
invalid = [lambda:s.create_conversation(a, title=''), lambda:s.create_conversation(a, temporary=1),
|
|
lambda:s.create_workspace(a, 'x', instructions=7), lambda:s.list_conversations(a, limit=True),
|
|
lambda:s.list_conversations(a, cursor='-1'), lambda:s.set_preferences(a, True, 'opal', 'system'),
|
|
lambda:s.set_preferences(a, 1, [], 'dark')]
|
|
for call in invalid:
|
|
with pytest.raises(ValueError): call()
|
|
|
|
|
|
def test_send_durable_idempotency_concurrent_claim_and_fk(setup):
|
|
accounts, s, a, b = setup
|
|
cid = s.create_conversation(a)['id']
|
|
with ThreadPoolExecutor(max_workers=8) as pool:
|
|
replies = list(pool.map(lambda _:s.send(a,cid,'hello','unique-key'), range(16)))
|
|
rid = replies[0]['run']['id']
|
|
assert len({x['run']['id'] for x in replies})==1
|
|
assert len(s.get_conversation(a,cid)['messages'])==1
|
|
with pytest.raises(ChatConflict): s.send(a,cid,'different','unique-key')
|
|
with pytest.raises(KeyError): s.send(b,cid,'hello','unique-key')
|
|
with pytest.raises(KeyError): s.send(a,cid,'hello','attachments',['fake'])
|
|
s2 = ChatStore(accounts)
|
|
assert s2.send(a,cid,'hello','unique-key')['run']['id']==rid
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
assert sum(pool.map(lambda _:s.claim_run(a,rid),range(8)))==1
|
|
assert accounts.db.execute('PRAGMA foreign_keys').fetchone()[0]==1
|
|
other = s.create_conversation(b)['id']
|
|
with pytest.raises(sqlite3.IntegrityError), accounts.db:
|
|
accounts.db.execute('UPDATE chat_conversations SET workspace_id=? WHERE id=?',('missing',other))
|
|
s.delete_conversation(a,cid)
|
|
assert accounts.db.execute('SELECT count(*) FROM chat_runs WHERE id=?',(rid,)).fetchone()[0]==0
|
|
assert accounts.db.execute('PRAGMA foreign_key_check').fetchall()==[]
|
|
|
|
|
|
def test_run_messages_events_and_terminal_states(setup):
|
|
_, s, a, b = setup
|
|
cid = s.create_conversation(a)['id']
|
|
rid = s.send(a,cid,'hello','one')['run']['id']
|
|
s.update_run(a,rid,'running')
|
|
m = s.add_message(a,cid,'assistant','','pending',rid)
|
|
s.update_message(a,m['id'],content='part',status='streaming')
|
|
e1 = s.append_event(a,rid,'message',{'content':'part'})
|
|
e2 = s.append_event(a,rid,'progress',{'label':'Working'})
|
|
assert [e1['sequence'],e2['sequence']]==[1,2]
|
|
assert s.events(a,rid,after=1)==[e2]
|
|
with pytest.raises(KeyError): s.events(b,rid)
|
|
with pytest.raises(KeyError): s.update_message(b,m['id'],content='steal')
|
|
s.update_run(a,rid,'cancelled')
|
|
with pytest.raises(ChatConflict): s.update_run(a,rid,'running')
|
|
with pytest.raises(ChatConflict): s.update_run(a,rid,'completed')
|
|
with pytest.raises(ChatConflict): s.update_message(a,m['id'],content='late',status='completed')
|
|
s.update_message(a,m['id'],status='cancelled')
|
|
with pytest.raises(ChatConflict): s.update_message(a,m['id'],content='late')
|
|
assert s.get_run(a,rid)['run']['status']=='cancelled'
|
|
|
|
|
|
def test_branch_copies_history_without_execution_or_session(setup):
|
|
_, s, a, b = setup
|
|
cid = s.create_conversation(a)['id']
|
|
u = s.add_message(a,cid,'user','original')
|
|
reply = s.add_message(a,cid,'assistant','answer')
|
|
s.add_message(a,cid,'user','later')
|
|
s.set_hermes_session(a,cid,'trusted-session')
|
|
branch = s.branch(a,cid,reply['id'])
|
|
assert s.history(a,branch['id'])==[{'role':'user','content':'original'},{'role':'assistant','content':'answer'}]
|
|
assert s.get_hermes_session(a,branch['id']) is None
|
|
assert s.get_conversation(a,branch['id'])['runs']==[]
|
|
assert all(m['run_id'] is None for m in s.get_conversation(a,branch['id'])['messages'])
|
|
edited = s.branch(a,cid,u['id'],content='replacement')
|
|
assert s.history(a,edited['id'])==[]
|
|
assert len(s.history(a,cid))==3
|
|
with pytest.raises(KeyError): s.branch(b,cid,u['id'])
|
|
with pytest.raises(ValueError): s.branch(a,cid,reply['id'],content='bad edit')
|
|
s.set_hermes_session(a,branch['id'],'branch-session')
|
|
with pytest.raises(ChatConflict): s.set_session(a,branch['id'],'branch-session',seeded=False)
|
|
with pytest.raises(ChatConflict): s.set_hermes_session(a,branch['id'],'different')
|
|
assert 'trusted-session' not in str(s.get_conversation(a,cid))
|
|
|
|
|
|
def test_pagination_search_and_revocation(setup):
|
|
accounts,s,a,b = setup
|
|
c1 = s.create_conversation(a,title='Needle')['id']
|
|
c2 = s.create_conversation(a,title='Other')['id']
|
|
s.create_conversation(b,title='Needle')
|
|
s.add_message(a,c2,'user','searchable phrase')
|
|
assert len(s.list_conversations(a,q='needle')['conversations'])==1
|
|
assert s.list_conversations(a,q='searchable')['conversations'][0]['id']==c2
|
|
first = s.list_conversations(a,limit=1)
|
|
second = s.list_conversations(a,cursor=first['next_cursor'],limit=1)
|
|
assert first['conversations'][0]['id']!=second['conversations'][0]['id']
|
|
assert second['next_cursor'] is None
|
|
s.add_message(a,c1,'user','1'); s.add_message(a,c1,'assistant','2')
|
|
page = s.get_conversation(a,c1,limit=1)
|
|
assert s.get_conversation(a,c1,cursor=page['next_cursor'],limit=1)['messages'][0]['content']=='1'
|
|
accounts.begin_delete(b)
|
|
with pytest.raises(KeyError): s.get_preferences(b)
|
|
accounts.finish_delete(b)
|
|
assert accounts.db.execute('PRAGMA foreign_key_check').fetchall()==[]
|
|
|
|
|
|
def test_disk_reopen_preserves_admission_and_events(tmp_path):
|
|
path=str(tmp_path/'persist.db')
|
|
accounts=Accounts(path)
|
|
owner=accounts.create('alice','Alice','admin','correct-password-alice')['id']
|
|
store=ChatStore(accounts)
|
|
cid=store.create_conversation(owner)['id']
|
|
admitted=store.send(owner,cid,'durable','key')
|
|
event=store.append_event(owner,admitted['run']['id'],'run',{'status':'queued'})
|
|
accounts.close()
|
|
accounts=Accounts(path)
|
|
store=ChatStore(accounts)
|
|
assert store.send(owner,cid,'durable','key')==admitted
|
|
assert store.events(owner,admitted['run']['id'])==[event]
|
|
accounts.close()
|