feat(video_analysis): download_video.py with TDD (5 tests)

scripts/video_analysis/download_video.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from scripts.video_analysis.error_types import ErrorInfo, make_error
+
+
+@dataclass
+class _Ok:
+ value: Any
+
+ def is_ok(self) -> bool:
+  return True
+
+ def is_err(self) -> bool:
+  return False
+
+
+@dataclass
+class _Err:
+ err: ErrorInfo
+
+ def is_ok(self) -> bool:
+  return False
+
+ def is_err(self) -> bool:
+  return True
+
+
+def ok(value: Any) -> _Ok:
+ return _Ok(value)
+
+
+def err(error: ErrorInfo) -> _Err:
+ return _Err(error)
+
+
+def validate_output_path(path: Path) -> _Ok | _Err:
+ if path.exists() and path.is_dir():
+  return err(make_error("OutputIsDirectory", "validate_output_path", str(path)))
+ path.parent.mkdir(parents=True, exist_ok=True)
+ return ok(path)
+
+
+def build_ydl_args(url: str, output: Path) -> list[str]:
+ return [
+  "yt-dlp",
+  "--format", "bestvideo[ext=mp4]/best",
+  "--output", str(output),
+  "--no-warnings",
+  "--quiet",
+  url,
+ ]
+
+
+def download_video(url: str, output: Path) -> _Ok | _Err:
+ validated = validate_output_path(output)
+ if validated.is_err():
+  return validated
+ completed = subprocess.run(
+  build_ydl_args(url, output),
+  capture_output=True,
+  text=True,
+ )
+ log_path = output.with_suffix(".log")
+ log_path.write_text(
+  f"# yt-dlp log\n# url: {url}\n# output: {output}\n# returncode: {completed.returncode}\n\nstdout:\n{completed.stdout}\n\nstderr:\n{completed.stderr}\n",
+  encoding="utf-8",
+ )
+ if completed.returncode != 0:
+  return err(make_error("YtdlpError", "download_video", completed.stderr[:500]))
+ return ok({"output": str(output), "log": str(log_path), "returncode": completed.returncode})

tests/test_video_analysis_download_video.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from scripts.video_analysis.download_video import (
+ build_ydl_args,
+ download_video,
+ validate_output_path,
+)
+
+
+def test_validate_output_path_creates_parent(tmp_path: Path) -> None:
+ out = tmp_path / "subdir" / "video.mp4"
+ result = validate_output_path(out)
+ assert result.is_ok()
+ assert out.parent.exists()
+
+
+def test_validate_output_path_rejects_existing_dir(tmp_path: Path) -> None:
+ out = tmp_path / "existing_dir"
+ out.mkdir()
+ result = validate_output_path(out)
+ assert result.is_err()
+
+
+def test_build_ydl_args_basic(tmp_path: Path) -> None:
+ out = tmp_path / "v.mp4"
+ args = build_ydl_args("https://youtu.be/VID", out)
+ assert "--output" in args
+ assert str(out) in args
+ assert "https://youtu.be/VID" in args
+
+
+def test_download_video_success(tmp_path: Path) -> None:
+ out = tmp_path / "video.mp4"
+ out.write_bytes(b"fake-mp4-content")
+ with patch("subprocess.run") as mock_run:
+  mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+  result = download_video("https://youtu.be/VID", out)
+  assert result.is_ok()
+  assert (tmp_path / "video.log").exists()
+
+
+def test_download_video_failure(tmp_path: Path) -> None:
+ out = tmp_path / "video.mp4"
+ with patch("subprocess.run") as mock_run:
+  mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="ERROR: video unavailable")
+  result = download_video("https://youtu.be/VID", out)
+  assert result.is_err()