ytdlp/tests/test_inspection.py

import socket
from typing import Any

import pytest
from fastapi.testclient import TestClient

import app.inspection as inspection
import app.main as main_module
from app.inspection import UnsafeUrlError
from app.main import app

client = TestClient(app)


def sample_inspection() -> dict[str, Any]:
    return {
        "id": "abc123",
        "title": "Example report",
        "extractor": "Example",
        "uploader": "Newsroom",
        "duration": 42.5,
        "upload_date": "20260811",
        "source_url": "https://video.example/watch/abc123",
        "webpage_url": "https://video.example/watch/abc123",
        "thumbnail": "https://cdn.example/abc123.jpg",
        "is_live": False,
        "formats": [
            {
                "format_id": "1080p",
                "extension": "mp4",
                "width": 1920,
                "height": 1080,
                "fps": 50.0,
                "video_codec": "avc1.64002a",
                "audio_codec": "mp4a.40.2",
                "filesize": 12_345_678,
            }
        ],
    }


def test_inspect_endpoint_returns_normalized_metadata(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(main_module, "inspect_url", lambda _url: sample_inspection())

    response = client.post("/api/inspect", json={"url": "https://video.example/watch/abc123"})

    assert response.status_code == 200
    assert response.json()["title"] == "Example report"
    assert response.json()["formats"][0]["height"] == 1080


def test_inspect_endpoint_rejects_unsafe_url(monkeypatch: pytest.MonkeyPatch) -> None:
    def reject_url(_url: str) -> dict[str, Any]:
        raise UnsafeUrlError("The source hostname resolves to a non-public address.")

    monkeypatch.setattr(main_module, "inspect_url", reject_url)

    response = client.post("/api/inspect", json={"url": "http://127.0.0.1/private"})

    assert response.status_code == 400
    assert response.json() == {
        "detail": "The source hostname resolves to a non-public address."
    }


def test_validate_public_url_accepts_public_addresses(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [
            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))
        ],
    )

    url = "https://video.example/watch/abc123"

    assert inspection.validate_public_url(url) == url


def test_validate_public_url_rejects_if_any_address_is_private(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(
        socket,
        "getaddrinfo",
        lambda *_args, **_kwargs: [
            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)),
            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443)),
        ],
    )

    with pytest.raises(UnsafeUrlError, match="non-public"):
        inspection.validate_public_url("https://video.example/watch/abc123")


def test_validate_public_url_rejects_credentials() -> None:
    with pytest.raises(UnsafeUrlError, match="Credentials"):
        inspection.validate_public_url("https://user:password@video.example/watch/abc123")


def test_inspect_url_removes_raw_format_urls(monkeypatch: pytest.MonkeyPatch) -> None:
    captured_options: dict[str, Any] = {}

    class FakeYoutubeDL:
        def __init__(self, options: dict[str, Any]) -> None:
            captured_options.update(options)

        def __enter__(self) -> "FakeYoutubeDL":
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def extract_info(self, url: str, *, download: bool) -> dict[str, Any]:
            assert url == "https://video.example/watch/abc123"
            assert download is False
            return {
                **sample_inspection(),
                "extractor_key": "Example",
                "formats": [
                    {
                        **sample_inspection()["formats"][0],
                        "ext": "mp4",
                        "vcodec": "avc1.64002a",
                        "acodec": "mp4a.40.2",
                        "url": "https://secret-cdn.example/signed-stream-url",
                    }
                ],
            }

        def sanitize_info(self, info: dict[str, Any]) -> dict[str, Any]:
            return info

    monkeypatch.setattr(inspection, "validate_public_url", lambda url: url)
    monkeypatch.setattr(inspection.yt_dlp, "YoutubeDL", FakeYoutubeDL)

    result = inspection.inspect_url("https://video.example/watch/abc123")

    assert captured_options["skip_download"] is True
    assert captured_options["noplaylist"] is True
    assert "url" not in result["formats"][0]


def test_normalized_formats_remove_storyboards_and_cdn_duplicates() -> None:
    formats = inspection._normalized_formats(
        [
            {
                "format_id": "storyboard",
                "ext": "mhtml",
                "vcodec": "none",
                "acodec": "none",
            },
            {
                "format_id": "akamai-720p",
                "ext": "mp4",
                "width": 1280,
                "height": 720,
                "fps": 50.0,
                "vcodec": "avc1.640020",
                "acodec": "mp4a.40.5",
            },
            {
                "format_id": "cloudfront-720p",
                "ext": "mp4",
                "width": 1280,
                "height": 720,
                "fps": 50.0,
                "vcodec": "avc1.640020",
                "acodec": "mp4a.40.5",
            },
        ]
    )

    assert len(formats) == 1
    assert formats[0]["format_id"] == "akamai-720p"


def test_inspect_url_accepts_single_video_wrapper(monkeypatch: pytest.MonkeyPatch) -> None:
    class FakeYoutubeDL:
        def __init__(self, _options: dict[str, Any]) -> None:
            pass

        def __enter__(self) -> "FakeYoutubeDL":
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def extract_info(self, _url: str, *, download: bool) -> dict[str, Any]:
            assert download is False
            return {
                "_type": "playlist",
                "title": "BBC article title",
                "entries": [
                    {
                        "id": "bbc-video-1",
                        "title": "BBC video title",
                        "formats": [
                            {
                                "format_id": "h264-1080",
                                "ext": "mp4",
                                "width": 1920,
                                "height": 1080,
                                "vcodec": "avc1.640028",
                                "acodec": "mp4a.40.2",
                            }
                        ],
                    }
                ],
            }

        def sanitize_info(self, info: dict[str, Any]) -> dict[str, Any]:
            return info

    monkeypatch.setattr(inspection, "validate_public_url", lambda url: url)
    monkeypatch.setattr(inspection.yt_dlp, "YoutubeDL", FakeYoutubeDL)

    result = inspection.inspect_url("https://www.bbc.com/news/videos/example")

    assert result["id"] == "bbc-video-1"
    assert result["title"] == "BBC video title"
    assert result["source_url"] == "https://www.bbc.com/news/videos/example"


def test_inspect_url_rejects_wrapper_with_multiple_videos(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    class FakeYoutubeDL:
        def __init__(self, _options: dict[str, Any]) -> None:
            pass

        def __enter__(self) -> "FakeYoutubeDL":
            return self

        def __exit__(self, *_args: object) -> None:
            return None

        def extract_info(self, _url: str, *, download: bool) -> dict[str, Any]:
            assert download is False
            return {"_type": "playlist", "entries": [{"id": "one"}, {"id": "two"}]}

        def sanitize_info(self, info: dict[str, Any]) -> dict[str, Any]:
            return info

    monkeypatch.setattr(inspection, "validate_public_url", lambda url: url)
    monkeypatch.setattr(inspection.yt_dlp, "YoutubeDL", FakeYoutubeDL)

    with pytest.raises(inspection.InspectionError, match="multi-video"):
        inspection.inspect_url("https://video.example/playlist")