tboudreaux/PYTV
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(main): main

Browse Source
This commit is contained in:
Emily Boudreaux
2026-03-10 08:39:28 -04:00
parent b1a93161c0
commit af3076342a
18 changed files with 826 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
tests/test_channel_actions.py Normal file
View File

@@ -0,0 +1,89 @@
import pytest
import datetime
from django.utils import timezone
from core.models import Channel, AppUser, Library, MediaSource, ChannelSourceRule, ScheduleTemplate, ScheduleBlock
@pytest.mark.django_db
def test_channel_status_and_download_flow(client):
# 1. Setup mock user, library, channel
user = AppUser.objects.create(username="testuser")
library = Library.objects.create(name="Test Library", owner_user=user)
channel = Channel.objects.create(
name="Action Channel",
slug="action-ch",
owner_user=user,
library=library,
scheduling_mode="fill_blocks"
)
# 2. Add the Solid Color source (this requires no real downloading)
source = MediaSource.objects.create(
library=library,
name="Color Test",
source_type="solid_color"
)
from core.models import MediaItem
MediaItem.objects.create(
media_source=source,
title="Test Solid Color Item",
item_kind="movie",
runtime_seconds=3600,
file_path="dummy_path"
)
# 3. Create a schedule template and a 24/7 block
template = ScheduleTemplate.objects.create(
channel=channel,
name="Daily",
timezone_name="UTC",
is_active=True
)
block = ScheduleBlock.objects.create(
schedule_template=template,
name="All Day Block",
block_type="programming",
start_local_time=datetime.time(0, 0),
end_local_time=datetime.time(23, 59),
day_of_week_mask=127
)
# Map the solid_color source to the channel
ChannelSourceRule.objects.create(
channel=channel,
media_source=source,
rule_mode="allow",
weight=1.0,
schedule_block_label="All Day Block"
)
# 4. Check initial status (should be 0 airings)
resp = client.get(f"/api/channel/{channel.id}/status")
assert resp.status_code == 200
data = resp.json()
assert data["total_upcoming_airings"] == 0
# 5. Generate schedule for today
resp = client.post(f"/api/schedule/generate-today/{channel.id}")
assert resp.status_code == 200
assert resp.json()["status"] == "success"
assert resp.json()["airings_created"] > 0
# 6. Check status again (airings exist, but might not be cached yet depending on source type logic)
resp = client.get(f"/api/channel/{channel.id}/status")
assert resp.status_code == 200
data = resp.json()
assert data["total_upcoming_airings"] > 0
# 7. Trigger the channel download endpoint
resp = client.post(f"/api/channel/{channel.id}/download")
assert resp.status_code == 200
cache_data = resp.json()
assert "downloaded" in cache_data
# 8. Final status check
resp = client.get(f"/api/channel/{channel.id}/status")
assert resp.status_code == 200
final_data = resp.json()
# Solid colors don't need real downloads, but we ensure the API reported successfully
assert "percent_cached" in final_data

72
tests/test_frontend_reload.py Normal file
View File

@@ -0,0 +1,72 @@
import asyncio
from playwright.async_api import async_playwright
import sys
import time
async def main():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
url = "http://localhost:5173/"
print("Testing frontend playback sync consistency on reload...")
times = []
wall_times = []
# Intercept console messages
page.on("console", lambda msg: print(f"BROWSER LOG: {msg.text}"))
for i in range(5):
start_wall = time.time()
await page.goto(url)
# Wait for the video element and for it to begin playing
try:
await page.wait_for_selector('video.tuner-video', state='attached', timeout=5000)
# wait 1 second to let metadata and ref execute
await page.wait_for_timeout(1000)
current_time = await page.evaluate("() => { const v = document.querySelector('video.tuner-video.playing'); return v ? v.currentTime : null; }")
current_src = await page.evaluate("() => { const v = document.querySelector('video.tuner-video.playing'); return v ? v.currentSrc : null; }")
if current_time is None:
print(f"Reload {i+1}: Video element found but currentTime is null")
continue
times.append(current_time)
wall_times.append(time.time() - start_wall)
print(f"Reload {i+1}: src = {current_src}, currentTime = {current_time:.2f} seconds")
except Exception as e:
print(f"Reload {i+1}: Error - {e}")
await browser.close()
if not times:
print("No times recorded. Ensure the frontend and backend are running.")
sys.exit(1)
diffs = [times[i] - times[i-1] for i in range(1, len(times))]
print("Differences in video time between loads:", [f"{d:.2f}s" for d in diffs])
# If the video restarts on reload, the current time will always be ~1.0 seconds (since we wait 1000ms).
# Normal behavior should have the time incrementing by the wall clock duration of the reload loop.
bug_present = False
for i, t in enumerate(times):
if t < 2.5: # Should definitely be higher than 2.5 after a few reloads if it started > 0, OR if it started near 0, subsequent ones should go up.
pass
# Actually, let's just check if the sequence of times is monotonically increasing
# given that the item is supposed to be continuous wall-clock time
is_monotonic = all(times[i] > times[i-1] for i in range(1, len(times)))
if not is_monotonic:
print("TEST FAILED: BUG REPRODUCED. The currentTime is not continuous across reloads.")
sys.exit(1)
else:
print("TEST PASSED: The currentTime is continuous across reloads.")
sys.exit(0)
if __name__ == "__main__":
asyncio.run(main())

