import threading import unittest from unittest.mock import MagicMock, patch from pathlib import Path from config import DB_HOST, DB_POOL_SIZE from database import ( db_pool, delete_account, delete_video, get_account_by_id, get_accounts, get_all_videos_with_details, get_media_item, get_saved_series_list, get_series_episodes_from_db, get_video_by_id, init_db, save_account, save_media_item, save_video, update_account_status, update_video_progress, ) from converter import EmbedConverter from mozisarok_client import MozisarokClient from providers import get_provider, list_available_providers from queue_manager import UploadQueueManager from stream_uploader import DiskStreamingMultipart, FastHTTPAdapter, UploadCancelledException, create_fast_session class TestMozisarokUploader(unittest.TestCase): @classmethod def setUpClass(cls): init_db() def test_database_connection_and_pool(self): """Test MySQL connection pool and ping.""" conn = db_pool.get_connection() self.assertIsNotNone(conn) self.assertTrue(conn.open) db_pool.release_connection(conn) def test_account_crud(self): """Test Account creation, retrieval, and deletion in MySQL.""" acc_id = save_account( provider="videa", name="Test Videa Account Unit", username="test_videa@example.com", password="secretpassword", ) self.assertGreater(acc_id, 0) acc = get_account_by_id(acc_id) self.assertIsNotNone(acc) self.assertEqual(acc["name"], "Test Videa Account Unit") self.assertEqual(acc["provider"], "videa") update_account_status(acc_id, 0) acc_updated = get_account_by_id(acc_id) self.assertEqual(acc_updated["is_active"], 0) delete_account(acc_id) self.assertIsNone(get_account_by_id(acc_id)) def test_media_and_video_lifecycle(self): """Test media datasheet and video recording in MySQL.""" media_id = save_media_item( tmdb_id=9999999, media_type="movie", title="Unit Test Movie", overview="Test overview", release_date="2026-01-01", ) self.assertGreater(media_id, 0) media = get_media_item(9999999, "movie") self.assertIsNotNone(media) self.assertEqual(media["title"], "Unit Test Movie") # Save video vid_id = save_video( media_id=media_id, account_id=None, file_path="D:/dummy/test.mp4", file_name="test.mp4", file_size=1024000, provider="videa", quality="1080p", status="queued", ) self.assertGreater(vid_id, 0) # Update progress and embed link update_video_progress( vid_id, status="completed", provider_video_id="vcode123", direct_url="https://videa.hu/videok/test-vcode123", embed_url="https://videa.hu/player?f=test", mozisarok_submitted=1, mozisarok_verified=1, completed=True, ) videos = get_all_videos_with_details() found = next((v for v in videos if v["id"] == vid_id), None) self.assertIsNotNone(found) self.assertEqual(found["provider_video_id"], "vcode123") self.assertEqual(found["mozisarok_submitted"], 1) self.assertEqual(found["mozisarok_verified"], 1) # Cleanup delete_video(vid_id) def test_provider_factory(self): """Test provider instantiation.""" providers = list_available_providers() self.assertIn("videa", providers) self.assertIn("vk", providers) videa_prov = get_provider("videa") self.assertEqual(videa_prov.name, "videa") vk_prov = get_provider("vk") self.assertEqual(vk_prov.name, "vk") def test_embed_converter_service(self): """Test converter with live Lune's host Video Scraper service.""" converter = EmbedConverter() real_videa_url = "https://videa.hu/videok/haon/hajdu-bihar-egyik-legolcsobb-telepulesen-zBzvbUqp6WKKGsCm" embed = converter.convert_url(real_videa_url) self.assertIsNotNone(embed) self.assertIn("videa.hu", embed) self.assertIn("player?f=", embed) def test_mozisarok_search(self): """Test Mozisarok API search.""" client = MozisarokClient() results = client.search("avatar", limit=2) self.assertIsInstance(results, list) self.assertGreater(len(results), 0) first = results[0] self.assertTrue("id" in first or "tmdb_id" in first) def test_disk_streaming_multipart(self): """Test that DiskStreamingMultipart streams chunks directly from disk without loading file into RAM.""" temp_file = Path("test_stream_temp.mp4") payload_data = b"STREAMING_TEST_BINARY_DATA" * 500 # ~13 KB temp_file.write_bytes(payload_data) progresses = [] def on_prog(pct): progresses.append(pct) try: stream = DiskStreamingMultipart( fields={"token": "abc123token", "type": "video"}, file_field_name="uploadfile", file_path=temp_file, filename="test_stream_temp.mp4", mime_type="video/mp4", chunk_size=1024, progress_callback=on_prog ) total_len = len(stream) self.assertGreater(total_len, len(payload_data)) self.assertIn("multipart/form-data", stream.content_type) # Read all chunks in tiny 1KB buffers chunks = [] while True: chunk = stream.read(1024) if not chunk: break chunks.append(chunk) full_body = b"".join(chunks) self.assertEqual(len(full_body), total_len) self.assertIn(b"abc123token", full_body) self.assertIn(payload_data, full_body) self.assertTrue(len(progresses) > 0) self.assertAlmostEqual(progresses[-1], 90.0, delta=1.0) stream.close() finally: if temp_file.exists(): temp_file.unlink() def test_queue_manager_add_and_concurrency(self): """Test adding items to queue and setting concurrency.""" qm = UploadQueueManager(videa_concurrency=3, auto_recover=False) self.assertEqual(qm.videa_concurrency, 3) qm.set_videa_concurrency(5) self.assertEqual(qm.videa_concurrency, 5) # Create a temp file to simulate video file temp_file = Path("test_dummy_video.mp4") temp_file.write_bytes(b"0" * 1024) try: task_id = qm.add_task( media_data={"id": 19995, "media_type": "movie", "title": "Avatar"}, file_path=str(temp_file.resolve()), provider="videa", ) self.assertGreater(task_id, 0) items = qm.get_queue_status() self.assertEqual(len(items), 1) self.assertEqual(items[0]["id"], task_id) # Remove task removed = qm.remove_task(task_id) self.assertTrue(removed) self.assertEqual(len(qm.get_queue_status()), 0) # Cleanup DB record delete_video(task_id) finally: if temp_file.exists(): temp_file.unlink() def test_upload_cancellation_stream(self): """Test that DiskStreamingMultipart aborts immediately when cancel_event is signaled.""" temp_file = Path("test_cancel_stream.mp4") temp_file.write_bytes(b"A" * (2 * 1024 * 1024)) # 2 MB cancel_event = threading.Event() try: stream = DiskStreamingMultipart( fields={"token": "test"}, file_field_name="video", file_path=temp_file, chunk_size=64 * 1024, cancel_event=cancel_event, ) # Read first chunk normally chunk1 = stream.read(64 * 1024) self.assertTrue(len(chunk1) > 0) # Signal cancellation cancel_event.set() # Next read must raise UploadCancelledException with self.assertRaises(UploadCancelledException): stream.read(64 * 1024) # Verify file handle is closed self.assertTrue(stream._file_handle.closed) finally: if temp_file.exists(): temp_file.unlink() def test_fast_http_adapter_buffers(self): """Test that FastHTTPAdapter and create_fast_session configure multi-gigabit send buffers.""" adapter = FastHTTPAdapter(blocksize=1024 * 1024, sndbuf=4 * 1024 * 1024) self.assertEqual(adapter.custom_blocksize, 1048576) self.assertEqual(adapter.custom_sndbuf, 4194304) session = create_fast_session(blocksize_mb=1, sndbuf_mb=4) self.assertIn("https://", session.adapters) self.assertIn("http://", session.adapters) self.assertIsInstance(session.adapters["https://"], FastHTTPAdapter) def test_queue_task_cancellation(self): """Test canceling an in-queue task marks DB status and aborts.""" qm = UploadQueueManager(auto_recover=False) temp_file = Path("test_queue_cancel.mp4") temp_file.write_bytes(b"DATA" * 256) try: task_id = qm.add_task( media_data={"id": 12345, "media_type": "movie", "title": "Test Cancel"}, file_path=str(temp_file.resolve()), provider="videa", ) # Cancel task success = qm.cancel_task(task_id) self.assertTrue(success) status = qm.get_queue_status() task_entry = next((t for t in status if t["id"] == task_id), None) self.assertIsNotNone(task_entry) self.assertEqual(task_entry["status"], "cancelled") # Remove task removed = qm.remove_task(task_id) self.assertTrue(removed) self.assertEqual(len(qm.get_queue_status()), 0) # Check DB record (status is stored in Hungarian, status_code in English) db_video = get_video_by_id(task_id) self.assertIsNotNone(db_video) self.assertEqual(db_video["status"], "megszakítva") self.assertEqual(db_video.get("status_code"), "cancelled") delete_video(task_id) finally: if temp_file.exists(): temp_file.unlink() def test_videa_wait_for_conversion_mock(self): """Test Videa conversion polling logic and completion detection.""" from unittest.mock import MagicMock, patch from providers.videa import VideaProvider videa_prov = VideaProvider() mock_session = MagicMock() # Simulate 2 polling iterations: first in-progress (50%), then finished mock_session.post.side_effect = [ MagicMock( status_code=200, json=lambda: { "response": { "code": 0, "progresses": [ { "id": "test_vcode_123", "server_time": 1000, "start_time": 900, "end_time": 1100, "finished": False, } ], } }, ), MagicMock( status_code=200, json=lambda: { "response": { "code": 0, "progresses": [ { "id": "test_vcode_123", "server_time": 1100, "start_time": 900, "end_time": 1100, "finished": True, } ], } }, ), ] recorded_progresses = [] def on_prog(pct, msg): recorded_progresses.append((pct, msg)) finished = videa_prov.wait_for_conversion( account={"name": "test"}, vcode="test_vcode_123", progress_callback=on_prog, poll_interval=0.1, session=mock_session, ) self.assertTrue(finished) self.assertGreater(len(recorded_progresses), 0) self.assertEqual(recorded_progresses[0][0], 50.0) self.assertEqual(recorded_progresses[-1][0], 100.0) def test_episode_parser_formats(self): """Test series Season and Episode automatic extraction across various formats.""" from episode_parser import extract_season_episode test_cases = [ ("s01e01.mp4", 1, 1), ("s1e1.mkv", 1, 1), ("s1e01.avi", 1, 1), ("Series.Title.S02E08.1080p.WEB-DL.mkv", 2, 8), ("Breaking.Bad.2x04.HDTV.mp4", 2, 4), ("Dark.Season.3.Episode.5.mp4", 3, 5), ("1.evad 03.resz.mp4", 1, 3), ("Sorozat 2. évad 4. rész.mkv", 2, 4), ("ep05.mp4", 1, 5), ("movie_without_episode.mp4", None, None), ] for filename, expected_s, expected_e in test_cases: s, e = extract_season_episode(filename) self.assertEqual((s, e), (expected_s, expected_e), f"Failed for {filename}: got {s}, {e}") # Directory structure test dir_path = "D:/Series/Stranger Things/Season 4/07 - The Massacre.mkv" s, e = extract_season_episode(dir_path) self.assertEqual(s, 4) self.assertEqual(e, 7) def test_mozisarok_quality_mapping(self): """Verify Mozisarok quality mapping: 1080p MUST be 4 (FHD) and not 1 (CAM).""" client = MozisarokClient(token="dummy_token") self.assertEqual(client.quality_map.get("1080p"), 4, "1080p must map to 4 (Full HD)") self.assertEqual(client.quality_map.get("fhd"), 4, "fhd must map to 4") self.assertEqual(client.quality_map.get("cam"), 1, "cam must map to 1 (CAM)") self.assertEqual(client.quality_map.get("720p"), 3, "720p must map to 3 (HD)") self.assertEqual(client.quality_map.get("4k"), 5, "4k must map to 5 (UHD)") def test_database_hungarian_statuses(self): """Test Hungarian status persistence in MySQL.""" from database import STATUS_TO_HUNGARIAN, HUNGARIAN_TO_STATUS self.assertEqual(STATUS_TO_HUNGARIAN.get("uploaded"), "feltöltve") self.assertEqual(STATUS_TO_HUNGARIAN.get("queued"), "várakozik") self.assertEqual(STATUS_TO_HUNGARIAN.get("converting"), "konvertálás alatt") self.assertEqual(STATUS_TO_HUNGARIAN.get("completed"), "kész") vid_id = save_video( media_id=1, account_id=None, file_path="D:/dummy/status_test.mp4", file_name="status_test.mp4", file_size=1024, provider="videa", quality="1080p", status="uploaded", ) self.assertGreater(vid_id, 0) v = get_video_by_id(vid_id) self.assertEqual(v["status"], "feltöltve") update_video_progress(vid_id, status="converting") v2 = get_video_by_id(vid_id) self.assertEqual(v2["status"], "konvertálás alatt") delete_video(vid_id) def test_crypto_encryption_and_decryption(self): """Test encryption, decryption, idempotency, and key derivation.""" from crypto import decrypt_value, encrypt_value, is_encrypted raw_pwd = "MySuperSecretPassword123!" encrypted = encrypt_value(raw_pwd) self.assertNotEqual(encrypted, raw_pwd) self.assertTrue(encrypted.startswith("gAAAAA")) self.assertTrue(is_encrypted(encrypted)) # Idempotency: encrypting an already encrypted token does NOT double-encrypt double_enc = encrypt_value(encrypted) self.assertEqual(double_enc, encrypted) # Decryption returns original plaintext decrypted = decrypt_value(encrypted) self.assertEqual(decrypted, raw_pwd) # Legacy unencrypted plaintext is returned as-is self.assertEqual(decrypt_value("legacy_plaintext"), "legacy_plaintext") # Custom passphrase derivation custom_key = "MyCustomPassphrase2026!" custom_enc = encrypt_value("CustomSecret", custom_key=custom_key) self.assertTrue(custom_enc.startswith("gAAAAA")) self.assertEqual(decrypt_value(custom_enc, custom_key=custom_key), "CustomSecret") def test_database_credentials_encrypted_in_db(self): """Test that passwords and tokens are stored encrypted in MySQL and decrypted on retrieval.""" from crypto import is_encrypted from database import ( db_pool, delete_account, get_account_by_id, get_setting, migrate_encrypt_existing_credentials, save_account, set_setting, ) test_pwd = "TestEncryptedPassword_XYZ_987" test_tok = "TestEncryptedToken_ABC_123" acc_id = save_account( provider="videa", name="Crypto Verification Account", username="crypto_test@example.com", password=test_pwd, token=test_tok, ) self.assertGreater(acc_id, 0) orig_token = get_setting("mozisarok_token") try: # 1. Verify directly in raw MySQL that the columns are encrypted with db_pool.cursor() as cur: cur.execute("SELECT password, token FROM accounts WHERE id = %s", (acc_id,)) raw_row = cur.fetchone() self.assertIsNotNone(raw_row) self.assertNotEqual(raw_row["password"], test_pwd) self.assertTrue(raw_row["password"].startswith("gAAAAA")) self.assertTrue(is_encrypted(raw_row["password"])) self.assertNotEqual(raw_row["token"], test_tok) self.assertTrue(raw_row["token"].startswith("gAAAAA")) self.assertTrue(is_encrypted(raw_row["token"])) # 2. Verify get_account_by_id transparently decrypts acc = get_account_by_id(acc_id) self.assertEqual(acc["password"], test_pwd) self.assertEqual(acc["token"], test_tok) # 3. Test sensitive setting encryption set_setting("mozisarok_token", "eyJhbGciOi_TestToken_12345") with db_pool.cursor() as cur: cur.execute("SELECT `value` FROM app_settings WHERE `key` = %s", ("mozisarok_token",)) raw_s = cur.fetchone() self.assertTrue(raw_s["value"].startswith("gAAAAA")) # Transparent retrieval dec_s = get_setting("mozisarok_token") self.assertEqual(dec_s, "eyJhbGciOi_TestToken_12345") # 4. Migration is idempotent mig_res = migrate_encrypt_existing_credentials() self.assertIsInstance(mig_res, dict) finally: if orig_token is not None: set_setting("mozisarok_token", orig_token) delete_account(acc_id) def test_system_recovery_uploading_not_auto_resumed(self): """ Verify that a video left in 'feltöltés alatt' (uploading) before a crash/shutdown is NOT resumed automatically, but reset to 'várakozik' with error message and paused in queue. """ from database import delete_video, get_video_by_id, save_video from queue_manager import UploadQueueManager # Create a video in DB simulating an interrupted upload during crash vid_id = save_video( media_id=1, account_id=None, file_path="D:/dummy/interrupted_upload.mp4", file_name="interrupted_upload.mp4", file_size=5000000, provider="videa", quality="1080p", status="uploading", # feltöltés alatt ) self.assertGreater(vid_id, 0) try: # Instantiate queue manager with auto_recover=False to test recovery method directly qm = UploadQueueManager(auto_recover=False) summary = qm.recover_from_database(run_in_background=False) self.assertGreaterEqual(summary["uploading_paused"], 1) self.assertFalse(qm.is_running, "Queue must NOT be running automatically after crash recovery!") # Verify in DB db_v = get_video_by_id(vid_id) self.assertEqual(db_v["status"], "várakozik", "Interrupted upload must be reset to 'várakozik'!") self.assertIn("félbeszakadt", (db_v.get("error_message") or "").lower()) # Verify in queue memory q_task = next((t for t in qm.get_queue_status() if t["id"] == vid_id), None) self.assertIsNotNone(q_task) self.assertEqual(q_task["status"], "queued") finally: delete_video(vid_id) def test_system_recovery_verifying_completed(self): """ Verify that a video left in 'ellenőrzés alatt' (verifying) is checked to see if it is uploaded and live, and then completed in MySQL. """ from unittest.mock import MagicMock, patch from database import delete_video, get_video_by_id, save_video, update_video_progress from queue_manager import UploadQueueManager vid_id = save_video( media_id=1, account_id=None, file_path="D:/dummy/verifying_test.mp4", file_name="verifying_test.mp4", file_size=10000, provider="videa", quality="1080p", status="verifying", # ellenőrzés alatt ) self.assertGreater(vid_id, 0) update_video_progress( vid_id, status="verifying", embed_url="https://videa.hu/player?v=valid_recovery_test", mozisarok_submitted=1, ) try: qm = UploadQueueManager(auto_recover=False) # Mock Mozisarok verify_video to simulate that the video is uploaded and reachable qm.mozisarok.verify_video = MagicMock( return_value={ "mozisarok_found": True, "url_reachable": True, "status_code": 200, "message": "OK", } ) summary = qm.recover_from_database(run_in_background=False) self.assertGreaterEqual(summary["verifying_completed"], 1) # Verify in DB db_v = get_video_by_id(vid_id) self.assertEqual(db_v["status"], "kész", "Verifying video must be marked as 'kész'!") self.assertEqual(db_v["mozisarok_verified"], 1) finally: delete_video(vid_id) def test_force_check_video_advancement(self): """ Verify that manual forced check (force_check_video) properly advances a video stuck at 0% in 'konvertálás alatt' to 'kész' with scraped embed link. """ from unittest.mock import MagicMock, patch from database import delete_video, get_video_by_id, save_video, update_video_progress from queue_manager import UploadQueueManager vid_id = save_video( media_id=1, account_id=None, file_path="D:/dummy/force_check_test.mp4", file_name="force_check_test.mp4", file_size=20000, provider="videa", quality="1080p", status="converting", # konvertálás alatt ) self.assertGreater(vid_id, 0) update_video_progress( vid_id, status="converting", provider_video_id="vcode_stuck_test", direct_url="https://videa.hu/videok/vcode_stuck_test", ) try: qm = UploadQueueManager(auto_recover=False) # Mock Video Scraper conversion qm.converter.convert_url = MagicMock(return_value="https://videa.hu/player?f=clean_scraped_embed_123") # Mock Mozisarok submission & verification qm.mozisarok.submit_link = MagicMock(return_value={"success": True}) qm.mozisarok.verify_video = MagicMock( return_value={"mozisarok_found": True, "url_reachable": True, "status_code": 200} ) res = qm.force_check_video(vid_id) self.assertTrue(res["success"]) self.assertEqual(res["status"], "completed") self.assertEqual(res["embed_url"], "https://videa.hu/player?f=clean_scraped_embed_123") # Check DB record db_v = get_video_by_id(vid_id) self.assertEqual(db_v["status"], "kész") self.assertEqual(db_v["embed_url"], "https://videa.hu/player?f=clean_scraped_embed_123") self.assertEqual(db_v["mozisarok_submitted"], 1) self.assertEqual(db_v["mozisarok_verified"], 1) finally: delete_video(vid_id) @patch("queue_manager.count_videa_converting_videos", return_value=0) def test_videa_upload_concurrency_and_conversion_limit(self, mock_count): """ Tests: 1. Upload concurrency limits only active uploads ('feltöltés alatt'). 2. While tasks are in conversion or verification, new uploads can proceed ('ellenőrzés mellett mehet feltöltés is'). 3. If 16 videos are in conversion, no new Videa upload may start until converting count drops below 16. 4. VK uploads are independent and run in parallel alongside Videa uploads and conversions. """ qm = UploadQueueManager(videa_concurrency=2, vk_concurrency=2, max_videa_converting=16, auto_recover=False) # 1. Initial empty queue self.assertEqual(qm.get_videa_uploading_count(), 0) self.assertEqual(qm.get_videa_converting_count(), 0) self.assertEqual(qm.get_vk_uploading_count(), 0) self.assertTrue(qm.can_start_videa_upload()) self.assertTrue(qm.can_start_vk_upload()) # 2. Add 1st uploading Videa task task1 = {"id": 1001, "provider": "videa", "status": "uploading"} qm.queue.append(task1) self.assertEqual(qm.get_videa_uploading_count(), 1) self.assertTrue(qm.can_start_videa_upload()) # 3. Add 2nd uploading Videa task -> reaches videa_concurrency (2) task2 = {"id": 1002, "provider": "videa", "status": "uploading"} qm.queue.append(task2) self.assertEqual(qm.get_videa_uploading_count(), 2) self.assertFalse(qm.can_start_videa_upload()) # Videa upload slots full! # Even with Videa upload slots full, VK upload CAN run alongside it! self.assertTrue(qm.can_start_vk_upload()) task_vk = {"id": 2001, "provider": "vk", "status": "uploading"} qm.queue.append(task_vk) self.assertEqual(qm.get_vk_uploading_count(), 1) # 4. Task 1 finishes upload and transitions to 'converting' # The upload slot must be freed immediately, allowing new uploads! task1["status"] = "converting" self.assertEqual(qm.get_videa_uploading_count(), 1) self.assertEqual(qm.get_videa_converting_count(), 1) self.assertTrue(qm.can_start_videa_upload()) # Upload can proceed alongside conversion! # 5. Task 2 also transitions to 'converting' task2["status"] = "converting" self.assertEqual(qm.get_videa_uploading_count(), 0) self.assertEqual(qm.get_videa_converting_count(), 2) self.assertTrue(qm.can_start_videa_upload()) # 6. Simulate 16 total videos in conversion (biztonsági limit = 16) for i in range(3, 17): qm.queue.append({"id": 1000 + i, "provider": "videa", "status": "converting"}) self.assertEqual(qm.get_videa_converting_count(), 16) self.assertEqual(qm.get_videa_uploading_count(), 0) # 16 are converting -> MUST NOT START NEW VIDEA UPLOAD! self.assertFalse(qm.can_start_videa_upload()) # But VK is NOT blocked by Videa conversion limit! self.assertTrue(qm.can_start_vk_upload()) # 7. One video finishes conversion and moves to 'submitting' or 'completed' qm.queue[0]["status"] = "submitting" self.assertEqual(qm.get_videa_converting_count(), 15) # Drops below 16 -> new Videa upload can start immediately! self.assertTrue(qm.can_start_videa_upload()) def test_mozisarok_check_and_resubmit(self): """ Tests: 1. When embed link is already on Mozisarok -> returns already_exists=True, avoids duplicate submission. 2. When embed link is missing from Mozisarok -> submits and verifies. 3. When force=True -> submits even if already found. """ from unittest.mock import MagicMock, patch, patch client = MozisarokClient(token="mock_valid_token") # Mock get_links to return an existing link client.get_links = MagicMock(return_value=[ {"id": 555, "url": "https://videa.hu/player?f=already_existing_embed_123", "quality": 4, "language": "hu"} ]) client.submit_link = MagicMock(return_value={"success": True, "response": {"link_id": 999}}) client.verify_video = MagicMock(return_value={"mozisarok_found": True, "url_reachable": True}) with patch("requests.get") as mock_get: mock_get.return_value.status_code = 200 # Case A: Link already exists -> returns already_exists=True without submitting res = client.check_and_resubmit_link( media_type="movie", tmdb_id=19995, url="https://videa.hu/player?f=already_existing_embed_123", force=False, ) self.assertTrue(res["success"]) self.assertTrue(res["already_exists"]) self.assertFalse(res["submitted"]) client.submit_link.assert_not_called() # Case B: Link already exists, but force=True -> submits anyway res_force = client.check_and_resubmit_link( media_type="movie", tmdb_id=19995, url="https://videa.hu/player?f=already_existing_embed_123", force=True, ) self.assertTrue(res_force["success"]) self.assertFalse(res_force["already_exists"]) self.assertTrue(res_force["submitted"]) client.submit_link.assert_called_once() # Case C: Link does NOT exist yet -> submits and verifies client.submit_link.reset_mock() res_new = client.check_and_resubmit_link( media_type="tv", tmdb_id=1399, season=1, episode=1, url="https://vk.com/video_ext.php?oid=-123&id=456&hash=abc", force=False, ) self.assertTrue(res_new["success"]) self.assertFalse(res_new["already_exists"]) self.assertTrue(res_new["submitted"]) self.assertTrue(res_new["verified"]) client.submit_link.assert_called_once() def test_resubmit_video_from_db(self): """ Tests queue_mgr.resubmit_video_to_mozisarok with a database record: Checks Mozisarok, submits if missing, updates MySQL video status. """ from unittest.mock import MagicMock, patch vid_id = save_video( media_id=1, account_id=None, file_path="D:/dummy/resubmit_test.mp4", file_name="resubmit_test.mp4", file_size=50000, provider="videa", quality="1080p", status="kész", ) self.assertGreater(vid_id, 0) update_video_progress( vid_id, status="kész", embed_url="https://videa.hu/player?f=test_resubmit_embed_777", mozisarok_submitted=0, mozisarok_verified=0, ) try: qm = UploadQueueManager(auto_recover=False) # Mock Mozisarok check_and_resubmit_link qm.mozisarok.check_and_resubmit_link = MagicMock(return_value={ "success": True, "already_exists": False, "submitted": True, "verified": True, "message": "Sikeresen beküldve!", "embed_url": "https://videa.hu/player?f=test_resubmit_embed_777", }) res = qm.resubmit_video_to_mozisarok(vid_id, force=False) self.assertTrue(res["success"]) self.assertTrue(res["submitted"]) # Verify DB was updated db_v = get_video_by_id(vid_id) self.assertEqual(db_v["mozisarok_submitted"], 1) self.assertEqual(db_v["mozisarok_verified"], 1) finally: delete_video(vid_id) def test_saved_series_list_and_episodes(self): """Test retrieving TV series and episodes grouped from MySQL database.""" test_tmdb = 77766655 media_id = save_media_item( tmdb_id=test_tmdb, media_type="tv", title="Test Batch TV Series", overview="Series for testing batch submission", release_date="2026-05-01", ) self.assertGreater(media_id, 0) vid_1 = save_video( media_id=media_id, account_id=None, file_path="D:/dummy/test_s1e1.mp4", file_name="test_s1e1.mp4", file_size=1024, provider="videa", season=1, episode=1, quality="1080p", status="kész", ) vid_2 = save_video( media_id=media_id, account_id=None, file_path="D:/dummy/test_s1e2.mp4", file_name="test_s1e2.mp4", file_size=1024, provider="videa", season=1, episode=2, quality="1080p", status="kész", ) update_video_progress(vid_1, embed_url="https://videa.hu/player?f=batch_s1e1") update_video_progress(vid_2, embed_url="https://videa.hu/player?f=batch_s1e2") try: # Check get_saved_series_list series_list = get_saved_series_list() matched = [s for s in series_list if s["tmdb_id"] == test_tmdb] self.assertEqual(len(matched), 1) self.assertGreaterEqual(matched[0]["total_episodes"], 2) # Check get_series_episodes_from_db episodes = get_series_episodes_from_db(test_tmdb, season=1) self.assertGreaterEqual(len(episodes), 2) self.assertEqual(episodes[0]["season"], 1) self.assertEqual(episodes[0]["episode"], 1) self.assertEqual(episodes[1]["season"], 1) self.assertEqual(episodes[1]["episode"], 2) finally: delete_video(vid_1) delete_video(vid_2) def test_check_and_submit_series_batch_skips_existing(self): """ Verify that check_and_submit_series_batch checks Mozisarok and only submits missing episodes, skipping episodes that are already on Mozisarok when only_missing=True. """ qm = UploadQueueManager(auto_recover=False) # Mock Mozisarok: # S01E01 is already on Mozisarok # S01E02 is missing from Mozisarok def mock_get_links(media_type, tmdb_id, season=None, episode=None): if season == 1 and episode == 1: return [{"id": 901, "url": "https://videa.hu/player?f=ep1_embed"}] return [] qm.mozisarok.get_links = MagicMock(side_effect=mock_get_links) qm.mozisarok.submit_link = MagicMock(return_value={"success": True, "response": {"id": 999}}) qm.mozisarok.verify_video = MagicMock(return_value={"url_reachable": True, "mozisarok_found": True}) test_episodes = [ { "id": None, "season": 1, "episode": 1, "file_name": "Series.S01E01.mp4", "embed_url": "https://videa.hu/player?f=ep1_embed", }, { "id": None, "season": 1, "episode": 2, "file_name": "Series.S01E02.mp4", "embed_url": "https://videa.hu/player?f=ep2_embed", }, ] res = qm.check_and_submit_series_batch( tmdb_id=123456, season=1, only_missing=True, force=False, episodes_data=test_episodes, ) self.assertTrue(res["success"]) self.assertEqual(res["total_episodes"], 2) self.assertEqual(res["already_present"], 1) # S01E01 was skipped self.assertEqual(res["submitted"], 1) # S01E02 was submitted self.assertEqual(res["failed"], 0) # Ensure submit_link was called ONLY ONCE, for episode 2 self.assertEqual(qm.mozisarok.submit_link.call_count, 1) call_kwargs = qm.mozisarok.submit_link.call_args[1] self.assertEqual(call_kwargs["season"], 1) self.assertEqual(call_kwargs["episode"], 2) self.assertEqual(call_kwargs["url"], "https://videa.hu/player?f=ep2_embed") def test_check_and_submit_series_batch_check_only(self): """Verify check_only mode does not call submit_link and reports status correctly.""" qm = UploadQueueManager(auto_recover=False) def mock_get_links(media_type, tmdb_id, season=None, episode=None): if season == 1 and episode == 1: return [{"id": 901, "url": "https://videa.hu/player?f=ep1_embed"}] return [] qm.mozisarok.get_links = MagicMock(side_effect=mock_get_links) qm.mozisarok.submit_link = MagicMock() test_episodes = [ {"id": None, "season": 1, "episode": 1, "file_name": "S01E01.mp4", "embed_url": "https://videa.hu/player?f=ep1"}, {"id": None, "season": 1, "episode": 2, "file_name": "S01E02.mp4", "embed_url": "https://videa.hu/player?f=ep2"}, ] res = qm.check_and_submit_series_batch( tmdb_id=987654, season=1, only_missing=True, check_only=True, episodes_data=test_episodes, ) self.assertTrue(res["success"]) self.assertEqual(res["total_episodes"], 2) self.assertEqual(res["already_present"], 1) self.assertEqual(res["submitted"], 0) # submit_link must NEVER be called in check_only mode qm.mozisarok.submit_link.assert_not_called() def test_system_logger_ring_buffer_and_file(self): """ Verify that logger_config correctly buffers logs in-memory, writes to rotating file, and supports filtering by level, source, and search term. """ import logging from logger_config import clear_logs, get_log_file_path, get_recent_logs, setup_logging setup_logging(level=logging.DEBUG) clear_logs() t_logger = logging.getLogger("TestDebugModule") t_logger.info("Információs tesztüzenet.") t_logger.warning("Figyelmeztetés: speciális tesztkulcs.") t_logger.error("Hiba történt a videó feldolgozása közben: #999.") # 1. Fetch all logs_res = get_recent_logs(limit=50) self.assertTrue(logs_res["success"]) messages = [l["message"] for l in logs_res["logs"]] self.assertTrue(any("Információs tesztüzenet." in m for m in messages)) self.assertTrue(any("speciális tesztkulcs" in m for m in messages)) self.assertTrue(any("Hiba történt a videó" in m for m in messages)) # 2. Filter by level err_res = get_recent_logs(level="ERROR") err_msgs = [l["message"] for l in err_res["logs"]] self.assertTrue(all(l["level_no"] >= logging.ERROR for l in err_res["logs"])) self.assertTrue(any("Hiba történt a videó" in m for m in err_msgs)) # 3. Filter by search search_res = get_recent_logs(search="speciális tesztkulcs") self.assertEqual(len(search_res["logs"]), 1) self.assertIn("speciális tesztkulcs", search_res["logs"][0]["message"]) # 4. Filter by source src_res = get_recent_logs(source="TestDebugModule") self.assertTrue(len(src_res["logs"]) >= 3) self.assertTrue(all("TestDebugModule" in l["name"] for l in src_res["logs"])) # 5. File exists log_file = get_log_file_path() self.assertTrue(log_file.exists()) with open(log_file, "r", encoding="utf-8") as f: content = f.read() self.assertIn("TestDebugModule", content) def test_api_logs_endpoints(self): """ Verify FastAPI /api/logs endpoints directly: - get_logs_endpoint - download_logs_endpoint - clear_logs_endpoint """ from app import clear_logs_endpoint, download_logs_endpoint, get_logs_endpoint import logging logging.getLogger("ApiTestModule").error("Kritikus hiba az API teszteléshez.") # 1. GET /api/logs data = get_logs_endpoint(level="ERROR", search="Kritikus") self.assertTrue(data["success"]) self.assertTrue(len(data["logs"]) >= 1) self.assertIn("Kritikus", data["logs"][-1]["message"]) # 2. GET /api/logs/download dl_res = download_logs_endpoint() self.assertEqual(dl_res.media_type, "text/plain; charset=utf-8") self.assertTrue(dl_res.path.endswith(".log")) # 3. DELETE /api/logs del_res = clear_logs_endpoint() self.assertTrue(del_res["success"]) def test_admin_auth_and_lockout(self): """ Teszteli az Adminisztrátori hitelesítést, jelszóhashelést, munkamenetet, brute-force védelmet és jelszómódosítást. """ from auth import ( authenticate_admin, change_admin_credentials, hash_password, is_ip_locked, record_login_failure, reset_login_failures, setup_initial_admin, validate_session, verify_password, ) from database import db_pool # 1. Password hashing & constant-time verify p_hash, salt = hash_password("SuperSecret123!") self.assertTrue(verify_password("SuperSecret123!", p_hash, salt)) self.assertFalse(verify_password("WrongPassword!", p_hash, salt)) # 2. Reset admin_users table for test with db_pool.cursor() as cur: cur.execute("DELETE FROM admin_users") # 3. Setup initial admin setup_res = setup_initial_admin("superadmin", "Password123!") self.assertTrue(setup_res) # Duplicate setup should fail self.assertFalse(setup_initial_admin("anotheradmin", "Password123!")) # 4. Authenticate valid admin test_ip = "192.168.1.155" ok, token, err = authenticate_admin("superadmin", "Password123!", test_ip) self.assertTrue(ok) self.assertIsNotNone(token) self.assertIsNone(err) # 5. Validate session token session = validate_session(token) self.assertIsNotNone(session) self.assertEqual(session["username"], "superadmin") # 6. Brute force lockout test (5 failed attempts) bad_ip = "10.0.0.99" reset_login_failures(bad_ip) for i in range(5): ok, tok, err = authenticate_admin("superadmin", "WrongGuess!", bad_ip) self.assertFalse(ok) locked, remaining = is_ip_locked(bad_ip) self.assertTrue(locked) self.assertTrue(remaining > 0) # Locked IP authentication blocked ok, tok, err = authenticate_admin("superadmin", "Password123!", bad_ip) self.assertFalse(ok) self.assertIn("zárolva", err) reset_login_failures(bad_ip) # 7. Credential modification changed = change_admin_credentials( current_password="Password123!", new_username="bossadmin", new_password="NewSecretPass999!", ) self.assertTrue(changed) # Old password must now fail ok, tok, err = authenticate_admin("bossadmin", "Password123!", test_ip) self.assertFalse(ok) # New credentials succeed ok, tok, err = authenticate_admin("bossadmin", "NewSecretPass999!", test_ip) self.assertTrue(ok) # 8. Restore default admin account so database remains clean for user with db_pool.cursor() as cur: cur.execute("DELETE FROM admin_users") from auth import ensure_default_admin ensure_default_admin() def test_account_edit_details(self): """ Teszteli a fiók adatainak (név, felhasználónév, jelszó) utólagos szerkesztését. """ from database import ( delete_account, get_account_by_id, save_account, update_account_details, ) acc_id = save_account( provider="videa", name="Eredeti Fiók Név", username="original@example.com", password="InitialPassword123", is_active=1, ) self.assertTrue(acc_id > 0) # 1. Update name and username only (leave password unchanged) updated = update_account_details( account_id=acc_id, name="Módosított Fiók Név", username="modified@example.com", password="", # Empty means don't change ) self.assertTrue(updated) acc = get_account_by_id(acc_id) self.assertEqual(acc["name"], "Módosított Fiók Név") self.assertEqual(acc["username"], "modified@example.com") self.assertEqual(acc["password"], "InitialPassword123") # Kept intact! # 2. Update password explicitly update_account_details( account_id=acc_id, password="BrandNewEncryptedPass456", ) acc_new = get_account_by_id(acc_id) self.assertEqual(acc_new["password"], "BrandNewEncryptedPass456") # Cleanup delete_account(acc_id) def test_centralized_settings_and_ram_buffer(self): """ Teszteli a központosított beállítások lekérését és mentését, valamint a 128 MB RAM lemezolvasási limit érvényesülését. """ from settings_manager import get_centralized_settings, save_centralized_settings # 1. Get settings structure settings = get_centralized_settings() self.assertIn("version", settings) self.assertIn("server", settings) self.assertIn("performance", settings) self.assertIn("browser", settings) self.assertIn("database", settings) # 2. Save modified settings (RAM limit, port, folder preferences) saved = save_centralized_settings({ "server_port": 8090, "max_disk_buffer_mb": 64, "default_browse_folder": "C:\\Videos", "remember_last_folder": True, "videa_concurrency": 3, }) self.assertEqual(saved["server"]["port"], 8090) self.assertEqual(saved["performance"]["max_disk_buffer_mb"], 64) self.assertEqual(saved["browser"]["default_browse_folder"], "C:\\Videos") self.assertEqual(saved["performance"]["videa_concurrency"], 3) # 3. Verify DiskStreamingMultipart respects clamped RAM limit from stream_uploader import DiskStreamingMultipart import tempfile from pathlib import Path with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp: tmp.write(b"0" * 1024 * 512) tmp_path = Path(tmp.name) try: stream = DiskStreamingMultipart( fields={"title": "Test"}, file_field_name="file", file_path=tmp_path, chunk_size=500 * 1024 * 1024, # Intentionally large 500 MB requested ) # Must be clamped by max_ram_bytes (64 MB = 67,108,864 bytes) self.assertLessEqual(stream.chunk_size, 64 * 1024 * 1024) stream.close() finally: if tmp_path.exists(): tmp_path.unlink() # Restore default 128 MB and 8000 save_centralized_settings({ "server_port": 8000, "max_disk_buffer_mb": 128, "default_browse_folder": "", "videa_concurrency": 2, }) def test_system_recovery_converting_preserved(self): """ Verify that a video left in 'konvertálás alatt' is NOT prematurely closed on restart, but restored into the queue with converting status. """ from database import delete_video, get_video_by_id, save_video, update_video_progress from queue_manager import UploadQueueManager vid_id = save_video( media_id=1, account_id=None, file_path="D:/dummy/converting_test.mp4", file_name="converting_test.mp4", file_size=15000, provider="videa", quality="1080p", status="konvertálás alatt", ) self.assertGreater(vid_id, 0) update_video_progress( vid_id, status="konvertálás alatt", provider_video_id="vcode_converting_recovery", direct_url="https://videa.hu/player?v=vcode_converting_recovery", ) try: qm = UploadQueueManager(auto_recover=False) summary = qm.recover_from_database(run_in_background=False) self.assertGreaterEqual(summary["converting_pending"], 1) # Must NOT be completed db_v = get_video_by_id(vid_id) self.assertEqual(db_v["status"], "konvertálás alatt") self.assertIsNone(db_v.get("completed_at")) # Must be in memory queue as converting q_task = next((t for t in qm.get_queue_status() if t["id"] == vid_id), None) self.assertIsNotNone(q_task) self.assertEqual(q_task["status"], "converting") self.assertIn("átalakító sorára", q_task.get("error_message", "")) finally: delete_video(vid_id) def test_scan_files_episodes_api(self): """ Verify that scan_files_episodes returns parsed episodes for specified files. """ import tempfile from pathlib import Path from app import scan_files_episodes, ScanFilesRequest with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4", prefix="Show.S02E05.") as tmp1: p1 = Path(tmp1.name) with tempfile.NamedTemporaryFile(delete=False, suffix=".mkv", prefix="Show.S02E06.") as tmp2: p2 = Path(tmp2.name) try: req = ScanFilesRequest(file_paths=[str(p1), str(p2)]) data = scan_files_episodes(req) self.assertEqual(data["count"], 2) eps = data["episodes"] self.assertEqual(eps[0]["season"], 2) self.assertEqual(eps[0]["episode"], 5) self.assertEqual(eps[1]["season"], 2) self.assertEqual(eps[1]["episode"], 6) finally: if p1.exists(): p1.unlink() if p2.exists(): p2.unlink() def test_database_schema_check(self): """ Verify that check_database_schema() correctly inspects all 5 MySQL tables, columns, row counts, sizes, and returns a healthy schematic report. """ from database import check_database_schema, EXPECTED_SCHEMA report = check_database_schema() self.assertIsNotNone(report) self.assertTrue(report.get("healthy"), f"Schema must be healthy! Issues: {report.get('issues')}") self.assertIsNotNone(report.get("latency_ms")) self.assertIn("server_info", report) self.assertIn("tables", report) self.assertEqual(len(report["tables"]), len(EXPECTED_SCHEMA)) table_names = [t["table_name"] for t in report["tables"]] for exp in ["accounts", "media_items", "videos", "app_settings", "admin_users"]: self.assertIn(exp, table_names) for t in report["tables"]: self.assertTrue(t["exists"], f"Table {t['table_name']} must exist!") self.assertEqual(len(t["missing_columns"]), 0, f"Table {t['table_name']} has missing columns: {t['missing_columns']}") self.assertGreaterEqual(t["columns_count"], 3) self.assertIn(t["status"], ["ok", "warning"]) def test_database_schema_repair_and_endpoints(self): """ Verify repair_database_schema() and FastAPI schema endpoints. """ from database import repair_database_schema from app import get_database_schema_endpoint, repair_database_schema_endpoint # 1. Repair function rep = repair_database_schema() self.assertTrue(rep.get("healthy")) # 2. GET endpoint ep_rep = get_database_schema_endpoint() self.assertTrue(ep_rep.get("healthy")) self.assertEqual(ep_rep["summary"]["total_expected_tables"], 5) # 3. POST repair endpoint post_rep = repair_database_schema_endpoint() self.assertTrue(post_rep.get("success")) self.assertTrue(post_rep["report"].get("healthy")) def test_updater_version_comparison(self): """Test semantic version parsing and comparison logic.""" from updater import compare_versions, parse_version_tuple self.assertEqual(parse_version_tuple("1.2.0"), (1, 2, 0)) self.assertEqual(compare_versions("1.2.0", "1.2.0"), 0) self.assertEqual(compare_versions("1.2.0", "1.3.0"), -1) self.assertEqual(compare_versions("1.3.0", "1.2.0"), 1) self.assertEqual(compare_versions("1.2.0", "1.2.1"), -1) self.assertEqual(compare_versions("1.2", "1.2.0"), 0) def test_updater_check_for_updates_mocked(self): """Test check_for_updates() with mocked server responses.""" from updater import check_for_updates from unittest.mock import MagicMock, patch # 1. Update available mock_resp = MagicMock() mock_resp.status_code = 200 mock_resp.json.return_value = { "status": "success", "latest_version": "9.9.9", "release_date": "2026-11-01", "download_url": "https://test.nyamnet.hu/download.php?v=9.9.9", "file_name": "uploader-v9.9.9.zip", "file_size": 123456, "sha256": "abcdef1234567890", "changelog": ["Feature A", "Feature B"], } with patch("requests.get", return_value=mock_resp): res = check_for_updates("https://test.nyamnet.hu") self.assertTrue(res["success"]) self.assertTrue(res["update_available"]) self.assertEqual(res["latest_version"], "9.9.9") self.assertEqual(len(res["changelog"]), 2) # 2. Already latest version mock_resp.json.return_value["latest_version"] = "1.0.0" with patch("requests.get", return_value=mock_resp): res = check_for_updates("https://test.nyamnet.hu") self.assertTrue(res["success"]) self.assertFalse(res["update_available"]) # 3. Server unreachable with patch("requests.get", side_effect=Exception("Connection refused")): res = check_for_updates("https://test.nyamnet.hu") self.assertFalse(res["success"]) self.assertFalse(res["update_available"]) self.assertIn("Nem sikerült kapcsolódni", res["error"]) def test_updater_apply_update_and_backup(self): """Test apply_update() differential file download, sha256 check, and backup creation.""" import hashlib from pathlib import Path from updater import apply_update from unittest.mock import MagicMock, patch dummy_bytes = b"DIFFERENTIAL_SYNC_TEST_CONTENT" real_sha256 = hashlib.sha256(dummy_bytes).hexdigest() mock_dl_resp = MagicMock() mock_dl_resp.status_code = 200 mock_dl_resp.content = dummy_bytes mock_check_data = { "success": True, "update_available": True, "current_version": "1.2.0", "latest_version": "1.3.0", "files_to_update": [ {"path": "test_diff_dummy.txt", "size": len(dummy_bytes), "sha256": real_sha256, "status": "modified"} ], "download_base_url": "https://test.nyamnet.hu/file.php?path=", "changelog": ["Teszt diff"], } # 1. Successful differential update with patch("updater.check_for_updates", return_value=mock_check_data), \ patch("requests.get", return_value=mock_dl_resp): report = apply_update(force=True) self.assertTrue(report["success"]) self.assertTrue(report["updated"]) self.assertEqual(report["new_version"], "1.3.0") self.assertEqual(report["sync_type"], "differential") self.assertTrue(Path("test_diff_dummy.txt").exists()) self.assertEqual(Path("test_diff_dummy.txt").read_bytes(), dummy_bytes) Path("test_diff_dummy.txt").unlink() # 2. SHA256 mismatch detection bad_check_data = dict(mock_check_data) bad_check_data["files_to_update"] = [ {"path": "test_diff_dummy.txt", "size": len(dummy_bytes), "sha256": "WRONG_CHECKSUM", "status": "modified"} ] with patch("updater.check_for_updates", return_value=bad_check_data), \ patch("requests.get", return_value=mock_dl_resp): with self.assertRaises(ValueError) as ctx: apply_update(force=True) self.assertIn("Integritás hiba", str(ctx.exception)) def test_sync_push_tool_scan(self): """Verify that sync_push tool correctly scans local workspace and excludes sensitive files.""" from sync_push import scan_local_files scanned = scan_local_files() self.assertIn("app.py", scanned) self.assertIn("database.py", scanned) self.assertIn("updater.py", scanned) # Critical: .env and private files must NEVER be in scan result self.assertNotIn(".env", scanned) self.assertNotIn("config.zip", scanned) for path in scanned: self.assertFalse(path.startswith(".git/")) self.assertFalse(path.startswith("logs/")) self.assertFalse(path.startswith("backups/")) self.assertFalse(path.startswith("update_server/")) def test_updater_endpoints(self): """Test FastAPI update endpoints.""" from app import check_updates_endpoint, set_update_server_url_endpoint from unittest.mock import patch # 1. Server URL endpoint res = set_update_server_url_endpoint({"server_url": "https://test.nyamnet.hu/updater"}) self.assertTrue(res["success"]) self.assertEqual(res["server_url"], "https://test.nyamnet.hu/updater") # Reset to default set_update_server_url_endpoint({"server_url": "https://test.nyamnet.hu"}) # 2. Check updates endpoint with patch("app.check_for_updates", return_value={"success": True, "update_available": False, "latest_version": "1.2.0"}): res_check = check_updates_endpoint() self.assertTrue(res_check["success"]) def test_sync_push_ssl_fallback_to_http(self): """Test that sync_push.safe_request automatically retries with plain HTTP on SSL errors.""" import requests from sync_push import is_ssl_error, safe_request ssl_err = requests.exceptions.SSLError("certificate verify failed: self-signed certificate") self.assertTrue(is_ssl_error(ssl_err)) self.assertTrue(is_ssl_error(Exception("SSL handshake failed: certificate expired"))) self.assertFalse(is_ssl_error(Exception("Connection refused"))) # Test safe_request fallback mock_success_resp = MagicMock() mock_success_resp.status_code = 200 def mock_request(method, url, **kwargs): if url.startswith("https://"): raise requests.exceptions.SSLError("SSL: CERTIFICATE_VERIFY_FAILED") elif url.startswith("http://"): return mock_success_resp raise RuntimeError("Unexpected URL") with patch("requests.request", side_effect=mock_request): resp, used_url = safe_request("POST", "https://test.nyamnet.hu/sync_api.php?action=check") self.assertEqual(resp, mock_success_resp) self.assertEqual(used_url, "http://test.nyamnet.hu/sync_api.php?action=check") def test_updater_ssl_fallback_to_http(self): """Test that updater.check_for_updates retries with plain HTTP when HTTPS fails with SSL error.""" import requests from updater import check_for_updates mock_resp = MagicMock() mock_resp.status_code = 200 mock_resp.json.return_value = { "status": "success", "latest_version": "1.2.1", "files": {}, "download_base_url": "http://test.nyamnet.hu/file.php?path=", } def mock_get(url, **kwargs): if url.startswith("https://"): raise requests.exceptions.SSLError("HTTPS SSL handshake failed") elif url.startswith("http://"): return mock_resp raise RuntimeError("Unexpected URL") with patch("requests.get", side_effect=mock_get): res = check_for_updates("https://test.nyamnet.hu") self.assertTrue(res["success"]) self.assertEqual(res["latest_version"], "1.2.1") self.assertTrue(res["server_url"].startswith("http://")) if __name__ == "__main__": unittest.main()