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