path: root/tests/test_mpv.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# vim: ts=4 sw=4 et
#
# Python MPV library module
# Copyright (C) 2017-2022 Sebastian Götte <code@jaseg.net>
#
# This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public
# License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later
# version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with this program; if not, write to the Free
# Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#

import unittest
from unittest import mock
import math
import threading
from contextlib import contextmanager
from functools import wraps
import gc
import os.path
import os
import sys
import time
import io
import platform
import ctypes
from concurrent.futures import Future

os.environ["PATH"] = os.path.dirname(__file__) + os.pathsep + os.environ["PATH"]

import mpv


if os.name == 'nt':
  Xvfb = mock.Mock()
  testvo='gpu'

else:
  from xvfbwrapper import Xvfb
  testvo='x11'


TESTVID = os.path.join(os.path.dirname(__file__), 'test.webm')
TESTSRT = os.path.join(os.path.dirname(__file__), 'sub_test.srt')
MPV_ERRORS = [ l(ec) for ec, l in mpv.ErrorCode.EXCEPTION_DICT.items() if l ]

def timed_print():
    start_time = time.time()
    def do_print(level, prefix, text):
        td = time.time() - start_time
        print('{:.3f} [{}] {}: {}'.format(td, level, prefix, text), flush=True)

class MpvTestCase(unittest.TestCase):
    def setUp(self):
        self.disp = Xvfb()
        self.disp.start()
        self.m = mpv.MPV(vo=testvo, loglevel='debug', log_handler=timed_print())

    def tearDown(self):
        self.m.terminate()
        self.disp.stop()

