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