130 lines
4.8 KiB
Python
130 lines
4.8 KiB
Python
"""Tests for the MeTube media ingestion client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
import respx
|
|
from httpx import Response
|
|
|
|
from conduit_client import AsyncConduitClient, ConduitClient
|
|
from conduit_client.models.media_ingest import (
|
|
CreateIngestJobRequest,
|
|
CreateIngestSubscriptionRequest,
|
|
PatchIngestSubscriptionRequest,
|
|
)
|
|
|
|
|
|
@respx.mock
|
|
def test_list_jobs(api_key: str) -> None:
|
|
respx.get("https://api.example.com/media/ingest/jobs").mock(
|
|
return_value=Response(
|
|
200,
|
|
json={"done": [{"id": "job1", "status": "done"}], "downloading": []},
|
|
)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.media_ingest.list_jobs()
|
|
assert len(result.done or []) == 1
|
|
|
|
|
|
@respx.mock
|
|
def test_create_job(api_key: str) -> None:
|
|
route = respx.post("https://api.example.com/media/ingest/jobs").mock(
|
|
return_value=Response(200, json={"id": "job2", "status": "pending"})
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.media_ingest.create_job(
|
|
CreateIngestJobRequest(url="https://youtube.com/watch?v=abc", quality="best")
|
|
)
|
|
assert result.id == "job2"
|
|
assert json.loads(route.calls[0].request.content) == {
|
|
"url": "https://youtube.com/watch?v=abc",
|
|
"quality": "best",
|
|
}
|
|
|
|
|
|
@respx.mock
|
|
def test_start_cancel_delete_job(api_key: str) -> None:
|
|
respx.post("https://api.example.com/media/ingest/jobs/job3/start").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.post("https://api.example.com/media/ingest/jobs/job3/cancel").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.delete("https://api.example.com/media/ingest/jobs/job3").mock(
|
|
return_value=Response(204)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
client.media_ingest.start_job("job3")
|
|
client.media_ingest.cancel_job("job3")
|
|
client.media_ingest.delete_job("job3", delete_from_file=True)
|
|
|
|
|
|
@respx.mock
|
|
def test_list_presets(api_key: str) -> None:
|
|
respx.get("https://api.example.com/media/ingest/presets").mock(
|
|
return_value=Response(200, json=[{"name": "best", "quality": "best", "format": "mp4"}])
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.media_ingest.list_presets()
|
|
assert result[0].name == "best"
|
|
|
|
|
|
@respx.mock
|
|
def test_subscriptions(api_key: str) -> None:
|
|
respx.get("https://api.example.com/media/ingest/subscriptions").mock(
|
|
return_value=Response(200, json=[{"id": "sub1", "url": "https://feed.example.com"}])
|
|
)
|
|
create_route = respx.post("https://api.example.com/media/ingest/subscriptions").mock(
|
|
return_value=Response(200, json={"id": "sub2", "url": "https://feed2.example.com"})
|
|
)
|
|
respx.patch("https://api.example.com/media/ingest/subscriptions/sub1").mock(
|
|
return_value=Response(200, json={"id": "sub1", "url": "https://feed.example.com", "name": "new"})
|
|
)
|
|
respx.delete("https://api.example.com/media/ingest/subscriptions/sub1").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.post("https://api.example.com/media/ingest/subscriptions/sub1/check").mock(
|
|
return_value=Response(204)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
subs = client.media_ingest.list_subscriptions()
|
|
assert len(subs) == 1
|
|
new = client.media_ingest.create_subscription(
|
|
CreateIngestSubscriptionRequest(url="https://feed2.example.com")
|
|
)
|
|
assert new.id == "sub2"
|
|
client.media_ingest.patch_subscription(
|
|
"sub1", PatchIngestSubscriptionRequest(name="new")
|
|
)
|
|
client.media_ingest.delete_subscription("sub1")
|
|
client.media_ingest.check_subscription("sub1")
|
|
assert json.loads(create_route.calls[0].request.content) == {
|
|
"url": "https://feed2.example.com"
|
|
}
|
|
|
|
|
|
@respx.mock
|
|
def test_version(api_key: str) -> None:
|
|
respx.get("https://api.example.com/media/ingest/version").mock(
|
|
return_value=Response(200, json={"metube_version": "1.0", "ytdlp_version": "2024.01.01"})
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.media_ingest.version()
|
|
assert result.metube_version == "1.0"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_async_create_job(api_key: str) -> None:
|
|
respx.post("https://api.example.com/media/ingest/jobs").mock(
|
|
return_value=Response(200, json={"id": "job2", "status": "pending"})
|
|
)
|
|
async with AsyncConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = await client.media_ingest.create_job(
|
|
CreateIngestJobRequest(url="https://youtube.com/watch?v=abc")
|
|
)
|
|
assert result.id == "job2"
|