unkinben
2309e9f43a
Five-service streaming platform: auth, catalogue, streaming, ingest, thumbnailer. Includes React frontend served by nginx, NATS JetStream event bus, aiobotocore async S3, PyAV video metadata + thumbnail extraction, service-to-service JWT auth, and a full unit + e2e test suite.
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
import os
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
GATEWAY = os.environ.get("GATEWAY_URL", "http://nginx:80")
|
|
|
|
|
|
@pytest.mark.e2e
|
|
def test_catalogue_lists_seeded_items(seeded_media):
|
|
with httpx.Client(base_url=GATEWAY) as client:
|
|
resp = client.get("/api/v1/catalogue/")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["total"] >= 2
|
|
titles = {item["title"] for item in body["items"]}
|
|
assert "Big Buck Bunny" in titles
|
|
assert "Jellyfish" in titles
|
|
|
|
|
|
@pytest.mark.e2e
|
|
def test_catalogue_get_by_id(seeded_media):
|
|
with httpx.Client(base_url=GATEWAY) as client:
|
|
for item in seeded_media:
|
|
resp = client.get(f"/api/v1/catalogue/{item['id']}")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["id"] == item["id"]
|
|
assert body["title"] == item["title"]
|
|
assert body["media_type"] == item["media_type"]
|
|
|
|
|
|
@pytest.mark.e2e
|
|
def test_catalogue_filter_by_media_type(seeded_media):
|
|
with httpx.Client(base_url=GATEWAY) as client:
|
|
resp = client.get("/api/v1/catalogue/?media_type=movie")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert all(i["media_type"] == "movie" for i in body["items"])
|
|
assert body["total"] >= 2
|
|
|
|
|
|
@pytest.mark.e2e
|
|
def test_stream_token_issued(seeded_media, admin_token):
|
|
item = seeded_media[0]
|
|
with httpx.Client(base_url=GATEWAY) as client:
|
|
resp = client.post(
|
|
f"/api/v1/catalogue/{item['id']}/stream-token",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "stream_url" in body
|
|
assert body["stream_url"].startswith("/api/v1/stream/")
|
|
|
|
|
|
@pytest.mark.e2e
|
|
def test_stream_full_response(seeded_media, admin_token):
|
|
item = seeded_media[0]
|
|
with httpx.Client(base_url=GATEWAY) as client:
|
|
token_resp = client.post(
|
|
f"/api/v1/catalogue/{item['id']}/stream-token",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
token_resp.raise_for_status()
|
|
stream_url = token_resp.json()["stream_url"]
|
|
resp = client.get(stream_url, timeout=30)
|
|
assert resp.status_code == 200
|
|
assert "accept-ranges" in resp.headers
|
|
assert int(resp.headers["content-length"]) > 0
|
|
|
|
|
|
@pytest.mark.e2e
|
|
def test_stream_range_request(seeded_media, admin_token):
|
|
item = seeded_media[0]
|
|
with httpx.Client(base_url=GATEWAY) as client:
|
|
token_resp = client.post(
|
|
f"/api/v1/catalogue/{item['id']}/stream-token",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
token_resp.raise_for_status()
|
|
stream_url = token_resp.json()["stream_url"]
|
|
resp = client.get(stream_url, headers={"Range": "bytes=0-65535"}, timeout=30)
|
|
assert resp.status_code == 206
|
|
assert "content-range" in resp.headers
|
|
assert resp.headers["content-length"] == "65536"
|
|
assert len(resp.content) == 65536
|
|
|
|
|
|
@pytest.mark.e2e
|
|
def test_stream_both_videos(seeded_media, admin_token):
|
|
with httpx.Client(base_url=GATEWAY) as client:
|
|
for item in seeded_media:
|
|
token_resp = client.post(
|
|
f"/api/v1/catalogue/{item['id']}/stream-token",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
assert token_resp.status_code == 200
|
|
stream_url = token_resp.json()["stream_url"]
|
|
resp = client.get(stream_url, headers={"Range": "bytes=0-1023"}, timeout=30)
|
|
assert resp.status_code == 206, f"Failed for {item['title']}"
|