class TestProperties(MpvTestCase):
    @contextmanager
    def swallow_mpv_errors(self, exception_exceptions=[]):
        try:
            yield
        except Exception as e:
            if any(e.args[:2] == ex.args for ex in MPV_ERRORS):
                if e.args[1] not in exception_exceptions:
                    raise
            else:
                raise

    def test_read(self):
        self.m.loop = 'inf'
        self.m.play(TESTVID)
        while self.m.core_idle:
            time.sleep(0.05)
        for name in sorted(self.m.property_list):
            name = name.replace('-', '_')
            with self.subTest(property_name=name), self.swallow_mpv_errors([
                mpv.ErrorCode.PROPERTY_UNAVAILABLE,
                mpv.ErrorCode.PROPERTY_ERROR,
                mpv.ErrorCode.PROPERTY_NOT_FOUND]):
                getattr(self.m, name)

    def test_write(self):
        self.m.loop = 'inf'
        self.m.play(TESTVID)
        while self.m.core_idle:
            time.sleep(0.05)
        check_canaries = lambda: os.path.exists('100') or os.path.exists('foo')
        for name in sorted(self.m.property_list):
            # See issue #108 and upstream mpv issues #7919 and #7920.
            if name in ('demuxer', 'audio-demuxer', 'audio-files'):
                continue
            # These may cause files to be created
            if name in ('external-file', 'heartbeat-cmd', 'wid', 'dump-stats', 'log-file') or name.startswith('input-'):
                continue
            # Caues segmentation faults on wayland
            if name in ('current-window-scale',):
                continue
            name =  name.replace('-', '_')
            old_canaries = check_canaries()
            with self.subTest(property_name=name), self.swallow_mpv_errors([
                    mpv.ErrorCode.PROPERTY_UNAVAILABLE,
                    mpv.ErrorCode.PROPERTY_ERROR,
                    mpv.ErrorCode.PROPERTY_FORMAT,
                    mpv.ErrorCode.PROPERTY_NOT_FOUND]): # This is due to a bug with option-mapped properties in mpv 0.18.1
                    setattr(self.m, name, 100)
                    setattr(self.m, name, 1)
                    setattr(self.m, name, 0)
                    setattr(self.m, name, -1)
                    setattr(self.m, name, 1)
                    setattr(self.m, name, 1.0)
                    setattr(self.m, name, 0.0)
                    setattr(self.m, name, -1.0)
                    setattr(self.m, name, float('nan'))
                    setattr(self.m, name, 'foo')
                    setattr(self.m, name, '')
                    setattr(self.m, name, 'bazbazbaz'*1000)
                    setattr(self.m, name, b'foo')
                    setattr(self.m, name, b'')
                    setattr(self.m, name, b'bazbazbaz'*1000)
                    setattr(self.m, name, True)
                    setattr(self.m, name, False)
            if not old_canaries and check_canaries():
                raise UserWarning('Property test for {} produced files on file system, might not be safe.'.format(name))

    def test_property_bounce(self):
        self.m.aid = False
        self.assertEqual(self.m.audio, False)
        self.m.aid = 'auto'
        self.assertEqual(self.m.audio, 'auto')
        self.m.aid = 'no'
        self.assertEqual(self.m.audio, False)
        self.m.audio = 'auto'
        self.assertEqual(self.m.aid, 'auto')
        self.m.audio = False
        self.assertEqual(self.m.aid, False)
        self.m.audio = 'auto'
        self.assertEqual(self.m.aid, 'auto')
        self.m.audio = 'no'
        self.assertEqual(self.m.aid, False)

    def test_array_property_bounce(self):
        self.m.alang = 'en'
        self.assertEqual(self.m.alang, ['en'])
        self.m.alang = 'de'
        self.assertEqual(self.m.alang, ['de'])
        self.m.alang = ['de', 'en']
        self.assertEqual(self.m.alang, ['de', 'en'])
        self.m.alang = 'de,en'
        self.assertEqual(self.m.alang, ['de', 'en'])
        self.m.alang = ['de,en']
        self.assertEqual(self.m.alang, ['de,en'])

    def test_osd_property_bounce(self):
        self.m.alang = ['en']
        self.assertEqual(self.m.osd.alang, 'en')
        self.m.alang = ['de']
        self.assertEqual(self.m.osd.alang, 'de')
        self.m.alang = ['en', 'de']
        self.assertEqual(self.m.osd.alang, 'en,de')

    def test_raw_property_bounce(self):
        self.m.alang = 'en'
        self.assertEqual(self.m.raw.alang, [b'en'])
        self.m.alang = 'de'
        self.assertEqual(self.m.raw.alang, [b'de'])
        self.m.alang = ['de', 'en']
        self.assertEqual(self.m.raw.alang, [b'de', b'en'])
        self.m.alang = 'de,en'
        self.assertEqual(self.m.raw.alang, [b'de', b'en'])
        self.m.alang = ['de,en']
        self.assertEqual(self.m.raw.alang, [b'de,en'])

    def test_property_decoding_invalid_utf8(self):
        invalid_utf8 = b'foo\xc3\x28bar'
        self.m.alang = invalid_utf8
        self.assertEqual(self.m.raw.alang, [invalid_utf8])
        with self.assertRaises(UnicodeDecodeError):
            self.m.strict.alang
        with self.assertRaises(UnicodeDecodeError):
            # alang is considered safe and pasted straight into the OSD string. But OSD strings should always be valid
            # UTF-8. This test may be removed in case OSD encoding sanitization is handled differently in the future.
            self.m.osd.alang

    def test_property_decoding_valid_utf8(self):
        valid_utf8 = 'pröpérty'
        self.m.alang = valid_utf8
        self.assertEqual(self.m.alang, [valid_utf8])
        self.assertEqual(self.m.raw.alang, [valid_utf8.encode('utf-8')])
        self.assertEqual(self.m.osd.alang, valid_utf8)
        self.assertEqual(self.m.strict.alang, [valid_utf8])

    def test_property_decoding_multi(self):
        valid_utf8 = 'pröpérty'
        invalid_utf8 = b'foo\xc3\x28bar'
        self.m.alang = [valid_utf8, 'foo', invalid_utf8]
        self.assertEqual(self.m.alang, [valid_utf8, 'foo', invalid_utf8])
        self.assertEqual(self.m.raw.alang, [valid_utf8.encode('utf-8'), b'foo', invalid_utf8])
        with self.assertRaises(UnicodeDecodeError):
            self.m.strict.alang
        with self.assertRaises(UnicodeDecodeError):
            # See comment in test_property_decoding_invalid_utf8
            self.m.osd.alang

    def test_option_read(self):
        self.m.loop = 'inf'
        self.m.play(TESTVID)
        while self.m.core_idle:
            time.sleep(0.05)
        for name in sorted(self.m):
            with self.subTest(option_name=name), self.swallow_mpv_errors([
                mpv.ErrorCode.PROPERTY_UNAVAILABLE, mpv.ErrorCode.PROPERTY_NOT_FOUND, mpv.ErrorCode.PROPERTY_ERROR]):
                self.m[name]

    def test_multivalued_option(self):
        self.m['external-files'] = ['test.webm', b'test.webm']
        self.assertEqual(self.m['external-files'], ['test.webm', 'test.webm'])


