33 lines
1.6 KiB
Python
33 lines
1.6 KiB
Python
|
"""Durable CAS/confirmation call prefix cannot overlap native call ordinals."""
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import sys
|
||
|
|
import uuid
|
||
|
|
import pytest
|
||
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
|
||
|
|
from task_execution_journal import Binding,Cleanup,Journal,Refused
|
||
|
|
|
||
|
|
|
||
|
|
def test_next_call_continues_auth_prefix_without_replay_or_resume(tmp_path):
|
||
|
|
os.chmod(tmp_path,0o700);journal=Journal(tmp_path)
|
||
|
|
binding=Binding('fixture-approval','sha256:'+'a'*64,str(uuid.uuid4()),'b'*64)
|
||
|
|
try:
|
||
|
|
with pytest.raises(Refused):journal.next_call_ordinal(binding)
|
||
|
|
with journal.locked():
|
||
|
|
journal.reserve(binding,(Cleanup('task-token-revoke','c'*64),))
|
||
|
|
assert journal.next_call_ordinal(binding)==0
|
||
|
|
for ordinal in [0,1]:
|
||
|
|
journal.before_call(binding,ordinal,'d'*64)
|
||
|
|
with pytest.raises(Refused):journal.next_call_ordinal(binding)
|
||
|
|
journal.after_call(binding,ordinal,'d'*64,'e'*64)
|
||
|
|
assert journal.next_call_ordinal(binding)==2
|
||
|
|
journal.transition(binding,'reserved','consume-confirmed')
|
||
|
|
journal.before_call(binding,2,'f'*64)
|
||
|
|
with pytest.raises(Refused):journal.before_call(binding,0,'f'*64)
|
||
|
|
journal.after_call(binding,2,'f'*64,'e'*64)
|
||
|
|
journal.before_cleanup(binding,0,Cleanup('task-token-revoke','c'*64))
|
||
|
|
with pytest.raises(Refused):journal.next_call_ordinal(binding)
|
||
|
|
with journal.locked():
|
||
|
|
with pytest.raises(Refused):journal.next_call_ordinal(binding)
|
||
|
|
finally:journal.close()
|