import tempfile import unittest from email.message import Message from io import BytesIO from pathlib import Path from unittest.mock import patch from urllib.error import URLError import download_images class FakeResponse(BytesIO): def __init__(self, body: bytes, content_type: str): super().__init__(body) self.headers = Message() self.headers["Content-Type"] = content_type def __enter__(self): return self def __exit__(self, *_args): self.close() class FakeCollection: def __init__(self, documents): self.documents = documents self.query = None self.projection = None def find(self, query, projection, **kwargs): self.query = query self.projection = projection self.batch_size = kwargs.get("batch_size") return iter(self.documents) class FakeDatabase: def __init__(self, collection): self.collection = collection def __getitem__(self, _name): return self.collection class FakeAdmin: def __init__(self): self.pinged = False def command(self, command): self.pinged = command == "ping" class FakeMongoClient: def __init__(self, documents): self.collection = FakeCollection(documents) self.admin = FakeAdmin() self.closed = False def __getitem__(self, _name): return FakeDatabase(self.collection) def close(self): self.closed = True class DownloadImagesTests(unittest.TestCase): def test_loads_and_flattens_gallery_arrays_from_mongodb(self): client = FakeMongoClient( [ { "_id": 1, "slug": "struttura-uno", "gallery": [" https://example.com/a.jpg "], }, { "_id": 2, "name": "Struttura Dùe!", "gallery": ["https://example.com/b.png"], }, ] ) urls = download_images.load_urls_from_mongodb( lambda *_args, **_kwargs: client ) self.assertEqual( urls, ["https://example.com/a.jpg", "https://example.com/b.png"], ) self.assertTrue(client.admin.pinged) self.assertTrue(client.closed) self.assertEqual( client.collection.query, {download_images.IMAGES_FIELD: {"$exists": True}}, ) self.assertEqual( client.collection.batch_size, download_images.MONGODB_BATCH_SIZE ) def test_rejects_invalid_gallery_data_and_closes_client(self): for gallery in ("https://example.com/a.jpg", ["ok", 3], [""]): client = FakeMongoClient( [{"_id": 42, "slug": "test", "gallery": gallery}] ) with self.subTest(gallery=gallery), self.assertRaises(ValueError): download_images.load_urls_from_mongodb( lambda *_args, **_kwargs: client ) self.assertTrue(client.closed) def test_accepts_no_matching_documents(self): client = FakeMongoClient([]) self.assertEqual( download_images.load_urls_from_mongodb( lambda *_args, **_kwargs: client ), [], ) def test_uses_slug_or_name_for_document_folder(self): client = FakeMongoClient( [ {"_id": 1, "slug": "Già-Pronto", "gallery": ["https://x/a"]}, {"_id": 2, "name": "Hotel À Mare!", "gallery": ["https://x/b"]}, ] ) jobs = list( download_images.iter_image_jobs_from_mongodb( lambda *_args, **_kwargs: client ) ) self.assertEqual( [(job.folder_slug, job.image_index) for job in jobs], [("gia-pronto", 1), ("hotel-a-mare", 1)], ) def test_rejects_document_without_slug_or_name(self): client = FakeMongoClient([{"_id": 99, "gallery": ["https://x/a"]}]) with self.assertRaisesRegex(ValueError, "non contiene"): list( download_images.iter_image_jobs_from_mongodb( lambda *_args, **_kwargs: client ) ) def test_slugify_handles_spaces_accents_and_symbols(self): self.assertEqual( download_images.slugify(" Fòto del Prödótto! 2026 "), "foto-del-prodotto-2026", ) @patch("download_images.urlopen") def test_downloads_multiple_formats_with_slug(self, mocked_urlopen): mocked_urlopen.side_effect = [ FakeResponse(b"jpeg", "image/jpeg"), FakeResponse(b"png", "image/png; charset=binary"), ] with tempfile.TemporaryDirectory() as directory: output = Path(directory) downloaded, failed = download_images.download_all( ["https://example.com/a", "https://example.com/b.webp"], output, "foto-prodotto", ) self.assertFalse(failed) self.assertEqual( [path.name for path in downloaded], ["foto-prodotto-1.jpg", "foto-prodotto-2.png"], ) @patch("download_images.urlopen") def test_url_name_extension_and_content_type_fallback(self, mocked_urlopen): mocked_urlopen.side_effect = [ FakeResponse(b"webp", "image/webp"), FakeResponse(b"gif", "image/gif"), ] with tempfile.TemporaryDirectory() as directory: output = Path(directory) downloaded, failed = download_images.download_all( [ "https://example.com/catalogo/Foto%20Uno.webp?size=2", "https://example.com/render?id=7", ], output, None, ) self.assertFalse(failed) self.assertEqual( [path.name for path in downloaded], ["Foto Uno.webp", "render.gif"] ) @patch("download_images.urlopen") def test_existing_file_gets_progressive_suffix(self, mocked_urlopen): mocked_urlopen.return_value = FakeResponse(b"new", "image/jpeg") with tempfile.TemporaryDirectory() as directory: output = Path(directory) (output / "foto-1.jpg").write_bytes(b"old") (output / "foto-1_1.jpg").write_bytes(b"old") path = download_images.download_one( "https://example.com/image.jpg", 1, output, "foto" ) self.assertEqual(path.name, "foto-1_2.jpg") self.assertEqual(path.read_bytes(), b"new") @patch("download_images.urlopen") def test_parallel_downloads_reserve_unique_names(self, mocked_urlopen): mocked_urlopen.side_effect = lambda *_args, **_kwargs: FakeResponse( b"image", "image/jpeg" ) with tempfile.TemporaryDirectory() as directory: output = Path(directory) (output / "same.jpg").write_bytes(b"existing") downloaded, failed = download_images.download_all( ["https://example.com/same.jpg"] * 6, output, None, workers=4, max_pending=6, ) self.assertFalse(failed) self.assertEqual(len({path.name for path in downloaded}), 6) self.assertFalse(list(output.glob("*.part"))) @patch("download_images.urlopen") def test_jobs_are_saved_in_document_folders(self, mocked_urlopen): mocked_urlopen.side_effect = lambda *_args, **_kwargs: FakeResponse( b"image", "image/jpeg" ) jobs = [ download_images.ImageJob("https://x/a.jpg", 1, "hotel-uno"), download_images.ImageJob("https://x/b.jpg", 1, "hotel-due"), ] with tempfile.TemporaryDirectory() as directory: output = Path(directory) downloaded, failed = download_images.download_all( jobs, output, "foto", workers=2, max_pending=2 ) self.assertFalse(failed) self.assertEqual( {path.relative_to(output).as_posix() for path in downloaded}, {"hotel-uno/foto-1.jpg", "hotel-due/foto-1.jpg"}, ) @patch("download_images.urlopen") def test_failures_do_not_stop_other_downloads(self, mocked_urlopen): mocked_urlopen.side_effect = [ URLError("host irraggiungibile"), FakeResponse(b"html", "text/html"), FakeResponse(b"png", "image/png"), ] with tempfile.TemporaryDirectory() as directory: downloaded, failed = download_images.download_all( [ "https://bad.example/a.jpg", "https://example.com/not-image", "https://example.com/ok.png", "not-a-url", ], Path(directory), None, ) self.assertEqual([path.name for path in downloaded], ["ok.png"]) self.assertEqual(len(failed), 3) self.assertIn("contenuto non immagine", failed[1].reason) self.assertIn("URL non valido", failed[2].reason) if __name__ == "__main__": unittest.main()