redmine/tests/test_cache_concurrency.py

"""Real process tests: no placeholder success assertions or live API access."""
import json
import os
from pathlib import Path
import subprocess
import sys
import time
import pytest
from redmine_cache_io import CacheError, atomic_write_json, cache_lock

ROOT=Path(__file__).resolve().parents[1]

def spawn(code,*args,extra=None):
    return subprocess.Popen([sys.executable,'-c',code,*map(str,args)],cwd=ROOT,
        env={**os.environ,**(extra or {})},stdout=subprocess.PIPE,stderr=subprocess.PIPE,text=True)


def finish(proc):
    out,err=proc.communicate(timeout=15)
    assert proc.returncode==0,err
    return out,err


def test_distinct_ticket_updates_from_stale_instances_survive(tmp_path):
    code='''from redmine_tickets import TicketCache
from pathlib import Path
import sys,time
c=TicketCache(Path(sys.argv[1]),'customer-service')
print('ready',flush=True)
time.sleep(.2)
for n in range(8): c.write_issue({'id':int(sys.argv[2])*100+n,'subject':'fixture','updated_on':str(n)})
'''
    children=[spawn(code,tmp_path,n) for n in range(3)]
    for child in children:
        assert child.stdout.readline().strip()=='ready'
    for child in children: finish(child)
    data=json.loads((tmp_path/'manifest.json').read_text())
    assert len(data['tickets'])==24
    assert len(list((tmp_path/'customer-service').glob('*.json')))==24


def test_distinct_contact_updates_survive(tmp_path):
    code='''from redmine_ticket_contacts import ContactCache
from pathlib import Path
import sys,time
c=ContactCache(Path(sys.argv[1]))
for n in range(10): c.write_contact({'id':int(sys.argv[2])*100+n,'first_name':'fixture'})
'''
    children=[spawn(code,tmp_path,n) for n in range(3)]
    for child in children: finish(child)
    data=json.loads((tmp_path/'contacts/manifest.json').read_text())
    assert len(data['contacts'])==30


def test_readers_see_complete_json_during_writes(tmp_path):
    path=tmp_path/'data.json'
    atomic_write_json(path,{'n':0,'blob':'x'*10000})
    child=spawn('''from redmine_cache_io import atomic_write_json
from pathlib import Path
import sys
for n in range(60): atomic_write_json(Path(sys.argv[1]),{'n':n,'blob':'x'*10000})
''',path)
    reads=0
    while child.poll() is None:
        data=json.loads(path.read_text())
        assert len(data['blob'])==10000
        reads+=1
    finish(child)
    assert reads>0


def holder(path):
    child=spawn('''from redmine_cache_io import cache_lock
from pathlib import Path
import sys,time
with cache_lock(Path(sys.argv[1])):
 print('locked',flush=True)
 time.sleep(20)
''',path)
    assert child.stdout.readline().strip()=='locked'
    return child


def test_contention_timeout_is_visible_and_nonzero(tmp_path):
    path=tmp_path/'manifest.json'; child=holder(path)
    try:
        waiter=spawn('''from redmine_cache_io import atomic_write_json
from pathlib import Path
import sys
atomic_write_json(Path(sys.argv[1]),{'x':1})
''',path,extra={'REDMINE_LOCK_TIMEOUT':'.12'})
        out,err=waiter.communicate(timeout=5)
        assert waiter.returncode!=0
        assert 'Waiting for cache lock' in err
        assert 'timeout after 0.12s' in err
    finally:
        child.kill();child.communicate(timeout=5)


def test_sigkill_releases_existing_lock_file(tmp_path):
    path=tmp_path/'manifest.json'; child=holder(path)
    child.kill(); child.communicate(timeout=5)
    assert path.with_name(path.name+'.lock').exists()
    waiter=spawn('''from redmine_cache_io import atomic_write_json
from pathlib import Path
import sys
atomic_write_json(Path(sys.argv[1]),{'recovered':True})
''',path,extra={'REDMINE_LOCK_TIMEOUT':'.2'})
    finish(waiter)
    assert json.loads(path.read_text())=={'recovered':True}


@pytest.mark.parametrize('value',['bogus','-1','nan','inf'])
def test_invalid_timeout_is_actionable(tmp_path,monkeypatch,value):
    monkeypatch.setenv('REDMINE_LOCK_TIMEOUT',value)
    with pytest.raises(CacheError,match='REDMINE_LOCK_TIMEOUT'):
        atomic_write_json(tmp_path/'x.json',{})


def test_stale_temporary_file_is_removed_on_next_lock(tmp_path):
    path=tmp_path/'manifest.json'
    stale=tmp_path/'.manifest.json.killed.tmp'
    stale.write_text('partial')
    atomic_write_json(path,{'recovered':True})
    assert not stale.exists()
    assert json.loads(path.read_text())=={'recovered':True}


def test_nested_lock_and_atomic_failure_preserves_original(tmp_path):
    path=tmp_path/'data.json'
    atomic_write_json(path,{'old':True})
    with cache_lock(path,timeout=.1):
        with pytest.raises(TypeError): atomic_write_json(path,{'bad':object()})
    assert json.loads(path.read_text())=={'old':True}
    assert not list(tmp_path.glob('*.tmp'))


def test_corrupt_contact_manifest_not_reset(tmp_path):
    from redmine_ticket_contacts import ContactCache
    c=ContactCache(tmp_path)
    c.manifest_path.write_text('{broken')
    with pytest.raises(CacheError): c.write_contact({'id':1})
    assert c.manifest_path.read_text()=='{broken'


def test_snapshot_fetches_are_serialized_before_replacement(tmp_path):
    code='''from redmine_contacts import snapshot_transaction,write_cache
from argparse import Namespace
from pathlib import Path
import sys,time
args=Namespace(cache=Path(sys.argv[1]),base_url='https://test.example',project='p',refresh=True)
@snapshot_transaction
def fetch(args):
 print('fetch',flush=True)
 time.sleep(float(sys.argv[2]))
 write_cache(args.cache,args,[{'id':int(sys.argv[3])}])
fetch(args)
'''
    path=tmp_path/'contacts.json'
    first=spawn(code,path,.3,1)
    assert first.stdout.readline().strip()=='fetch'
    second=spawn(code,path,0,2)
    finish(first); _,err=finish(second)
    assert 'Waiting for cache lock' in err
    assert json.loads(path.read_text())['contacts']==[{'id':2}]