# -*- coding: utf-8 -*- """L2 unit tests: next_up, rollover, specials, unaired, gaps, collections (docs/plan/milionar-library/050, 120 ยง2).""" import json import os import sys import unittest from datetime import date HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.dirname(os.path.dirname(HERE))) from wcs.library2 import next_up as nu # noqa: E402 TODAY = date(2026, 6, 15) with open(os.path.join(HERE, 'fixtures', 'tmdb_fixture.json'), encoding='utf-8') as handle: FIX = json.load(handle) def order_of(show_id='900001', today=TODAY): show = FIX['tv'][show_id] seasons = {int(k.split(':')[1]): v for k, v in FIX['season'].items() if k.startswith(show_id + ':')} return nu.episode_order(show, seasons, today), show['status'] def st(playcount=0, resume=0.0, last=0): return {'playcount': playcount, 'resume_pos': resume, 'last_played_at': last} def watched(*pairs): return {pair: st(1) for pair in pairs} class OrderTests(unittest.TestCase): def test_specials_excluded_and_count(self): order, _status = order_of() self.assertEqual(len(order), 18) # 10 + 8, season 0 skipped self.assertEqual(order[0]['s'], 1) self.assertTrue(all(ep['s'] > 0 for ep in order)) def test_aired_flags_use_today(self): order, _status = order_of() aired = [ep for ep in order if ep['aired']] self.assertEqual(len(aired), 16) # last two episodes of S02 are unaired self.assertEqual((order[-1]['s'], order[-1]['e'], order[-1]['aired']), (2, 8, False)) later, _ = order_of(today=date(2026, 12, 31)) self.assertTrue(all(ep['aired'] for ep in later)) def test_missing_air_date_is_unaired(self): show = {'seasons': [{'season_number': 1}]} order = nu.episode_order(show, {1: {'episodes': [{'episode_number': 1}, {'episode_number': 2, 'air_date': '2020-01-01'}]}}, TODAY) self.assertEqual([ep['aired'] for ep in order], [False, True]) class NextUpTests(unittest.TestCase): def setUp(self): self.order, self.status = order_of() def up(self, states): return nu.next_up(self.order, states, self.status, TODAY) def test_no_activity(self): self.assertEqual(self.up({}), {'kind': None, 'episode': None}) def test_mid_season_next(self): r = self.up(watched((1, 1), (1, 2), (1, 3))) self.assertEqual((r['kind'], r['episode']['s'], r['episode']['e']), ('next', 1, 4)) def test_season_rollover(self): r = self.up(watched(*[(1, n) for n in range(1, 11)])) self.assertEqual((r['kind'], r['episode']['s'], r['episode']['e']), ('next', 2, 1)) def test_gap_follows_latest_watched_by_order(self): r = self.up(watched((1, 1), (1, 5))) # 2-4 skipped on purpose self.assertEqual((r['kind'], r['episode']['e']), ('next', 6)) def test_latest_by_order_not_by_time(self): states = {(1, 8): st(1, last=10), (1, 2): st(1, last=999)} # E2 watched later in time self.assertEqual(self.up(states)['episode']['e'], 9) def test_specials_do_not_count(self): # a watched special has no place in the order: no activity for the regular episodes self.assertEqual(self.up({(0, 1): st(1)})['kind'], None) def test_in_progress_beats_next(self): states = {**watched((1, 1), (1, 2)), (1, 3): st(0, 600.0, 50)} r = self.up(states) self.assertEqual((r['kind'], r['episode']['e']), ('resume', 3)) def test_newest_in_progress_wins(self): states = {(1, 3): st(0, 100.0, 10), (1, 7): st(0, 100.0, 20)} self.assertEqual(self.up(states)['episode']['e'], 7) def test_next_unaired_is_upcoming_with_date(self): r = self.up(watched(*[(1, n) for n in range(1, 11)], *[(2, n) for n in range(1, 7)])) self.assertEqual((r['kind'], r['episode']['s'], r['episode']['e']), ('upcoming', 2, 7)) self.assertEqual(r['episode']['air_date'], '2026-06-22') def test_returning_series_new_episode_aired_is_next_again(self): states = watched(*[(1, n) for n in range(1, 11)], *[(2, n) for n in range(1, 7)]) later, status = order_of(today=date(2026, 6, 23)) r = nu.next_up(later, states, status, date(2026, 6, 23)) self.assertEqual((r['kind'], r['episode']['e']), ('next', 7)) def test_ended_series_finished(self): order, status = order_of('900002') r = nu.next_up(order, watched((1, 1), (1, 2), (1, 3)), status, TODAY) self.assertEqual(r, {'kind': 'finished', 'episode': None}) def test_returning_series_caught_up_without_announcement(self): order = order_of()[0][:10] # only season 1 known r = nu.next_up(order, watched(*[(1, n) for n in range(1, 11)]), 'Returning Series', TODAY) self.assertEqual(r['kind'], 'caught_up') def test_series_watched_counts_aired_episodes_only(self): states = watched(*[(1, n) for n in range(1, 11)], *[(2, n) for n in range(1, 7)]) self.assertTrue(nu.series_watched(self.order, states)) self.assertFalse(nu.series_watched(self.order, watched((1, 1)))) self.assertFalse(nu.series_watched([], {})) def test_next_after_for_autoplay(self): r = nu.next_after(self.order, 1, 10, self.status) # as if S01E10 were watched self.assertEqual((r['kind'], r['episode']['s'], r['episode']['e']), ('next', 2, 1)) r = nu.next_after(self.order, 2, 6, self.status) self.assertEqual((r['kind'], r['episode']['e']), ('upcoming', 7)) ended, status = order_of('900002') self.assertEqual(nu.next_after(ended, 1, 3, status)['kind'], 'finished') self.assertEqual(nu.next_after(self.order, 0, 1, self.status)['kind'], 'caught_up') # a special def test_air_label(self): self.assertEqual(nu.air_label('2026-10-12'), '12. 10.') self.assertEqual(nu.air_label(''), '') class CollectionTests(unittest.TestCase): COLLECTION = FIX['collection']['900900'] def test_next_unwatched_released_by_date(self): part = nu.next_in_collection(self.COLLECTION, 900101, {}, TODAY) self.assertEqual(part['id'], 900102) def test_skips_watched_and_unreleased(self): part = nu.next_in_collection(self.COLLECTION, 900101, {900102: {'playcount': 1}}, TODAY) self.assertIsNone(part) # 900103 is not released yet def test_released_later_becomes_available(self): part = nu.next_in_collection(self.COLLECTION, 900102, {900101: {'playcount': 1}}, date(2027, 1, 1)) self.assertEqual(part['id'], 900103) def test_current_film_is_never_offered(self): part = nu.next_in_collection(self.COLLECTION, 900102, {}, TODAY) self.assertEqual(part['id'], 900101) # earlier unwatched film of the collection is "next" def test_none_when_collection_empty(self): self.assertIsNone(nu.next_in_collection({'parts': []}, 1, {}, TODAY)) class RefresherTests(unittest.TestCase): def setUp(self): import stubs import tempfile stubs.install() from wcs.library2 import store store.set_path(os.path.join(tempfile.mkdtemp(prefix='milionar-db-'), 'library.db')) def test_stale_tracked_series_are_refreshed_fresh_ones_skipped(self): from wcs.library2 import meta, refresher, store, tmdb store.set_in_list('tv', 11, True, {'id': 11, 'name': 'listed'}) # in My series, never cached store.mark_watched('tv', 22, 1, 1) # activity, cached and fresh store.mark_watched('tv', 33, 1, 1) # activity, cached but stale meta.put('tv:22', {'status': 'Ended', 'seasons': []}, 1000, now=5000) meta.put('tv:33', {'status': 'Returning Series', 'seasons': []}, 1000, now=1000) calls = [] def fetch(path, params): calls.append(path) if path.startswith('tv/') and '/season/' not in path: return {'id': int(path[3:]), 'status': 'Returning Series', 'seasons': [{'season_number': 0}, {'season_number': 1}]} return {'season_number': 1, 'episodes': []} self.assertEqual(sorted(refresher.due_series(now=5500)), [11, 33]) done = refresher.refresh_due(now=5500, fetch=fetch, playing=lambda: False) self.assertEqual(sorted(done), [11, 33]) self.assertEqual(sorted(calls), ['tv/11', 'tv/11/season/1', 'tv/33', 'tv/33/season/1']) # season 0 skipped self.assertEqual(meta.get('tv:33')[0]['status'], 'Returning Series') self.assertEqual(refresher.due_series(now=int(__import__('time').time())), [22]) # 22's cache aged out by wall clock only def test_not_while_playing(self): from wcs.library2 import refresher, store store.set_in_list('tv', 11, True, {'id': 11}) self.assertEqual(refresher.refresh_due(fetch=lambda p, q: None, playing=lambda: True), []) if __name__ == '__main__': unittest.main()