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.
105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
from streamstack.catalogue.models import Base, MediaItem, Movie, TvSeries, YoutubeShow
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_session():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
async with session_factory() as session:
|
|
yield session
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_base_media_item(db_session: AsyncSession):
|
|
item = MediaItem(title="Generic Media", s3_key="media/generic.mp4")
|
|
db_session.add(item)
|
|
await db_session.commit()
|
|
await db_session.refresh(item)
|
|
assert item.id is not None
|
|
assert item.media_type == "media"
|
|
assert item.is_published is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_movie(db_session: AsyncSession):
|
|
movie = Movie(
|
|
title="Inception",
|
|
s3_key="media/inception.mp4",
|
|
director="Christopher Nolan",
|
|
release_year=2010,
|
|
mpaa_rating="PG-13",
|
|
imdb_id="tt1375666",
|
|
)
|
|
db_session.add(movie)
|
|
await db_session.commit()
|
|
await db_session.refresh(movie)
|
|
assert movie.media_type == "movie"
|
|
assert movie.director == "Christopher Nolan"
|
|
assert movie.release_year == 2010
|
|
assert movie.mpaa_rating == "PG-13"
|
|
assert movie.imdb_id == "tt1375666"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_tv_series(db_session: AsyncSession):
|
|
episode = TvSeries(
|
|
title="Breaking Bad S01E01",
|
|
s3_key="media/bb-s01e01.mp4",
|
|
show_name="Breaking Bad",
|
|
season=1,
|
|
episode=1,
|
|
episode_title="Pilot",
|
|
network="AMC",
|
|
)
|
|
db_session.add(episode)
|
|
await db_session.commit()
|
|
await db_session.refresh(episode)
|
|
assert episode.media_type == "tv_series"
|
|
assert episode.show_name == "Breaking Bad"
|
|
assert episode.season == 1
|
|
assert episode.episode == 1
|
|
assert episode.episode_title == "Pilot"
|
|
assert episode.network == "AMC"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_youtube_show(db_session: AsyncSession):
|
|
video = YoutubeShow(
|
|
title="Python Tutorial",
|
|
s3_key="media/yt-python.mp4",
|
|
youtube_video_id="dQw4w9WgXcQ",
|
|
youtube_channel_id="UCxxxxxx",
|
|
channel_name="TechChannel",
|
|
)
|
|
db_session.add(video)
|
|
await db_session.commit()
|
|
await db_session.refresh(video)
|
|
assert video.media_type == "youtube_show"
|
|
assert video.youtube_video_id == "dQw4w9WgXcQ"
|
|
assert video.channel_name == "TechChannel"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_polymorphic_query_returns_correct_types(db_session: AsyncSession):
|
|
db_session.add(Movie(title="Movie A", s3_key="media/a.mp4", is_published=True))
|
|
db_session.add(TvSeries(title="Episode B", s3_key="media/b.mp4", is_published=True))
|
|
db_session.add(YoutubeShow(title="Video C", s3_key="media/c.mp4", is_published=True))
|
|
await db_session.commit()
|
|
|
|
from sqlalchemy import select
|
|
|
|
result = await db_session.execute(select(MediaItem))
|
|
items = result.scalars().all()
|
|
assert len(items) == 3
|
|
types = {i.media_type for i in items}
|
|
assert types == {"movie", "tv_series", "youtube_show"}
|
|
assert any(isinstance(i, Movie) for i in items)
|
|
assert any(isinstance(i, TvSeries) for i in items)
|
|
assert any(isinstance(i, YoutubeShow) for i in items)
|