| 1 | """Regression tests for rate-limiter lock behavior, media resource cleanup, |
| 2 | packaging metadata, config templates, and test-suite isolation.""" |
| 3 | |
| 4 | import asyncio |
| 5 | import subprocess |
| 6 | import sys |
| 7 | import tomllib |
| 8 | import unittest |
| 9 | from contextlib import suppress |
| 10 | from pathlib import Path |
| 11 | from unittest.mock import MagicMock, patch |
| 12 | |
| 13 | import yaml |
| 14 | |
| 15 | from utils.rate_limiter import RateLimiter |
| 16 | from utils.video import concatenate_video_files |
| 17 | |
| 18 | REPO_ROOT = Path(__file__).resolve().parent.parent |
| 19 | |
| 20 | |
| 21 | class TestRateLimiterLocking(unittest.IsolatedAsyncioTestCase): |
| 22 | async def test_waiting_acquirer_does_not_hold_the_lock(self): |
| 23 | limiter = RateLimiter(max_requests_per_minute=1) |
| 24 | await limiter.acquire() # consume the only slot in the window |
| 25 | waiter = asyncio.create_task(limiter.acquire()) |
| 26 | await asyncio.sleep(0.05) # waiter is now waiting for the window to free |
| 27 | try: |
| 28 | try: |
| 29 | await asyncio.wait_for(limiter.lock.acquire(), timeout=0.25) |
| 30 | limiter.lock.release() |
| 31 | except asyncio.TimeoutError: |
| 32 | self.fail("rate limiter sleeps while holding its lock, blocking every other caller") |
| 33 | finally: |
| 34 | waiter.cancel() |
| 35 | with suppress(asyncio.CancelledError): |
| 36 | await waiter |
| 37 | |
| 38 | async def test_min_delay_smoothing_still_applies(self): |
| 39 | limiter = RateLimiter(max_requests_per_minute=600) # min delay 0.1s |
| 40 | loop = asyncio.get_running_loop() |
| 41 | start = loop.time() |
| 42 | await limiter.acquire() |
| 43 | await limiter.acquire() |
| 44 | self.assertGreaterEqual(loop.time() - start, 0.08) |
| 45 | |
| 46 | |
| 47 | class TestVideoConcatenationCleanup(unittest.TestCase): |
| 48 | def test_concatenate_closes_all_clips_even_on_failure(self): |
| 49 | clips = [MagicMock(), MagicMock()] |
| 50 | final = MagicMock() |
| 51 | with patch("utils.video.VideoFileClip", side_effect=clips), \ |
| 52 | patch("utils.video.concatenate_videoclips", return_value=final): |
| 53 | concatenate_video_files(["a.mp4", "b.mp4"], "out.mp4") |
| 54 | final.write_videofile.assert_called_once() |
| 55 | final.close.assert_called_once() |
| 56 | for clip in clips: |
| 57 | clip.close.assert_called_once() |
| 58 | |
| 59 | # And when writing fails, the ffmpeg readers must still be released. |
| 60 | clips = [MagicMock(), MagicMock()] |
| 61 | final = MagicMock() |
| 62 | final.write_videofile.side_effect = OSError("disk full") |
| 63 | with patch("utils.video.VideoFileClip", side_effect=clips), \ |
| 64 | patch("utils.video.concatenate_videoclips", return_value=final): |
| 65 | with self.assertRaises(OSError): |
| 66 | concatenate_video_files(["a.mp4", "b.mp4"], "out.mp4") |
| 67 | final.close.assert_called_once() |
| 68 | for clip in clips: |
| 69 | clip.close.assert_called_once() |
| 70 | |
| 71 | |
| 72 | class TestPackagingMetadata(unittest.TestCase): |
| 73 | def test_pyproject_is_consistent(self): |
| 74 | with open(REPO_ROOT / "pyproject.toml", "rb") as f: |
| 75 | data = tomllib.load(f) |
| 76 | readme = data["project"]["readme"] |
| 77 | self.assertTrue((REPO_ROOT / readme).exists(), f"readme points at missing file: {readme}") |
| 78 | self.assertNotIn("index", data, "top-level [[index]] is not valid pyproject TOML and is silently ignored") |
| 79 | dev_group = data.get("dependency-groups", {}).get("dev", []) |
| 80 | self.assertTrue(any("pytest" in str(item) for item in dev_group), "pytest must be a declared dev dependency so the suite runs from the venv") |
| 81 | |
| 82 | |
| 83 | class TestProviderConfigTemplates(unittest.TestCase): |
| 84 | def test_minimax_templates_do_not_ship_truthy_placeholders(self): |
| 85 | for name in ("idea2video_minimax.yaml", "script2video_minimax.yaml"): |
| 86 | with open(REPO_ROOT / "configs" / name, encoding="utf-8") as f: |
| 87 | config = yaml.safe_load(f) |
| 88 | chat_key = config["chat_model"]["init_args"].get("api_key") |
| 89 | self.assertFalse(chat_key, f"{name}: a truthy api_key placeholder defeats the MINIMAX_API_KEY env fallback") |
| 90 | for section in ("image_generator", "video_generator"): |
| 91 | key = config[section]["init_args"].get("api_key") |
| 92 | self.assertNotIn("<", str(key or ""), f"{name}: {section} ships an angle-bracket placeholder that would be sent as a bearer token") |
| 93 | |
| 94 | |
| 95 | class TestSuiteIsolation(unittest.TestCase): |
| 96 | def test_importing_minimax_tests_does_not_stub_global_modules(self): |
| 97 | code = ( |
| 98 | "import sys; import tests.test_minimax_integration; " |
| 99 | "mod = sys.modules.get('cv2'); " |
| 100 | "from unittest.mock import MagicMock; " |
| 101 | "exit(1 if isinstance(mod, MagicMock) else 0)" |
| 102 | ) |
| 103 | result = subprocess.run( |
| 104 | [sys.executable, "-c", code], |
| 105 | cwd=REPO_ROOT, |
| 106 | capture_output=True, |
| 107 | text=True, |
| 108 | ) |
| 109 | self.assertEqual( |
| 110 | result.returncode, 0, |
| 111 | "importing tests.test_minimax_integration replaces sys.modules entries at import time, " |
| 112 | "making every later-collected test module see MagicMock stubs instead of real libraries", |
| 113 | ) |
| 114 | |
| 115 | |
| 116 | if __name__ == "__main__": |
| 117 | unittest.main() |
| 118 |