Lavoro della sessione precedente: colonna sonora (titolo, paese, ricordi) con muto e volume, vecchietta e vecchietto sulla panchina, scene di Pieve, oratorio, Via de' Medici, Villa Adami, salita, campeggio, Bosco ai Frati e viale dei cipressi, panorama dell'introduzione, tools/genera.py per gli sfondi dalle foto. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UdeKd62opy9PKgpQXtVctF
180 lines
7.9 KiB
Python
180 lines
7.9 KiB
Python
"""Offline tests: every API interaction uses a fake client."""
|
|
import base64
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
from PIL import Image
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'tools'))
|
|
import genera
|
|
|
|
|
|
class ConnectionErrorFake(Exception):
|
|
pass
|
|
|
|
|
|
class StatusErrorFake(Exception):
|
|
def __init__(self, status_code):
|
|
self.status_code = status_code
|
|
|
|
|
|
def png(size=genera.SIZE):
|
|
stream = io.BytesIO()
|
|
Image.new('RGB', size, '#ad8c63').save(stream, format='PNG')
|
|
return stream.getvalue()
|
|
|
|
|
|
class GeneratorTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.input = self.root / 'foto'
|
|
self.output = self.root / 'generate'
|
|
self.prompts = self.root / 'prompt'
|
|
self.input.mkdir()
|
|
self.prompts.mkdir()
|
|
self.photo('a.jpg')
|
|
self.photo('b.WEBP')
|
|
(self.input / 'da-fare').mkdir()
|
|
self.photo('da-fare/oratorio.png')
|
|
self.response = SimpleNamespace(data=[SimpleNamespace(b64_json=base64.b64encode(png()).decode())],
|
|
_request_id='fake-request', usage=None)
|
|
self.client = Mock()
|
|
self.client.images.edit.return_value = self.response
|
|
self.sdk = SimpleNamespace(OpenAI=Mock(return_value=self.client),
|
|
APIConnectionError=ConnectionErrorFake, APIStatusError=StatusErrorFake)
|
|
|
|
def photo(self, name):
|
|
Image.new('RGB', (48, 36), 'brown').save(self.input / name)
|
|
|
|
def run_cli(self, *args, answer='GENERA', eof=False):
|
|
with patch.dict(sys.modules, {'openai': self.sdk}), \
|
|
patch.dict('os.environ', {'OPENAI_API_KEY': 'offline-test-only'}), \
|
|
patch('builtins.input', return_value=answer, side_effect=EOFError if eof else None), \
|
|
redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
|
|
return genera.main(['--input-dir', str(self.input), '--output-dir', str(self.output),
|
|
'--prompt-dir', str(self.prompts), *args])
|
|
|
|
def records(self):
|
|
return json.loads((self.output / 'registro.json').read_text())
|
|
|
|
def test_dry_run_does_not_write_or_instantiate_sdk(self):
|
|
self.assertEqual(self.run_cli('--dry-run'), 0)
|
|
self.assertFalse(self.output.exists())
|
|
self.sdk.OpenAI.assert_not_called()
|
|
self.assertEqual([p.name for p in genera.photos_in(self.input)], ['a.jpg', 'b.WEBP'])
|
|
|
|
def test_confirmation_declined_or_eof_never_calls_api(self):
|
|
self.assertEqual(self.run_cli(answer='no'), 0)
|
|
self.sdk.OpenAI.assert_not_called()
|
|
self.assertFalse(self.output.exists())
|
|
self.assertEqual(self.run_cli(eof=True), 0)
|
|
self.client.images.edit.assert_not_called()
|
|
|
|
def test_missing_key_blocks_before_client(self):
|
|
args = SimpleNamespace(input_dir=self.input, output_dir=self.output, prompt_dir=self.prompts,
|
|
gallery_only=False, dry_run=False, test=False, limit=None,
|
|
model='gpt-image-2', quality='high')
|
|
with patch.dict('os.environ', {}, clear=True), redirect_stdout(io.StringIO()):
|
|
with self.assertRaisesRegex(ValueError, 'OPENAI_API_KEY'):
|
|
genera.run(args)
|
|
self.sdk.OpenAI.assert_not_called()
|
|
|
|
def test_generate_then_resume_and_limit(self):
|
|
(self.prompts / 'a.txt').write_text('Preserve the lion.')
|
|
self.assertEqual(self.run_cli('--test'), 0)
|
|
self.assertEqual(self.client.images.edit.call_count, 1)
|
|
kwargs = self.client.images.edit.call_args.kwargs
|
|
self.assertEqual(kwargs['model'], 'gpt-image-2')
|
|
self.assertEqual(kwargs['size'], '1536x864')
|
|
self.assertIn('Preserve the lion.', kwargs['prompt'])
|
|
self.assertNotIn('input_fidelity', kwargs)
|
|
self.assertEqual(self.records()['a.jpg']['status'], 'completed')
|
|
self.assertEqual(self.run_cli('--limit', '1'), 0)
|
|
self.assertEqual(self.client.images.edit.call_count, 2)
|
|
self.assertEqual(self.run_cli(), 0)
|
|
self.assertEqual(self.client.images.edit.call_count, 2)
|
|
genera.validate_image(self.output / 'a.png', genera.SIZE)
|
|
genera.validate_image(self.output / 'b.png', genera.SIZE)
|
|
|
|
def test_completed_file_skipped_without_log(self):
|
|
self.output.mkdir()
|
|
(self.output / 'a.png').write_bytes(png())
|
|
self.assertEqual(self.run_cli('--test'), 0)
|
|
self.assertEqual(self.client.images.edit.call_count, 1)
|
|
self.assertIn('b.WEBP', self.records())
|
|
self.assertNotIn('a.jpg', self.records())
|
|
|
|
def test_corrupt_existing_output_blocks_paid_requests(self):
|
|
self.output.mkdir()
|
|
(self.output / 'a.png').write_bytes(b'not-an-image')
|
|
self.assertEqual(self.run_cli(), 1)
|
|
self.client.images.edit.assert_not_called()
|
|
self.assertEqual((self.output / 'a.png').read_bytes(), b'not-an-image')
|
|
|
|
def test_cached_response_requires_no_api_or_confirmation(self):
|
|
cache = self.output / '.responses'
|
|
cache.mkdir(parents=True)
|
|
(cache / 'a.png').write_bytes(png())
|
|
self.assertEqual(self.run_cli('--test', answer='no'), 0)
|
|
self.sdk.OpenAI.assert_not_called()
|
|
genera.validate_image(self.output / 'a.png', genera.SIZE)
|
|
|
|
def test_duplicate_stems_rejected_before_api(self):
|
|
self.photo('a.PNG')
|
|
self.assertEqual(self.run_cli('--dry-run'), 1)
|
|
self.sdk.OpenAI.assert_not_called()
|
|
|
|
def test_transient_errors_retry_with_bounded_attempts(self):
|
|
self.client.images.edit.side_effect = [ConnectionErrorFake(), StatusErrorFake(429), self.response]
|
|
with patch('genera.time.sleep') as sleep:
|
|
self.assertEqual(self.run_cli('--test'), 0)
|
|
self.assertEqual(sleep.call_count, 2)
|
|
self.assertEqual(self.records()['a.jpg']['attempts'], 3)
|
|
self.sdk.OpenAI.assert_called_once_with(timeout=180, max_retries=0)
|
|
|
|
def test_exhausted_timeouts_are_logged(self):
|
|
self.client.images.edit.side_effect = ConnectionErrorFake()
|
|
with patch('genera.time.sleep'):
|
|
self.assertEqual(self.run_cli('--test', '--attempts', '2'), 1)
|
|
self.assertEqual(self.client.images.edit.call_count, 2)
|
|
self.assertEqual(self.records()['a.jpg']['status'], 'error')
|
|
|
|
def test_permanent_error_not_retried_and_next_photo_processed(self):
|
|
self.client.images.edit.side_effect = [StatusErrorFake(400), self.response]
|
|
with patch('genera.time.sleep') as sleep:
|
|
self.assertEqual(self.run_cli(), 1)
|
|
sleep.assert_not_called()
|
|
self.assertEqual(self.client.images.edit.call_count, 2)
|
|
self.assertEqual(self.records()['a.jpg']['status'], 'error')
|
|
self.assertEqual(self.records()['b.WEBP']['status'], 'completed')
|
|
|
|
def test_bad_aspect_ratio_preserved_without_repeated_charge(self):
|
|
self.response.data[0].b64_json = base64.b64encode(png((100, 100))).decode()
|
|
self.assertEqual(self.run_cli('--test'), 1)
|
|
self.assertFalse((self.output / 'a.png').exists())
|
|
self.assertTrue((self.output / '.responses/a.png').exists())
|
|
self.assertEqual(self.run_cli('--test'), 1)
|
|
self.assertEqual(self.client.images.edit.call_count, 1)
|
|
|
|
def test_gallery_escapes_names_and_encodes_links(self):
|
|
self.photo('chiesa & "piazza".png')
|
|
self.assertEqual(self.run_cli('--gallery-only'), 0)
|
|
page = (self.output / 'index.html').read_text()
|
|
self.assertIn('chiesa & "piazza"', page)
|
|
self.assertIn('chiesa%20%26%20%22piazza%22.png', page)
|
|
self.assertIn('Non ancora generata', page)
|
|
self.sdk.OpenAI.assert_not_called()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|