250 lines
8.1 KiB
Python
250 lines
8.1 KiB
Python
|
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")
|