class ObservePropertyTest(MpvTestCase):
    def test_observe_property(self):
        handler = mock.Mock()

        m = self.m
        m.observe_property('vid', handler)

        time.sleep(0.1)
        m.play(TESTVID)

        time.sleep(0.5) #couple frames
        m.unobserve_property('vid', handler)

        time.sleep(0.1) #couple frames
        m.terminate() # needed for synchronization of event thread
        handler.assert_has_calls([mock.call('vid', 'auto')])

    def test_property_observer_decorator(self):
        handler = mock.Mock()

        m = self.m
        m.play(TESTVID)

        m.slang = 'ru'
        m.mute = True

        @m.property_observer('mute')
        @m.property_observer('slang')
        def foo(*args, **kwargs):
            handler(*args, **kwargs)

        m.mute = False
        m.slang = 'jp'
        self.assertEqual(m.mute, False)
        self.assertEqual(m.slang, ['jp'])

        # Wait for tick. AFAICT property events are only generated at regular
        # intervals, and if we change a property too fast we don't get any
        # events. This is a limitation of the upstream API.
        time.sleep(0.1)
        # Another API limitation is that the order of property change events on
        # different properties does not necessarily exactly match the order in
        # which these properties were previously accessed. Thus, any_order.
        handler.assert_has_calls([
            mock.call('mute', False),
            mock.call('slang', ['jp'])],
            any_order=True)
        handler.reset_mock()

        m.mute = True
        m.slang = 'ru'
        self.assertEqual(m.mute, True)
        self.assertEqual(m.slang, ['ru'])

        time.sleep(0.1)
        foo.unobserve_mpv_properties()

        m.mute = False
        m.slang = 'jp'
        m.mute = True
        m.slang = 'ru'
        m.terminate() # needed for synchronization of event thread
        handler.assert_has_calls([
            mock.call('mute', True),
            mock.call('slang', ['ru'])],
            any_order=True)

