"""Targeted offline Story checks: real rules, packs, modal, source lifetime and layout contract.""" import json import os import sys import threading import time import types import unittest from unittest import mock from pathlib import Path sys.path.insert(0, os.path.dirname(__file__)) import kodi_stubs kodi_stubs.install() import xbmcaddon from wcs.games import designs, game_settings from wcs.games.millionaire.subject import Subject from wcs.games.story.content import COUNTS, DEMO_DIR, ContentError, DemoSource, Unavailable, parse from wcs.games.story.engine import Engine from wcs.games.story.view import view MATRIX = Subject(title='Matrix', year=1999, tmdb_id='603') def raw(difficulty='medium'): return json.loads((Path(DEMO_DIR) / 'matrix' / (difficulty + '.json')).read_text()) def wait(predicate): deadline = time.monotonic() + 3 while not predicate(): if time.monotonic() >= deadline: raise AssertionError('Story did not settle') time.sleep(0.01) class PackAndEngineTests(unittest.TestCase): def test_three_authored_packs_finish_with_zero_one_and_two_mistakes(self): for difficulty, count in COUNTS.items(): for mistakes in ((), (1,), (1, count)): with self.subTest(difficulty=difficulty, mistakes=mistakes): engine = Engine(DemoSource('matrix').prepare(MATRIX, difficulty)) for number in range(1, count + 1): correct = engine.scene.correct self.assertTrue(engine.answer((correct + 1) % 4 if number in mistakes else correct)) self.assertFalse(engine.answer(correct)) self.assertEqual(engine.lives, 3 - len([n for n in mistakes if n <= number])) if number < count: self.assertTrue(engine.advance()) self.assertFalse(engine.advance()) self.assertEqual(engine.state, 'won') self.assertEqual(engine.answered, count) self.assertEqual(engine.correct, count - len(mistakes)) self.assertEqual(engine.scene.chapter, 5) def test_third_mistake_loses_early_or_on_final_scene_without_showing_ending(self): for final in (3, 10): engine = Engine(DemoSource('matrix').prepare(MATRIX, 'easy')) for number in range(1, final + 1): answer = engine.scene.correct engine.answer((answer + 1) % 4 if number in (1, 2, final) else answer) if number < final: engine.advance() self.assertEqual((engine.state, engine.lives), ('lost', 0)) values = view(engine, 'gameover', 'easy', str) self.assertTrue(values['Reveal']) self.assertEqual(values['Ending'], '') self.assertFalse(engine.advance()) def test_invalid_pack_is_rejected_not_repaired(self): for mutation in (lambda p: p['scenes'].pop(), lambda p: p['scenes'][0].pop('correct'), lambda p: p['scenes'][0].update(correct=True), lambda p: p['scenes'][0].update(chapter=2), lambda p: p['scenes'][0]['options'].update(B=p['scenes'][0]['options']['A']), lambda p: p['identity'].update(year=2003), lambda p: p['identity'].update(title='Titanic'), lambda p: p['identity'].update(episode_number=1), lambda p: p['identity'].update(episode_title='A different episode'), lambda p: p['scenes'][0]['options'].update(A='A meaningful action that is much too long'), lambda p: p['scenes'][0].update(setup='[COLOR red]spoof')): pack = raw() mutation(pack) with self.assertRaises(ContentError): parse(pack, 'medium', MATRIX) def test_unknown_pack_and_offline_title_fail_honestly(self): with self.assertRaises(Unavailable): parse({'schema_version': 1, 'status': 'unavailable', 'reason': 'I do not know this work.'}, 'easy') with self.assertRaises(Unavailable): DemoSource('unknown-title').prepare(MATRIX, 'easy') def test_literal_brackets_are_allowed_in_nonrendered_identity(self): value = raw('easy') value['identity']['title'] = '[REC]' self.assertEqual(len(parse(value, 'easy', MATRIX._replace(title='[REC]')).scenes), 10) def test_view_never_exposes_answer_or_ending_before_evaluation(self): engine = Engine(DemoSource('matrix').prepare(MATRIX, 'medium')) values = view(engine, 'question', 'medium', str) for key in ('Reveal', 'Ending', 'Aftermath', 'RightLetter', 'RightText'): self.assertEqual(values[key], '', key) engine.answer((engine.scene.correct + 1) % 4) values = view(engine, 'feedback', 'medium', str) self.assertEqual(values['Chapter.1.Progress'], '1 / 3') self.assertEqual(values['Score'], '0 / 1') self.assertTrue(values['Reveal']) engine.close() self.assertIsNone(engine.pack) class ModalTests(unittest.TestCase): def setUp(self): self.settings = {'story_difficulty': '1'} settings = self.settings class Addon: getSetting = lambda self, key: settings.get(key, '') setSetting = lambda self, key, value: settings.__setitem__(key, value) getAddonInfo = lambda self, key: kodi_stubs.ADDON_DIR if key == 'path' else 'plugin.video.milionar' getLocalizedString = lambda self, key: str(key) self.patch = mock.patch.object(xbmcaddon, 'Addon', lambda *args: Addon()) self.patch.start() self.panels = [] def tearDown(self): for panel in self.panels: panel.close() self.patch.stop() def panel(self, source=None, subject=MATRIX, mode='story', demo='matrix'): from wcs.games.DialogMilionarGame import MilionarGameModal from wcs.games.content.demo import DemoContent from wcs.games.demo import catalog panel = MilionarGameModal(title=subject.title, year=subject.year, media_type=subject.media_type, subject=subject, mode=mode, demo_id=demo, design=designs.MODERN, story_source=source, content=DemoContent(catalog.get('matrix'), pace=0)) panel._delayed_focus = lambda control, delay=0: panel.setFocusId(control) panel.onInit() self.panels.append(panel) return panel def test_real_modal_demo_all_difficulties_and_replay_are_silent(self): from wcs.games import millionaire_tts for level, count in enumerate((10, 15, 20)): self.settings['story_difficulty'] = str(level) panel = self.panel() with mock.patch.object(millionaire_tts, 'play_tts') as speech: panel.onClick(9601) wait(lambda: panel._adapter.screen() == 'question') original = panel._adapter.engine.pack for number in range(count): engine = panel._adapter.engine panel.onClick(9790 + engine.scene.correct) self.assertEqual(panel.getProperty('WCS.Story.AnsweredCount'), str(number + 1)) if number < count - 1: panel.onClick(9794) self.assertEqual(panel._adapter.screen(), 'win') self.assertTrue(panel.getProperty('WCS.Story.Ending')) panel.onClick(9794) wait(lambda: panel._adapter.screen() == 'question') self.assertEqual(panel._adapter.engine.pack, original) self.assertEqual(panel._adapter.engine.lives, 3) speech.assert_not_called() def test_error_retry_is_a_new_request_not_a_life_loss(self): class Source: name = 'demo' calls = 0 def prepare(self, subject, difficulty): self.calls += 1 if self.calls == 1: raise Unavailable('Unknown story') return DemoSource('matrix').prepare(subject, difficulty) source = Source() panel = self.panel(source) panel.onClick(9601) wait(lambda: panel._adapter.screen() == 'error') self.assertEqual(panel.getProperty('WCS.Story.Lives'), '3') panel.onClick(9784) wait(lambda: panel._adapter.screen() == 'question') self.assertEqual(source.calls, 2) def test_delayed_pack_cannot_reactivate_a_closed_attempt(self): released = threading.Event() finished = threading.Event() class Source: name = 'demo' def prepare(self, subject, difficulty): released.wait(2) pack = DemoSource('matrix').prepare(subject, difficulty) finished.set() return pack panel = self.panel(Source()) adapter = panel._adapter panel.onClick(9601) panel.close() released.set() self.assertTrue(finished.wait(2)) self.assertFalse(adapter._active) self.assertIsNone(adapter.engine) self.assertEqual(panel.getProperty('WCS.Story.Narration'), '') def test_mode_and_independent_difficulty_are_remembered(self): panel = self.panel() panel.onClick(9605) self.assertEqual(self.settings['game_last_mode'], 'millionaire') panel.onClick(9606) self.assertEqual(self.settings['game_last_mode'], 'story') panel.open_settings() panel.onClick(game_settings.CONTROLS['story_difficulty']) panel.close_settings() self.assertEqual(panel.getProperty('WCS.Story.QuestionCount'), '20') self.assertNotIn('millionaire_difficulty', self.settings) subject = Subject(title='Keanu Reeves', media_type='actor') other = self.panel(subject=subject, mode=None, demo='') self.assertEqual(other.mode, 'millionaire') self.assertEqual(self.settings['game_last_mode'], 'story') def test_native_modal_return_retires_and_releases_the_attempt(self): panel = self.panel() panel.onClick(9601) wait(lambda: panel._adapter.screen() == 'question') adapter = panel._adapter panel.doModal() # the stub returns as native Kodi closure would self.assertIsNone(panel._adapter) self.assertFalse(adapter._active) self.assertIsNone(adapter.engine) class AiPreparationTests(unittest.TestCase): def test_episode_and_counts_are_preserved_in_the_prompt(self): from wcs.games.story.prompts import build from wcs.games.millionaire.subject import query, from_params from urllib.parse import parse_qsl subject = Subject(title='A series', media_type='tv', year='2020', genre='Drama', plot='P', poster='p', backdrop='b', episode_title='The selected episode', season_number=2, episode_number=3) self.assertEqual(from_params(dict(parse_qsl(query(subject)))), subject) for difficulty, count in COUNTS.items(): system, request = build(subject, difficulty) self.assertIn('"season_number": 2', request) self.assertIn('"episode_number": 3', request) self.assertIn('"year": 2020', request) self.assertIn('Přesně {} scén'.format(count), request) self.assertIn('unavailable', system) def test_fresh_one_shot_request_and_local_answers_without_content_memory(self): from wcs.games.story.ai import AiSource sender = mock.Mock(return_value=('```json\n' + json.dumps(raw('easy')) + '\n```', {})) describe = mock.Mock() modules = {'wcs.ai.AIProvider': types.SimpleNamespace(send_prompt=sender), 'wcs.ai.UsageTracker': types.SimpleNamespace(describe_session=describe)} with mock.patch.dict(sys.modules, modules): source = AiSource() first = source.prepare(MATRIX, 'easy') engine = Engine(first) engine.answer(engine.scene.correct) engine.advance() self.assertEqual(sender.call_count, 1) source.prepare(MATRIX, 'easy') self.assertEqual(sender.call_count, 2) sessions = [call.kwargs['session_id'] for call in sender.call_args_list] self.assertNotEqual(*sessions) self.assertEqual(vars(source), {}) for call in sender.call_args_list: self.assertFalse(call.kwargs['use_tools']) self.assertTrue(call.kwargs['json_mode']) self.assertTrue(call.kwargs['private_content']) self.assertEqual(call.kwargs['max_output_tokens'], 16000) self.assertNotIn('request_timeout', call.kwargs) self.assertNotIn('messages', call.kwargs) self.assertIn('Přesně 10 scén', call.args[0]) def test_high_effort_budget_is_scoped_to_each_story_preparation(self): from wcs.games.story.ai import AiSource sender = mock.Mock(return_value=(json.dumps(raw('easy')), {})) modules = {'wcs.ai.AIProvider': types.SimpleNamespace(send_prompt=sender), 'wcs.ai.UsageTracker': types.SimpleNamespace(describe_session=mock.Mock())} with mock.patch.dict(sys.modules, modules): for provider, effort, extended in (('anthropic', 'low', False), ('anthropic', 'high', True), ('openai', 'medium', False), ('openai', 'high', True), ('openai', 'max', True), ('google', 'high', False)): with self.subTest(provider=provider, effort=effort): addon = mock.Mock() addon.getSetting.side_effect = lambda key: provider if key == 'ai_provider' else effort with mock.patch.object(xbmcaddon, 'Addon', return_value=addon): sender.reset_mock() AiSource().prepare(MATRIX, 'easy') self.assertEqual(sender.call_count, 1) self.assertEqual(sender.call_args.kwargs['max_output_tokens'], 32000 if extended else 16000) self.assertEqual(sender.call_args.kwargs.get('request_timeout'), 300 if extended else None) class VoiceOwnershipTests(unittest.TestCase): def test_cancelled_audio_generation_cannot_play_or_stop_the_background_video(self): import xbmc from wcs.games import millionaire_tts as tts from wcs.games.voice import TtsVoice jobs = [] class Thread: def __init__(self, target, **kwargs): jobs.append(target) def start(self): pass provider = types.SimpleNamespace(generate_tts=mock.Mock(return_value='/tmp/story-test.mp3')) settings = {'voice': 'test', 'provider': 'OpenAI', 'model': 'test'} voice = TtsVoice() with mock.patch.object(tts, '_load_tts_settings', return_value=settings), \ mock.patch.object(tts.threading, 'Thread', Thread), \ mock.patch.object(xbmc, 'executebuiltin') as builtin, \ mock.patch.dict(sys.modules, {'wcs.ai.TTSClient': provider}): tts.play_tts('Offline text; the provider is stubbed.', owner=voice) voice.stop() jobs[0]() builtin.assert_not_called() self.assertFalse(tts._tts_active.is_set()) def test_owned_stop_checks_the_actual_playing_file(self): import xbmc from wcs.games import millionaire_tts as tts from wcs.games.voice import TtsVoice voice = TtsVoice() for playing, expected in (('/tmp/video.mkv', False), ('/tmp/story.mp3', True)): tts._tts_owner, tts._tts_path = voice, '/tmp/story.mp3' with mock.patch.object(xbmc, 'getInfoLabel', return_value=playing), \ mock.patch.object(xbmc, 'executebuiltin') as builtin: voice.stop() self.assertEqual(builtin.called, expected) if __name__ == '__main__': unittest.main()