#!/usr/bin/env python3
"""Offline checks of the Home warm-up thread (wcs/skin/warm.py): batches, gates, skipped sections, change requests.

Run: python3 scripts/skin_harness/test_warm.py
"""
from pathlib import Path
import sys
import unittest
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / 'scripts/library_harness'))
import stubs
stubs.install()
import xbmc
import xbmcgui
from wcs.skin import warm


class FakeMonitor:
    def __init__(self):
        self.aborted = False

    def abortRequested(self):
        return self.aborted

    def waitForAbort(self, _seconds):
        return self.aborted


class WarmTests(unittest.TestCase):
    def setUp(self):
        xbmcgui.Window.props.clear()
        self.labels = {'Container(9000).ListItem.Property(id)': 'mil_foryou'}
        self.conditions = {'Window.IsActive(home)': True}
        for section in warm.SECTIONS:
            for number in (1, 2):
                self.labels['Skin.String(mil.{}.{}.type)'.format(section, number)] = 'row'
        self.patches = [
            patch.object(xbmc, 'getInfoLabel', side_effect=lambda label: self.labels.get(label, '')),
            patch.object(xbmc, 'getCondVisibility', side_effect=lambda cond: self.conditions.get(cond, False)),
        ]
        for item in self.patches:
            item.start()
            self.addCleanup(item.stop)
        self.monitor = FakeMonitor()
        # the lists answer "not updating" at once; the wait loops then end after MIN_DWELL + STABLE_POLLS

    def spy(self):
        seen = []
        original = xbmcgui.Window.setProperty

        def record(window, key, value):
            seen.append((key, value))
            return original(window, key, value)
        return seen, patch.object(xbmcgui.Window, 'setProperty', record)

    def test_a_batch_opens_the_gates_shows_the_groups_together_and_clears_them_again(self):
        seen, spy = self.spy()
        with spy:
            result = warm._warm(self.monitor, ['my_movies', 'my_series'])
        self.assertEqual(result, {'my_movies': warm.DONE, 'my_series': warm.DONE})
        for section, menu in (('my_movies', 'mil_mymovies'), ('my_series', 'mil_myseries')):
            self.assertIn(('milionar.go.' + section, '1'), seen)
            self.assertIn(('milionar.warm.' + menu, '1'), seen)
            self.assertIn(('milionar.warm.quiet.' + menu, '1'), seen)
            self.assertEqual(xbmcgui.Window(10000).getProperty('milionar.warm.' + menu), '')        # cleared again
            self.assertEqual(xbmcgui.Window(10000).getProperty('milionar.warm.quiet.' + menu), '')

    def test_hidden_menu_item_and_focused_section_are_not_shown(self):
        self.conditions['Skin.HasSetting(HomeMenuNoMilMyTVButton)'] = True
        self.labels['Container(9000).ListItem.Property(id)'] = 'mil_vinyl'
        seen, spy = self.spy()
        with spy:
            result = warm._warm(self.monitor, ['mytv', 'vinyl'])
        self.assertEqual(result, {'mytv': warm.DONE, 'vinyl': warm.DONE})
        self.assertEqual(xbmcgui.Window(10000).getProperty('milionar.go.mytv'), '')    # never opened: nothing to load
        self.assertFalse([key for key, _value in seen if key.startswith('milionar.warm.')])   # the user sees it: no trick

    def test_interrupted_by_playback_is_retried(self):
        self.conditions['Player.HasMedia'] = True
        self.assertEqual(warm._warm(self.monitor, ['my_series']), {'my_series': warm.RETRY})
        self.assertEqual(xbmcgui.Window(10000).getProperty('milionar.warm.mil_myseries'), '')

    def test_the_section_on_screen_goes_first(self):
        self.conditions['Container(32100).IsUpdating'] = True          # Pro tebe (focused) is loading
        self.assertEqual(warm._warm(self.monitor, ['my_movies']), {'my_movies': warm.RETRY})

    def test_a_section_that_keeps_loading_gives_way(self):
        self.conditions['Container(29100).IsUpdating'] = True
        self.assertEqual(warm._warm(self.monitor, ['my_movies', 'my_series']),
                         {'my_movies': warm.LATER, 'my_series': warm.DONE})

    def test_requests_are_merged_and_consumed_once(self):
        warm.request({'my_series', 'unknown', 'tmdb'})
        warm.request({'tmdb', 'cont'})
        self.assertEqual(sorted(warm._take_requests()), ['cont', 'my_series', 'tmdb'])
        self.assertEqual(warm._take_requests(), [])

    def test_section_table_matches_gen_slots(self):
        sys.path.insert(0, str(ROOT / 'scripts/skin_harness'))
        import gen_slots
        for section, (menu_id, _flag, base) in warm.SECTIONS.items():
            self.assertEqual(gen_slots.SECTIONS[section], base, section)
            self.assertEqual(gen_slots.MENU_IDS[section], menu_id, section)
        self.assertEqual(set(warm.SECTIONS), set(gen_slots.SECTIONS))

    def test_home_reads_the_properties_the_thread_sets(self):
        home = (ROOT / 'resources/skin.milionar/xml/Home.xml').read_text(encoding='utf-8')
        animations = (ROOT / 'resources/skin.milionar/xml/Includes_Animations.xml').read_text(encoding='utf-8')
        for menu_id, _flag, _base in warm.SECTIONS.values():
            self.assertIn('Window(Home).Property({}{})'.format(warm.WARM_PROPERTY, menu_id), home)
        self.assertIn('Window(Home).Property({}$PARAM[id])'.format(warm.QUIET_PROPERTY), animations)


if __name__ == '__main__':
    unittest.main()