# -*- coding: utf-8 -*- """Pro tebe P1: signals, scoring, rows and the whole build on a fake TMDb (docs/plan/pro-tebe/010 §8). Run: python3 -m unittest scripts/library_harness/test_foryou.py -v (from the addon folder) """ import os import sys import tempfile import time import unittest from datetime import date HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) import stubs # noqa: E402 stubs.install() from wcs.foryou import api, availability, compose, engine, features as F, score as S, signals # noqa: E402 from wcs.library2 import store # noqa: E402 TODAY = date(2026, 10, 9) NOW = time.mktime(TODAY.timetuple()) + 20 * 3600 def fresh_db(): store.set_path(os.path.join(tempfile.mkdtemp(prefix='milionar-fy-'), 'library.db')) # --------------------------------------------------------------------------- fake TMDb world def title(media, tmdb_id, genres, year=2015, votes=2000, rating=7.2, lang='en', runtime=110, collection=None, backdrop=True, date_=None, recs=(), sims=()): data = {'id': tmdb_id, 'genre_ids': list(genres), 'genres': [{'id': g, 'name': 'G%d' % g} for g in genres], 'original_language': lang, 'vote_count': votes, 'vote_average': rating, 'popularity': 10.0, 'poster_path': '/p%d.jpg' % tmdb_id, 'backdrop_path': '/b%d.jpg' % tmdb_id if backdrop else None, 'overview': 'Plot %d' % tmdb_id, 'credits': {'cast': [{'id': 9000 + tmdb_id, 'name': 'Actor'}], 'crew': []}, 'keywords': {'keywords': [{'id': 77, 'name': 'robots'}]} if 878 in genres else {'keywords': []}, '_recs': list(recs), '_sims': list(sims)} when = date_ or '{}-03-01'.format(year) if media == 'movie': data.update({'title': 'Film %d' % tmdb_id, 'release_date': when, 'runtime': runtime}) if collection: data['belongs_to_collection'] = {'id': collection, 'name': 'Saga %d' % collection} else: data.update({'name': 'Serial %d' % tmdb_id, 'first_air_date': when, 'episode_run_time': [runtime], 'number_of_seasons': 3, 'created_by': [{'id': 8000 + tmdb_id, 'name': 'Maker'}]}) return data def world(): w = {} def add(media, *args, **kw): w[(media, args[0])] = title(media, *args, **kw) add('movie', 1, [878, 28], collection=500, recs=[('movie', i) for i in [2, 3] + list(range(4, 20)) + list(range(200, 240))]) add('movie', 2, [878, 28], year=2016, collection=500) add('movie', 3, [878, 28], year=2018, collection=500) for i in range(4, 20): add('movie', i, [878, 53], year=2000 + i, runtime=95 + i) for i in range(200, 240): # more of the taste add('movie', i, [878, 53], year=1990 + i % 30, runtime=170) add('tv', 10, [35], runtime=22, recs=[('tv', i) for i in range(11, 30)]) for i in range(11, 30): add('tv', i, [35, 18], year=1995 + i, runtime=25) add('movie', 40, [16, 10751], year=2026, date_='2026-06-01', votes=900, rating=8.0) # kids film, trending add('tv', 50, [10767], year=2026, votes=300) # talk show, trending add('movie', 41, [28], lang='ja', year=2026, date_='2026-05-01', votes=500) # foreign, trending add('movie', 60, [878], date_='2027-01-01') # not released yet add('movie', 61, [878], backdrop=False) # no backdrop add('movie', 70, [27], year=2020) # disliked (horror) for i in range(80, 100): # new releases add('movie', i, [878, 28], year=2026, date_='2026-0{}-15'.format(1 + i % 8), votes=300, rating=6.7, runtime=170) w[('movie', i)]['credits']['cast'].append({'id': 9001, 'name': 'Star of film 1'}) # in the taste for i in range(100, 170): # older gems add('movie', i, [878, 28], year=2005, votes=600, rating=7.8) return w GB = 1024 ** 3 MISSING = {'movie:4', 'movie:5', 'tv:11'} # not on Webshare at all ENGLISH = {'movie:6', 'tv:12'} # only English files SLOVAK = {'movie:7'} # Slovak dubbing only SUBS = {'movie:80', 'movie:81'} # new, Czech subtitles only (not for "Nově s CZ dabingem") LOCKED = {'movie:8'} # the only file is password-protected class FakeWebshare: """Answers the play dialog's queries like Webshare would for the fake world.""" def __init__(self, w): self.queries = [] self.by_query = {} for (media, i), d in w.items(): key = '{}:{}'.format(media, i) if key in MISSING: continue name = d.get('title') or d.get('name') lang = 'EN' if key in ENGLISH else 'SK dabing' if key in SLOVAK else 'CZ titulky' if key in SUBS else 'CZ dabing' if media == 'movie': year = d['release_date'][:4] query = '{} {}'.format(name, year) file = {'ident': 'm{}'.format(i), 'name': '{} ({}) {} 1080p.mkv'.format(name, year, lang), 'size': str(2 * GB), 'password': '1' if key in LOCKED else '0'} else: query = '{} S01E01'.format(name) file = {'ident': 't{}'.format(i), 'name': '{} S01E01 {} 720p.mkv'.format(name, lang), 'size': str(GB // 2), 'password': '0'} self.by_query.setdefault(query, []).append(file) def __call__(self, query): self.queries.append(query) return [dict(f) for f in self.by_query.get(query, [])] class FakeTMDb: def __init__(self, w): self.w = w self.calls = [] def _list(self, keys): out = [] for media, i in keys: d = dict(self.w[(media, i)]) d['media_type'] = media out.append(d) return {'results': out} def __call__(self, path, params): self.calls.append(path) parts = path.split('/') if parts[0] in ('movie', 'tv') and parts[1].isdigit(): d = dict(self.w.get((parts[0], int(parts[1]))) or {}) if not d: raise RuntimeError('404') d['recommendations'] = self._list(d.pop('_recs')) d['similar'] = self._list(d.pop('_sims')) return d if parts[0] == 'collection': return {'parts': [self.w[('movie', i)] for i in (3, 1, 2)]} if parts[0] == 'genre': return {'genres': [{'id': 878, 'name': 'Sci-fi'}, {'id': 35, 'name': 'Komedie'}]} if path.startswith('trending/movie'): return self._list([('movie', 40), ('movie', 41), ('movie', 5)] + [('movie', i) for i in range(80, 86)]) if path.startswith('trending/tv'): return self._list([('tv', 50), ('tv', 12), ('tv', 13)]) if path == 'discover/movie' and params.get('sort_by') == 'vote_average.desc': return self._list([('movie', i) for i in range(100, 170)]) if path == 'discover/movie' and 'primary_release_date.gte' in params: return self._list([('movie', i) for i in range(80, 100)]) if path.startswith('discover/movie'): return self._list([('movie', i) for i in (60, 61, 70, 7, 8)]) if path.startswith('discover/tv'): return self._list([('tv', i) for i in range(20, 26)]) if path.endswith('top_rated'): media = parts[0] return self._list([k for k in self.w if k[0] == media][:20]) return {'results': []} def seed_library(): store.set_in_list('movie', 1, True) store.mark_watched('movie', 1, when=NOW - 86400) store.set_in_list('tv', 10, True) for e in range(1, 6): store.mark_watched('tv', 10, season=1, episode=e, when=NOW - 3600) store.set_feedback('movie', 70, 'dislike', when=NOW) # --------------------------------------------------------------------------- tests class StoreFeedbackTests(unittest.TestCase): def setUp(self): fresh_db() def test_schema_has_feedback_table(self): conn = store.connect() tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} conn.close() self.assertIn('feedback', tables) def test_set_get_clear(self): store.set_feedback('movie', 5, 'like', when=10) store.set_feedback('tv', 6, 'hide', when=11) store.set_feedback('movie', 5, 'dislike', when=12) self.assertEqual(store.get_feedback(), {('movie', 5): ('dislike', 12), ('tv', 6): ('hide', 11)}) store.set_feedback('movie', 5, None) self.assertEqual(list(store.get_feedback()), [('tv', 6)]) with self.assertRaises(ValueError): store.set_feedback('movie', 5, 'love') class SignalTests(unittest.TestCase): def test_weights_and_exclusions(self): titles = {('movie', 1): {'in_list': 1, 'last_activity_at': NOW}, ('tv', 2): {'in_list': 1}} movies = {1: {'playcount': 1, 'resume_pos': 0, 'last_played_at': NOW}} seeds, exclude = signals.seeds(titles, movies, {3: (8, 1, NOW)}, {('movie', 4): NOW}, {('movie', 5): ('like', NOW), ('movie', 6): ('dislike', NOW), ('movie', 7): ('hide', NOW)}, NOW) by = {(s['media'], s['id']): s for s in seeds} self.assertEqual(by[('movie', 5)]['kind'], 'like') self.assertLess(by[('movie', 6)]['weight'], 0) self.assertNotIn(('movie', 7), by) # hidden: no taste, just never shown self.assertEqual(by[('movie', 1)]['kind'], 'watched') # watching beats the list entry self.assertEqual(by[('tv', 2)]['kind'], 'list') self.assertEqual(by[('movie', 4)]['kind'], 'tried') self.assertGreater(by[('tv', 3)]['weight'], by[('movie', 4)]['weight']) self.assertTrue({('movie', 1), ('tv', 2), ('tv', 3), ('movie', 5), ('movie', 6), ('movie', 7)} <= exclude) self.assertNotIn(('movie', 4), exclude) # a short try is not "seen" def test_a_film_opened_for_minutes_is_a_hint_not_a_seen_film(self): movies = {1: {'playcount': 0, 'resume_pos': 240, 'last_played_at': NOW}, 2: {'playcount': 0, 'resume_pos': 1200, 'last_played_at': NOW}} seeds, exclude = signals.seeds({}, movies, {9: (0, 1, NOW)}, {}, {}, NOW) by = {(s['media'], s['id']): s for s in seeds} self.assertEqual(by[('movie', 1)]['kind'], 'started') # 4 minutes: no "Protože jsi viděl" row self.assertEqual(by[('movie', 2)]['kind'], 'watched') # 20 minutes: seen self.assertEqual(by[('tv', 9)]['kind'], 'started') # one episode in progress self.assertTrue({('movie', 1), ('movie', 2), ('tv', 9)} <= exclude) def test_a_seek_is_not_watching(self): movies = {3: {'playcount': 0, 'resume_pos': 1179, 'last_played_at': NOW}} # 10 s played, sought to 19:39 seeds, _ = signals.seeds({}, movies, {}, {}, {}, NOW, watched_seconds={('movie', 3): 10}) self.assertEqual(seeds[0]['kind'], 'started') seeds, _ = signals.seeds({}, movies, {}, {}, {}, NOW, watched_seconds={('movie', 3): 1100}) self.assertEqual(seeds[0]['kind'], 'watched') def test_old_watching_fades(self): movies = {1: {'playcount': 1, 'last_played_at': NOW - 3 * 365 * 86400}, 2: {'playcount': 1, 'last_played_at': NOW}} seeds, _ = signals.seeds({}, movies, {}, {}, {}, NOW) by = {s['id']: s['weight'] for s in seeds} self.assertGreater(by[2], by[1] * 1.5) class ScoreTests(unittest.TestCase): def test_match_percent_range(self): self.assertEqual(S.match_percent(10, 10), 98) self.assertLess(S.match_percent(5, 10), S.match_percent(8, 10)) self.assertGreaterEqual(S.match_percent(0, 10), 58) self.assertEqual(S.match_percent(1, 0), 0) def test_penalties(self): taste = S.taste_of({'g:878': 1.0, 'dec:2010': 1.0}) base = F.card(title('movie', 1, [878]), 'movie') self.assertEqual(S.penalty(base, taste), 1.0) self.assertEqual(S.penalty(F.card(title('tv', 2, [10767]), 'tv'), taste), 0.0) self.assertLess(S.penalty(F.card(title('movie', 3, [16]), 'movie'), taste), 1.0) self.assertLess(S.penalty(F.card(title('movie', 4, [878], lang='ko'), 'movie'), taste), 1.0) self.assertLess(S.penalty(F.card(title('movie', 5, [878], year=1962), 'movie'), taste), 0.6) def test_tv_genres_fold_onto_film_genres(self): feats = F.features(title('tv', 1, [10765]), 'tv') self.assertIn('g:878', feats) self.assertIn('g:14', feats) def test_jitter_is_stable_per_day(self): self.assertEqual(S.jitter('2026-10-09', 'movie:1'), S.jitter('2026-10-09', 'movie:1')) self.assertNotEqual(S.jitter('2026-10-09', 'movie:1'), S.jitter('2026-10-10', 'movie:1')) class ComposeTests(unittest.TestCase): def test_slots(self): self.assertEqual([compose.slot_of(h) for h in (6, 12, 20, 23, 2)], ['morning', 'afternoon', 'evening', 'night', 'night']) def test_tonight_fits_runtime(self): film = {'media': 'movie', 'runtime': 120, 'genres': [878]} self.assertTrue(compose._fits_slot(film, 'evening')) self.assertFalse(compose._fits_slot(film, 'morning')) self.assertFalse(compose._fits_slot({'media': 'movie', 'runtime': 0, 'genres': []}, 'evening')) family = {'media': 'movie', 'runtime': 100, 'genres': [16, 10751]} self.assertFalse(compose._fits_slot(family, 'evening')) # grown-ups' evening ... self.assertTrue(compose._fits_slot(family, 'evening', kids_share=0.8)) # ... unless the family watches them class BuildTests(unittest.TestCase): def setUp(self): fresh_db() self.world = world() self.fake = FakeTMDb(self.world) api.FETCH = self.fake self.webshare = FakeWebshare(self.world) availability.SEARCH = self.webshare def tearDown(self): api.FETCH = None availability.SEARCH = None def test_full_build(self): seed_library() snap = engine.build(now=NOW, hour=20, today=TODAY) rows = snap['modes']['all'] kinds = [r['kind'] for r in rows] self.assertEqual(kinds[0], 'for_you') self.assertIn('because', kinds) self.assertIn('new', kinds) self.assertIn('gems', kinds) self.assertTrue(snap['personal']) on_page = [k for r in rows if r['kind'] != 'top10' for k in r['items']] self.assertEqual(len(on_page), len(set(on_page)), 'a title appears once per page (Top 10 apart)') everything = {k for r in rows for k in r['items']} for gone in ('movie:1', 'tv:10', 'movie:60', 'movie:61', 'movie:70', 'tv:50', 'movie:3'): self.assertNotIn(gone, everything, gone) self.assertIn('movie:2', everything) # the next part of the watched saga ... self.assertNotIn('movie:3', everything) # ... not the one after it self.assertEqual(snap['items']['movie:2']['reason']['kind'], 'because_watched') self.assertEqual(snap['items']['movie:2']['reason']['seed'], 'movie:1') because = next(r for r in rows if r['kind'] == 'because') self.assertEqual(because['reason']['kind'], 'because_watched') self.assertIn(because['seed'], ('movie:1', 'tv:10')) for r in snap['modes']['movie']: self.assertTrue(all(k.startswith('movie:') for k in r['items'])) for r in snap['modes']['tv']: self.assertTrue(all(k.startswith('tv:') for k in r['items'])) for key in everything: item = snap['items'][key] self.assertTrue(58 <= item['match'] <= 98, (key, item['match'])) self.assertTrue(item['poster'] and item['backdrop']) gems = next(r for r in rows if r['kind'] == 'gems') self.assertTrue(all(snap['items'][k]['year'] < 2025 for k in gems['items'])) def test_cached_rebuild_is_offline(self): seed_library() engine.build(now=NOW, hour=20, today=TODAY) before = len(self.fake.calls) engine.build(now=NOW, hour=20, today=TODAY) self.assertEqual(len(self.fake.calls), before) def test_cold_start(self): snap = engine.build(now=NOW, hour=20, today=TODAY) self.assertFalse(snap['personal']) items = [snap['items'][k] for r in snap['modes']['all'] for k in r['items']] self.assertTrue(items) self.assertTrue(all(i['match'] == 0 for i in items)) def test_save_and_load(self): seed_library() snap = engine.build(now=NOW, hour=20, today=TODAY) engine.save(snap) loaded = engine.load() self.assertEqual(loaded['modes']['all'][0]['items'], snap['modes']['all'][0]['items']) def test_only_playable_titles_are_shown(self): seed_library() snap = engine.build(now=NOW, hour=20, today=TODAY) self.assertTrue(snap['verified']) shown = {k for r in snap['modes']['all'] for k in r['items']} for gone in MISSING | ENGLISH | LOCKED: self.assertNotIn(gone, shown, gone) for key in shown: self.assertIn(snap['items'][key]['lang'], ('cz_dub', 'cz_sub', 'sk_dub'), key) self.assertEqual(snap['items'].get('movie:7', {}).get('lang', 'sk_dub'), 'sk_dub') new = next(r for r in snap['modes']['all'] if r['kind'] == 'new') self.assertTrue(all(snap['items'][k]['lang'] == 'cz_dub' for k in new['items'])) # "s CZ dabingem" def test_availability_is_cached(self): seed_library() engine.build(now=NOW, hour=20, today=TODAY) asked = len(self.webshare.queries) self.assertGreater(asked, 0) engine.build(now=NOW, hour=20, today=TODAY) self.assertEqual(len(self.webshare.queries), asked) def test_series_are_checked_on_their_first_episode(self): target, queries, _fallback = availability.plan({'media': 'tv', 'title': 'Přátelé', 'original_title': 'Friends'}) self.assertEqual((target['season'], target['episode']), (1, 1)) self.assertEqual(queries, ['Přátelé S01E01', 'Friends S01E01', 'Přátelé 1x01']) target, queries, fallback = availability.plan({'media': 'movie', 'title': 'Matrix', 'original_title': 'The Matrix', 'year': 1999}) self.assertEqual(queries, ['Matrix 1999', 'The Matrix 1999']) self.assertEqual(fallback, ['Matrix']) def test_ambiguous_czech_titles_need_the_original_or_the_year(self): card = {'media': 'tv', 'id': 93290, 'title': 'Rodiče', 'original_title': 'Breeders', 'year': 2020} files = {'Rodiče S01E01': [{'ident': 'a', 'name': 'Rodiče.01x01.1080i.HEVC.HDTV.CZ.mkv', 'size': str(GB)}], 'Breeders S01E01': []} availability.SEARCH = lambda q: [dict(f) for f in files.get(q, [])] self.assertEqual(availability.check(card, 'x')['status'], 'missing') # could be another "Rodiče" files['Breeders S01E01'] = [{'ident': 'b', 'name': 'Breeders S01E01 CZ dabing 1080p.mkv', 'size': str(GB)}] self.assertEqual(availability.check(card, 'x')['lang'], 'cz_dub') film = {'media': 'movie', 'id': 251, 'title': 'Duch', 'original_title': 'Ghost', 'year': 1990} files['Duch 1990'] = [{'ident': 'c', 'name': 'Duch (1990) CZ dabing 1080p.mkv', 'size': str(2 * GB)}] self.assertEqual(availability.check(film, 'x')['status'], 'ok') # the year confirms it def test_no_webshare_account_shows_everything_unverified(self): availability.SEARCH = None # the stub addon has no Webshare account seed_library() snap = engine.build(now=NOW, hour=20, today=TODAY) self.assertFalse(snap['verified']) self.assertTrue(snap['modes']['all']) def test_too_few_playable_titles_keep_the_old_snapshot(self): availability.SEARCH = lambda query: [] seed_library() with self.assertRaises(engine.NotEnough): engine.build(now=NOW, hour=20, today=TODAY) def test_errors_are_not_cached(self): def broken(path, params): raise RuntimeError('offline') api.FETCH = broken self.assertIsNone(api.get('trending/movie/week')) api.FETCH = self.fake self.assertTrue(api.get('trending/movie/week')['results']) if __name__ == '__main__': unittest.main()