software/tests/test_request_archive.py

1import asyncio2import json3import subprocess4from unittest.mock import AsyncMock, Mock, patch5import pytest6from src.pipeline.request_archive import RequestArchive7from src.pipeline.translator import Translator8from src.data.storage import Storage9from src.utils.retry import RetryConfig101112def make_translator(tmp_path):13    return Translator(Mock(), Mock(), Storage(tmp_path), retry_config=RetryConfig(max_attempts=1))141516def records(tmp_path):17    return [json.loads(p.read_text()) for p in sorted((tmp_path/'request-archive').glob('*.json'))]181920def test_exact_unique_private_records(tmp_path):21    archive=RequestArchive(tmp_path)22    identifier=archive.start('model', 'system\n', '日本語\n')23    archive.finish(identifier,' response\n','warning',0)24    saved=json.loads((tmp_path/(identifier+'.input.json')).read_text())25    assert saved['user_prompt']=='日本語\n'26    assert saved['user_prompt_sha256']==archive.digest('日本語\n')27    assert (tmp_path/(identifier+'.output.json')).stat().st_mode & 0o777 == 0o60028    assert archive.start('model','system','user') != identifier29    with pytest.raises(FileExistsError):30        archive.finish(identifier)313233def test_sync_success_and_error_capture(tmp_path):34    translator=make_translator(tmp_path)35    with patch('src.pipeline.translator.subprocess.run',return_value=subprocess.CompletedProcess([],0,'  text\n','warning')):36        assert translator._call_claude('sys','user')=='text'37    output=next(r for r in records(tmp_path) if 'stdout' in r)38    assert output['stdout']=='  text\n' and output['stderr']=='warning'394041@pytest.mark.parametrize('error,kind',[(subprocess.TimeoutExpired('claude',1),'timeout'),(FileNotFoundError(),'cli_not_found')])42def test_sync_failure(tmp_path,error,kind):43    translator=make_translator(tmp_path)44    with patch('src.pipeline.translator.subprocess.run',side_effect=error),pytest.raises(Exception):45        translator._call_claude('sys','user')46    assert next(r for r in records(tmp_path) if 'error' in r)['error']==kind474849@pytest.mark.asyncio50async def test_async_capture_and_timeout_cleanup(tmp_path):51    translator=make_translator(tmp_path)52    process=Mock(returncode=0,communicate=AsyncMock(return_value=(b' raw\n',b'')))53    with patch('src.pipeline.translator.asyncio.create_subprocess_exec',new=AsyncMock(return_value=process)):54        assert await translator._call_claude_async('sys','user')=='raw'55    process=Mock(returncode=None,communicate=AsyncMock(side_effect=asyncio.TimeoutError()),wait=AsyncMock())56    with patch('src.pipeline.translator.asyncio.create_subprocess_exec',new=AsyncMock(return_value=process)),pytest.raises(Exception):57        await translator._call_claude_async('sys','user')58    process.kill.assert_called_once();process.wait.assert_awaited_once()59    assert len(records(tmp_path))==4606162def test_retries_keep_both_attempts(tmp_path):63    translator=make_translator(tmp_path)64    translator._retry_config=RetryConfig(max_attempts=2,base_delay=0,jitter=False)65    responses=[subprocess.CompletedProcess([],1,'','rate limit exceeded'),subprocess.CompletedProcess([],0,'success','')]66    with patch('src.pipeline.translator.subprocess.run',side_effect=responses):67        assert translator._call_claude('system','user')=='success'68    saved=records(tmp_path)69    assert len(saved)==470    inputs={row['id'] for row in saved if 'system_prompt' in row}71    outputs={row['id'] for row in saved if 'stdout' in row}72    assert inputs==outputs and len(inputs)==273    assert sorted(row['returncode'] for row in saved if 'stdout' in row)==[0,1]74