98
tests/test_playback_sync.py Normal file
View File

@@ -0,0 +1,98 @@
import pytest
from django.utils import timezone
from datetime import timedelta
from core.models import MediaSource, MediaItem, Channel, ScheduleBlock, ChannelSourceRule, Airing, AppUser, Library
from api.routers.channel import AiringSchema
from django.test import Client
@pytest.fixture
def test_channel(db):
user = AppUser.objects.create(username="playback_tester")
library = Library.objects.create(name="Test Library", owner_user=user)
channel = Channel.objects.create(name="Test Channel", channel_number=10, owner_user=user, library=library)
source = MediaSource.objects.create(library=library, name="Test Source", source_type="solid_color")
ChannelSourceRule.objects.create(channel=channel, media_source=source, rule_mode='allow', weight=1.0)
# Create media item
item = MediaItem.objects.create(
media_source=source,
title="Color Loop",
item_kind="solid_color",
runtime_seconds=600 # 10 minutes length
)
return channel, item
import uuid
@pytest.mark.django_db
def test_playback_offset_calculation(test_channel):
channel, item = test_channel
now = timezone.now()
# Airing started exactly 3 minutes ago
starts_at = now - timedelta(minutes=3)
ends_at = starts_at + timedelta(minutes=30)
airing = Airing.objects.create(
channel=channel,
media_item=item,
starts_at=starts_at,
ends_at=ends_at,
slot_kind='content',
generation_batch_uuid=uuid.uuid4()
)
schema = AiringSchema.from_airing(airing)
# Airing started 3 minutes (180 seconds) ago.
# The item runtime is 600s, so offset is simply 180s.
assert schema.exact_playback_offset_seconds == pytest.approx(180.0, rel=1e-2)
@pytest.mark.django_db
def test_playback_offset_modulo(test_channel):
channel, item = test_channel
now = timezone.now()
# Airing started exactly 14 minutes ago
starts_at = now - timedelta(minutes=14)
ends_at = starts_at + timedelta(minutes=30)
airing = Airing.objects.create(
channel=channel,
media_item=item,
starts_at=starts_at,
ends_at=ends_at,
slot_kind='content',
generation_batch_uuid=uuid.uuid4()
)
schema = AiringSchema.from_airing(airing)
# 14 minutes = 840 seconds.
# Video is 600 seconds.
# 840 % 600 = 240 seconds offset into the second loop.
assert schema.exact_playback_offset_seconds == pytest.approx(240.0, rel=1e-2)
@pytest.mark.django_db
def test_channel_now_api_returns_offset(test_channel, client: Client):
channel, item = test_channel
now = timezone.now()
starts_at = now - timedelta(seconds=45)
ends_at = starts_at + timedelta(minutes=10)
Airing.objects.create(
channel=channel,
media_item=item,
starts_at=starts_at,
ends_at=ends_at,
slot_kind='content',
generation_batch_uuid=uuid.uuid4()
)
response = client.get(f"/api/channel/{channel.id}/now")
assert response.status_code == 200
data = response.json()
assert "exact_playback_offset_seconds" in data
assert data["exact_playback_offset_seconds"] == pytest.approx(45.0, rel=1e-1)