class KeyBindingTest(MpvTestCase):
    def test_register_direct_cmd(self):
        self.m.register_key_binding('a', 'playlist-clear')
        self.assertEqual(self.m._key_binding_handlers, {})
        self.m.register_key_binding('Ctrl+Shift+a', 'playlist-clear')
        self.m.unregister_key_binding('a')
        self.m.unregister_key_binding('Ctrl+Shift+a')

    def test_register_direct_fun(self):
        b = mpv.MPV._binding_name

        def reg_test_fun(state, name, char):
            pass

        self.m.register_key_binding('a', reg_test_fun)
        self.assertIn(b('a'), self.m._key_binding_handlers)
        self.assertEqual(self.m._key_binding_handlers[b('a')], reg_test_fun)

        self.m.unregister_key_binding('a')
        self.assertNotIn(b('a'), self.m._key_binding_handlers)

    def test_register_direct_bound_method(self):
        b = mpv.MPV._binding_name

        class RegTestCls:
            def method(self, state, name, char):
                pass
        instance = RegTestCls()

        self.m.register_key_binding('a', instance.method)
        self.assertIn(b('a'), self.m._key_binding_handlers)
        self.assertEqual(self.m._key_binding_handlers[b('a')], instance.method)

        self.m.unregister_key_binding('a')
        self.assertNotIn(b('a'), self.m._key_binding_handlers)

    def test_register_decorator_fun(self):
        b = mpv.MPV._binding_name

        @self.m.key_binding('a')
        def reg_test_fun(state, name, char):
            pass
        self.assertEqual(reg_test_fun.mpv_key_bindings, ['a'])
        self.assertIn(b('a'), self.m._key_binding_handlers)
        self.assertEqual(self.m._key_binding_handlers[b('a')], reg_test_fun)

        reg_test_fun.unregister_mpv_key_bindings()
        self.assertNotIn(b('a'), self.m._key_binding_handlers)

    def test_register_decorator_fun_chaining(self):
        b = mpv.MPV._binding_name

        @self.m.key_binding('a')
        @self.m.key_binding('b')
        def reg_test_fun(state, name, char):
            pass

        @self.m.key_binding('c')
        def reg_test_fun_2_stay_intact(state, name, char):
            pass

        self.assertEqual(reg_test_fun.mpv_key_bindings, ['b', 'a'])
        self.assertIn(b('a'), self.m._key_binding_handlers)
        self.assertIn(b('b'), self.m._key_binding_handlers)
        self.assertIn(b('c'), self.m._key_binding_handlers)
        self.assertEqual(self.m._key_binding_handlers[b('a')], reg_test_fun)
        self.assertEqual(self.m._key_binding_handlers[b('b')], reg_test_fun)

        reg_test_fun.unregister_mpv_key_bindings()
        self.assertNotIn(b('a'), self.m._key_binding_handlers)
        self.assertNotIn(b('b'), self.m._key_binding_handlers)
        self.assertIn(b('c'), self.m._key_binding_handlers)

    def test_wait_for_event_error_forwarding(self):
        self.m.play(TESTVID)

        def check(evt):
            raise ValueError('fnord')

        with self.assertRaises(ValueError):
            self.m.wait_for_event('end_file', cond=check)

    def test_wait_for_property_error_forwarding(self):
        def run():
            nonlocal self
            self.m.wait_until_playing(timeout=2)
            self.m.mute = True
        t = threading.Thread(target=run, daemon=True)
        t.start()

        def cond(mute):
            if mute:
                raise ValueError('fnord')

        with self.assertRaises(ValueError):
            self.m.play(TESTVID)
            self.m.wait_for_property('mute', cond=cond)

    def test_register_simple_decorator_fun_chaining(self):
        self.m.loop = 'inf'
        self.m.play(TESTVID)
        self.m.wait_until_playing(timeout=2)

        handler1, handler2 = mock.Mock(), mock.Mock()

        @self.m.on_key_press('a')
        @self.m.on_key_press('b')
        def reg_test_fun(*args, **kwargs):
            handler1(*args, **kwargs)

        @self.m.on_key_press('c')
        def reg_test_fun_2_stay_intact(*args, **kwargs):
            handler2(*args, **kwargs)

        self.assertEqual(reg_test_fun.mpv_key_bindings, ['b', 'a'])

        def keypress_and_sync(key):
            with self.m.prepare_and_wait_for_event('client_message', timeout=2):
                self.m.keypress(key)

        keypress_and_sync('a')
        handler1.assert_has_calls([ mock.call() ])
        handler2.assert_has_calls([])
        handler1.reset_mock()

        self.m.keypress('x')
        self.m.keypress('X')
        keypress_and_sync('b')
        handler1.assert_has_calls([ mock.call() ])
        handler2.assert_has_calls([])
        handler1.reset_mock()

        keypress_and_sync('c')
        self.m.keypress('B')
        handler1.assert_has_calls([])
        handler2.assert_has_calls([ mock.call() ])
        handler2.reset_mock()

        reg_test_fun.unregister_mpv_key_bindings()
        self.m.keypress('a')
        keypress_and_sync('c')
        self.m.keypress('x')
        self.m.keypress('A')
        handler1.assert_has_calls([])
        handler2.assert_has_calls([ mock.call() ])

