diff options
| -rwxr-xr-x | mpv-test.py | 22 | ||||
| -rw-r--r-- | mpv.py | 67 |
2 files changed, 53 insertions, 36 deletions
diff --git a/mpv-test.py b/mpv-test.py index 96bf225..4b05a4d 100755 --- a/mpv-test.py +++ b/mpv-test.py @@ -42,7 +42,7 @@ class TestProperties(MpvTestCase): while self.m.core_idle: time.sleep(0.05) for name in sorted(self.m.property_list): - name = name.replace('-', '_') + name = name.replace('-', '_') with self.subTest(property_name=name), self.swallow_mpv_errors([ mpv.ErrorCode.PROPERTY_UNAVAILABLE, mpv.ErrorCode.PROPERTY_ERROR, @@ -132,7 +132,6 @@ class TestProperties(MpvTestCase): def test_property_decoding_invalid_utf8(self): invalid_utf8 = b'foo\xc3\x28bar' self.m.alang = invalid_utf8 - self.assertEqual(self.m.alang, [invalid_utf8]) self.assertEqual(self.m.raw.alang, [invalid_utf8]) with self.assertRaises(UnicodeDecodeError): self.m.strict.alang @@ -186,7 +185,7 @@ class ObservePropertyTest(MpvTestCase): time.sleep(0.1) m.play(TESTVID) - time.sleep(0.1) #couple frames + time.sleep(0.5) #couple frames m.unobserve_property('vid', handler) time.sleep(0.1) #couple frames @@ -228,7 +227,7 @@ class ObservePropertyTest(MpvTestCase): m.mute = True m.loop = 'inf' self.assertEqual(m.mute, True) - self.assertEqual(m.loop, 'inf') + self.assertEqual(m.loop, True) time.sleep(0.05) foo.unobserve_mpv_properties() @@ -240,7 +239,7 @@ class ObservePropertyTest(MpvTestCase): m.terminate() # needed for synchronization of event thread handler.assert_has_calls([ mock.call('mute', True), - mock.call('loop', 'inf')], + mock.call('loop', True)], any_order=True) class KeyBindingTest(MpvTestCase): @@ -373,7 +372,7 @@ class TestLifecycle(unittest.TestCase): mpv.MPV(this_option_does_not_exists=23) m = mpv.MPV(osd_level=0, loop='inf', deinterlace=False) self.assertEqual(m.osd_level, 0) - self.assertEqual(m.loop, 'inf') + self.assertEqual(m.loop, True) self.assertEqual(m.deinterlace, False) m.terminate() @@ -401,7 +400,12 @@ class TestLifecycle(unittest.TestCase): m.play(TESTVID) m.wait_for_playback() m.terminate() - handler.assert_any_call('info', 'cplayer', 'Playing: test.webm') + for call in handler.mock_calls: + _1, (a, b, c), _2 = call + if a == 'info' and b == 'cplayer' and c.startswith('Playing: '): + break + else: + self.fail('"Playing: foo..." call not found in log handler calls: '+','.join(repr(call) for call in handler.mock_calls)) class RegressionTests(MpvTestCase): @@ -449,7 +453,7 @@ class RegressionTests(MpvTestCase): # events. This is a limitation of the upstream API. time.sleep(0.01) m.loop = 'inf' - self.assertEqual(m.loop, 'inf') + self.assertEqual(m.loop, True) time.sleep(0.02) m.unobserve_property('loop', t.t) @@ -457,7 +461,7 @@ class RegressionTests(MpvTestCase): m.loop = False m.loop = 'inf' m.terminate() # needed for synchronization of event thread - handler.assert_has_calls([mock.call('loop', False), mock.call('loop', 'inf')]) + handler.assert_has_calls([mock.call('loop', False), mock.call('loop', True)]) if __name__ == '__main__': @@ -191,27 +191,37 @@ class MpvByteArray(Structure): return cast(self.data, POINTER(c_char))[:self.size] class MpvNode(Structure): - _fields_ = [('val', c_longlong), - ('format', MpvFormat)] - def node_value(self, decoder=identity_decoder): - return MpvNode.node_cast_value(byref(c_void_p(self.val)), self.format.value, decoder) + return MpvNode.node_cast_value(self.val, self.format.value, decoder) @staticmethod def node_cast_value(v, fmt=MpvFormat.NODE, decoder=identity_decoder): return { MpvFormat.NONE: lambda v: None, - MpvFormat.STRING: lambda v: decoder(cast(v, POINTER(c_char_p)).contents.value), - MpvFormat.OSD_STRING: lambda v: cast(v, POINTER(c_char_p)).contents.value.decode('utf-8'), - MpvFormat.FLAG: lambda v: bool(cast(v, POINTER(c_int)).contents.value), - MpvFormat.INT64: lambda v: cast(v, POINTER(c_longlong)).contents.value, - MpvFormat.DOUBLE: lambda v: cast(v, POINTER(c_double)).contents.value, - MpvFormat.NODE: lambda v: cast(v, POINTER(MpvNode)).contents.node_value(decoder), - MpvFormat.NODE_ARRAY: lambda v: cast(v, POINTER(POINTER(MpvNodeList))).contents.contents.array_value(decoder), - MpvFormat.NODE_MAP: lambda v: cast(v, POINTER(POINTER(MpvNodeList))).contents.contents.dict_value(decoder), - MpvFormat.BYTE_ARRAY: lambda v: cast(v, POINTER(POINTER(MpvByteArray))).contents.contents.bytes_value(), + MpvFormat.STRING: lambda v: decoder(v.string), + MpvFormat.OSD_STRING: lambda v: v.string.decode('utf-8'), + MpvFormat.FLAG: lambda v: bool(v.flag), + MpvFormat.INT64: lambda v: v.int64, + MpvFormat.DOUBLE: lambda v: v.double, + MpvFormat.NODE: lambda v: v.node.contents.node_value(decoder) if v.node else None, + MpvFormat.NODE_ARRAY: lambda v: v.list.contents.array_value(decoder) if v.list else None, + MpvFormat.NODE_MAP: lambda v: v.map.contents.dict_value(decoder) if v.map else None, + MpvFormat.BYTE_ARRAY: lambda v: v.byte_array.contents.bytes_value() if v.byte_array else None, }[fmt](v) +class MpvNodeUnion(Union): + _fields_ = [('string', c_char_p), + ('flag', c_int), + ('int64', c_int64), + ('double', c_double), + ('node', POINTER(MpvNode)), + ('list', POINTER(MpvNodeList)), + ('map', POINTER(MpvNodeList)), + ('byte_array', POINTER(MpvByteArray))] + +MpvNode._fields_ = [('val', MpvNodeUnion), + ('format', MpvFormat)] + MpvNodeList._fields_ = [('num', c_int), ('values', POINTER(MpvNode)), ('keys', POINTER(c_char_p))] @@ -241,7 +251,7 @@ class MpvEvent(Structure): class MpvEventProperty(Structure): _fields_ = [('name', c_char_p), ('format', MpvFormat), - ('data', c_void_p)] + ('data', MpvNodeUnion)] def as_dict(self, decoder=identity_decoder): value = MpvNode.node_cast_value(self.data, self.format.value, decoder) return {'name': self.name.decode('utf-8'), @@ -409,11 +419,11 @@ def _make_node_str_list(l): keys=None, values=( MpvNode * len(l))( *[ MpvNode( format=MpvFormat.STRING, - val=cast(pointer(p), POINTER(c_longlong)).contents) # NOTE: ctypes is kinda crappy here + val=MpvNodeUnion(string=p)) for p in char_ps ])) node = MpvNode( format=MpvFormat.NODE_ARRAY, - val=addressof(node_list)) + val=MpvNodeUnion(list=pointer(node_list))) return char_ps, node_list, node, cast(pointer(node), c_void_p) @@ -607,11 +617,10 @@ class MPV(object): def node_command(self, name, *args, decoder=strict_decoder): _1, _2, _3, pointer = _make_node_str_list([name, *args]) out = cast(create_string_buffer(sizeof(MpvNode)), POINTER(MpvNode)) - outptr = out #byref(out) ppointer = cast(pointer, POINTER(MpvNode)) - _mpv_command_node(self.handle, ppointer, outptr) - rv = MpvNode.node_cast_value(outptr, MpvFormat.NODE, decoder) - _mpv_free_node_contents(outptr) + _mpv_command_node(self.handle, ppointer, out) + rv = out.contents.node_value(decoder=decoder) + _mpv_free_node_contents(out) return rv def seek(self, amount, reference="relative", precision="default-precise"): @@ -1023,14 +1032,18 @@ class MPV(object): # Property accessors def _get_property(self, name, decoder=strict_decoder, fmt=MpvFormat.NODE): - out = cast(create_string_buffer(sizeof(c_void_p)), c_void_p) - outptr = byref(out) + out = create_string_buffer(sizeof(MpvNode)) try: - cval = _mpv_get_property(self.handle, name.encode('utf-8'), fmt, outptr) - rv = MpvNode.node_cast_value(outptr, fmt, decoder) - if fmt is MpvFormat.NODE: - _mpv_free_node_contents(outptr) - return rv + cval = _mpv_get_property(self.handle, name.encode('utf-8'), fmt, out) + + if fmt is MpvFormat.OSD_STRING: + return cast(out, POINTER(c_char_p)).contents.value.decode('utf-8') + elif fmt is MpvFormat.NODE: + rv = cast(out, POINTER(MpvNode)).contents.node_value(decoder=decoder) + _mpv_free_node_contents(out) + return rv + else: + raise TypeError('_get_property only supports NODE and OSD_STRING formats.') except PropertyUnavailableError as ex: return None |