116
tests/test_source_rules.py Normal file
View File

@@ -0,0 +1,116 @@
import pytest
from datetime import date, timedelta, datetime, timezone
from django.utils import timezone as django_timezone
from core.models import AppUser, Library, Channel, ScheduleTemplate, ScheduleBlock, MediaSource, MediaItem, ChannelSourceRule, Airing
from core.services.scheduler import ScheduleGenerator
@pytest.mark.django_db
def test_repeat_gap_hours():
# Setup user, library, channel
user = AppUser.objects.create_user(username="rule_tester", password="pw")
library = Library.objects.create(owner_user=user, name="Rule Test Lib")
channel = Channel.objects.create(
owner_user=user, library=library, name="Rule TV", slug="r-tv", channel_number=99, timezone_name="UTC"
)
source = MediaSource.objects.create(
library=library,
name="Gap Source",
source_type="local_directory",
uri="/tmp/fake",
min_repeat_gap_hours=5
)
# Create 6 items (each 1 hour long) -> 6 hours total duration
# This means a 5 hour gap is mathematically satisfiable without breaking cooldowns.
for i in range(6):
MediaItem.objects.create(
media_source=source,
title=f"Vid {i}",
item_kind="movie",
runtime_seconds=3600, # 1 hour
file_path=f"/tmp/fake/{i}.mp4",
cached_file_path=f"/tmp/fake/{i}.mp4",
is_active=True
)
ChannelSourceRule.objects.create(channel=channel, media_source=source, rule_mode="allow", weight=1.0)
template = ScheduleTemplate.objects.create(channel=channel, name="T", priority=10, is_active=True)
block = ScheduleBlock.objects.create(
schedule_template=template, name="All Day", block_type="programming",
start_local_time="00:00:00", end_local_time="12:00:00", day_of_week_mask=127
)
target_date = django_timezone.now().date()
gen = ScheduleGenerator(channel)
created = gen.generate_for_date(target_date)
assert created == 12 # 12 1-hour slots
airings = list(Airing.objects.filter(channel=channel).order_by('starts_at'))
# Assert that NO item is repeated within 5 hours of itself
last_played = {}
for a in airings:
if a.media_item_id in last_played:
gap = (a.starts_at - last_played[a.media_item_id]).total_seconds() / 3600.0
assert gap >= 5.0, f"Item {a.media_item_id} played too soon. Gap: {gap} hours"
last_played[a.media_item_id] = a.starts_at
@pytest.mark.django_db
def test_youtube_age_filter():
from unittest.mock import patch
from core.services.youtube import sync_source
user = AppUser.objects.create_user(username="rule_tester2", password="pw")
library = Library.objects.create(owner_user=user, name="Rule Test Lib 2")
source = MediaSource.objects.create(
library=library,
name="Old Videos Only",
source_type="youtube_playlist",
uri="https://www.youtube.com/playlist?list=fake_playlist",
max_age_days=10 # Anything older than 10 days is skipped
)
# Mock yt-dlp extract_info to return 3 videos
# 1. 2 days old (keep)
# 2. 15 days old (skip)
# 3. 5 days old (keep)
from datetime import date, timedelta
today = date.today()
with patch("core.services.youtube.yt_dlp.YoutubeDL") as mock_ydl:
mock_instance = mock_ydl.return_value.__enter__.return_value
mock_instance.extract_info.return_value = {
"entries": [
{
"id": "vid1",
"title": "New Video",
"duration": 600,
"upload_date": (today - timedelta(days=2)).strftime("%Y%m%d")
},
{
"id": "vid2",
"title": "Old Video",
"duration": 600,
"upload_date": (today - timedelta(days=15)).strftime("%Y%m%d")
},
{
"id": "vid3",
"title": "Medium Video",
"duration": 600,
"upload_date": (today - timedelta(days=5)).strftime("%Y%m%d")
}
]
}
res = sync_source(source, max_videos=5)
assert res['created'] == 2 # Only vid1 and vid3
assert res['skipped'] == 1 # vid2 skipped due to age
# Verify the items in DB
items = list(MediaItem.objects.filter(media_source=source).order_by('title'))
assert len(items) == 2
assert "vid2" not in [item.title for item in items]