class TestStreams(unittest.TestCase):
    def test_python_stream(self):
        handler = mock.Mock()

        disp = Xvfb()
        disp.start()
        m = mpv.MPV(vo=testvo)
        def cb(evt):
            handler(evt.as_dict(decoder=mpv.lazy_decoder))
        m.register_event_callback(cb)

        @m.python_stream('foo')
        def foo_gen():
            with open(TESTVID, 'rb') as f:
                yield f.read()

        @m.python_stream('bar')
        def bar_gen():
            yield b''

        m.play('python://foo')
        m.wait_for_playback()
        handler.assert_any_call({'event': 'end-file', 'reason': 'eof', 'playlist_entry_id': 1})
        handler.reset_mock()

        m.play('python://bar')
        m.wait_for_playback()
        handler.assert_any_call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 2, 'file_error': 'unrecognized file format'})
        handler.reset_mock()

        m.play('python://baz')
        m.wait_for_playback()
        handler.assert_any_call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 3, 'file_error': 'loading failed'})
        handler.reset_mock()

        m.play('foo://foo')
        m.wait_for_playback()
        handler.assert_any_call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 4, 'file_error': 'loading failed'})
        handler.reset_mock()

        foo_gen.unregister()

        m.play('python://foo')
        m.wait_for_playback()
        handler.assert_any_call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 5, 'file_error': 'loading failed'})
        handler.reset_mock()

        m.play('python://bar')
        m.wait_for_playback()
        handler.assert_any_call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 6, 'file_error': 'unrecognized file format'})
        handler.reset_mock()

        m.terminate()
        disp.stop()

    def test_custom_stream(self):
        handler = mock.Mock()
        fail_mock = mock.Mock(side_effect=ValueError)
        stream_mock = mock.Mock()
        stream_mock.seek = mock.Mock(return_value=0)
        stream_mock.read = mock.Mock(return_value=b'')

        disp = Xvfb()
        disp.start()
        m = mpv.MPV(vo=testvo, video=False)
        def cb(evt):
            handler(evt.as_dict(decoder=mpv.lazy_decoder))
        m.register_event_callback(cb)

        m.register_stream_protocol('pythonfail', fail_mock)

        @m.register_stream_protocol('pythonsuccess')
        def open_fn(uri):
            self.assertEqual(uri, 'pythonsuccess://foo')
            return stream_mock

        m.play('pythondoesnotexist://foo')
        m.wait_for_playback()
        handler.assert_any_call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 1, 'file_error': 'loading failed'})
        handler.reset_mock()

        m.play('pythonfail://foo')
        m.wait_for_playback()
        handler.assert_any_call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 2, 'file_error': 'loading failed'})
        handler.reset_mock()

        m.play('pythonsuccess://foo')
        m.wait_for_playback()
        stream_mock.seek.assert_any_call(0)
        stream_mock.read.assert_called()
        handler.assert_any_call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 3, 'file_error': 'unrecognized file format'})

        m.terminate()
        disp.stop()

class TestLifecycle(unittest.TestCase):
    def test_create_destroy(self):
        thread_names = lambda: [ t.name for t in threading.enumerate() ]
        self.assertNotIn('MPVEventHandlerThread', thread_names())
        m = mpv.MPV()
        self.assertIn('MPVEventHandlerThread', thread_names())
        m.terminate()
        self.assertNotIn('MPVEventHandlerThread', thread_names())

    def test_flags(self):
        with self.assertRaises(AttributeError):
            mpv.MPV('this-option-does-not-exist')
        m = mpv.MPV('cursor-autohide-fs-only', 'fs', video=False)
        self.assertTrue(m.fullscreen)
        self.assertEqual(m.cursor_autohide, 1000)
        m.terminate()

    def test_options(self):
        with self.assertRaises(AttributeError):
            mpv.MPV(this_option_does_not_exists=23)
        m = mpv.MPV(osd_level=0, loop='inf', deinterlace=False)
        self.assertEqual(m.osd_level, 0)
        # For compatibility with mpv master (v0.32.0-585-gfba1c681b8) accept both
        self.assertIn(m.loop, ['inf', True])
        self.assertEqual(m.deinterlace, False)
        m.terminate()

    def test_event_callback(self):
        handler = mock.Mock()
        m = mpv.MPV(video=False)
        def cb(evt):
            handler(evt.as_dict(decoder=mpv.lazy_decoder))
        m.register_event_callback(cb)
        m.play(TESTVID)
        m.wait_for_playback()

        m.unregister_event_callback(cb)
        handler.assert_has_calls([
                mock.call({'event': 'start-file', 'playlist_entry_id': 1}),
                mock.call({'event': 'end-file', 'reason': 'error', 'playlist_entry_id': 1, 'file_error': 'no audio or video data played'})
            ], any_order=True)
        handler.reset_mock()

        m.terminate()
        handler.assert_not_called()

    def test_wait_for_property_negative(self):
        self.disp = Xvfb()
        self.disp.start()
        m = mpv.MPV(vo=testvo)
        m.play(TESTVID)
        result = Future()
        def run():
            nonlocal self
            result.set_running_or_notify_cancel()
            try:
                m.wait_for_property('mute')
                result.set_result(False)
            except mpv.ShutdownError:
                result.set_result(True)
        t = threading.Thread(target=run, daemon=True)
        t.start()
        time.sleep(1)
        m.terminate()
        t.join()
        self.disp.stop()
        assert result.result()

    def test_wait_for_property_positive(self):
        self.disp = Xvfb()
        self.disp.start()
        handler = mock.Mock()
        m = mpv.MPV(vo=testvo)
        m.play(TESTVID)
        def run():
            nonlocal self
            m.wait_for_property('mute')
            handler()
        t = threading.Thread(target=run, daemon=True)
        t.start()
        m.wait_until_playing(timeout=2)
        m.mute = True
        t.join()
        m.terminate()
        handler.assert_called()
        self.disp.stop()

    def test_wait_for_event(self):
        self.disp = Xvfb()
        self.disp.start()
        handler = mock.Mock()
        m = mpv.MPV(vo=testvo)
        m.play(TESTVID)
        result = Future()
        def run():
            nonlocal self
            result.set_running_or_notify_cancel()
            try:
                m.wait_for_event('seek')
                result.set_result(False)
            except mpv.ShutdownError:
                result.set_result(True)
        t = threading.Thread(target=run, daemon=True)
        t.start()
        time.sleep(1)
        m.terminate()
        t.join()
        self.disp.stop()
        assert result.result()

    def test_wait_for_property_shutdown(self):
        self.disp = Xvfb()
        self.disp.start()
        handler = mock.Mock()
        m = mpv.MPV(vo=testvo)
        m.play(TESTVID)
        with self.assertRaises(mpv.ShutdownError):
            # level_sensitive=false needed to prevent get_property on dead
            # handle
            with m.prepare_and_wait_for_property('mute', level_sensitive=False):
                m.terminate()
        self.disp.stop()

    def test_wait_for_prooperty_event_overflow(self):
        self.disp = Xvfb()
        self.disp.start()
        handler = mock.Mock()
        m = mpv.MPV(vo=testvo)
        m.play(TESTVID)
        with self.assertRaises(mpv.EventOverflowError):
            # level_sensitive=false needed to prevent get_property on dead
            # handle
            with m.prepare_and_wait_for_property('mute', cond=lambda val: time.sleep(0.001)):
                for i in range(10000):
                    try:
                        # We really have to try hard to fill up the queue here. Simple async commands will not work,
                        # since then command_async will throw a memory error first. Property changes also do not work,
                        # since they are only processsed when the event loop is idle. This here works reliably.
                        m.command_async('script-message', 'foo', 'bar')
                    except:
                        pass
        self.disp.stop()



    def test_wait_for_event_shutdown(self):
        self.disp = Xvfb()
        self.disp.start()
        m = mpv.MPV(vo=testvo)
        m.play(TESTVID)
        with self.assertRaises(mpv.ShutdownError):
            with m.prepare_and_wait_for_event('seek'):
                m.terminate()
        self.disp.stop()

    def test_wait_for_shutdown(self):
        self.disp = Xvfb()
        self.disp.start()
        handler = mock.Mock()
        m = mpv.MPV(vo=testvo)
        m.play(TESTVID)
        with self.assertRaises(mpv.ShutdownError):
            with m.prepare_and_wait_for_event(None) as result:
                m.terminate()
            result.result()
        self.disp.stop()

    def test_log_handler(self):
        handler = mock.Mock()
        self.disp = Xvfb()
        self.disp.start()
        m = mpv.MPV(vo=testvo, log_handler=handler)
        m.play(TESTVID)
        # Wait for playback to start
        m.wait_until_playing(timeout=2)
        m.command("print-text", 'This is a python-mpv test')
        m.wait_for_playback()
        m.terminate()
        for call in handler.mock_calls:
            _1, (a, b, c), _2 = call
            if a == 'info' and b == 'cplayer' and 'This is a python-mpv test' in c:
                break
        else:
            self.fail('"Test log entry not found in log handler calls: '+','.join(repr(call) for call in handler.mock_calls))
        self.disp.stop()


class CommandTests(MpvTestCase):

    def test_loadfile_with_subtitles(self):
        handler = mock.Mock()
        self.m.property_observer('sub-text')(handler)

        self.m.loadfile(TESTVID, sub_file=TESTSRT)

        self.m.wait_for_playback()
        handler.assert_any_call('sub-text', 'This is\na subtitle test.')
        handler.assert_any_call('sub-text', 'This is the second subtitle line.')

    def test_sub_add(self):
        handler = mock.Mock()
        self.m.property_observer('sub-text')(handler)
        time.sleep(0.5)

        self.m.loadfile(TESTVID)
        self.m.wait_until_playing(timeout=2)
        self.m.sub_add(TESTSRT)

        self.m.wait_for_playback()
        handler.assert_any_call('sub-text', 'This is\na subtitle test.')
        handler.assert_any_call('sub-text', 'This is the second subtitle line.')

    def test_async_command(self):
        handler = mock.Mock()
        callback = mock.Mock()
        self.m.property_observer('sub-text')(handler)
        time.sleep(0.5)

        self.m.loadfile(TESTVID)
        self.m.wait_until_playing(timeout=2)
        self.m.command_async('sub_add', TESTSRT, callback=callback)
        reply = self.m.command_async('expand-text', 'test ${mute}')
        assert reply.result() == 'test no'

        self.m.wait_for_playback()
        handler.assert_any_call('sub-text', 'This is\na subtitle test.')
        handler.assert_any_call('sub-text', 'This is the second subtitle line.')
        callback.assert_any_call(None, None)



class RegressionTests(MpvTestCase):

    def test_unobserve_property_runtime_error(self):
        """
        Ensure a `RuntimeError` is not thrown within
        `unobserve_property`.
        """
        handler = mock.Mock()

        self.m.observe_property('loop', handler)

        try:
            self.m.unobserve_property('loop', handler)
        except RuntimeError:
            self.fail(
                """
                "RuntimeError" exception thrown within
                `unobserve_property`
                """,
            )

    def test_instance_method_property_observer(self):
        """
        Ensure that bound method objects can be used as property observers.
        See issue #26
        """
        handler = mock.Mock()
        m = self.m

        class T(object):
            def t(self, *args, **kw):
                handler(*args, **kw)
        t =  T()

        m.slang = 'ru'
        time.sleep(0.5)

        m.observe_property('slang', t.t)
        time.sleep(0.5)

        m.slang = 'jp'
        time.sleep(0.5)

        m.slang = 'ru'
        time.sleep(0.5)

        m.unobserve_property('slang', t.t)
        time.sleep(0.5)

        m.slang = 'jp'
        m.slang = 'ru'
        m.terminate() # needed for synchronization of event thread
        handler.assert_has_calls([mock.call('slang', ['jp']), mock.call('slang', ['ru'])])


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