Add initial test code for MRP

This is just some dummy code to get going with MRP tests. Currently no
tests work but all common tests run.

tests/airplay/fake_airplay_device.py

@@ -37,7 +37,7 @@
 
 class FakeAirPlayDevice(web.Application):
 
-    def __init__(self, loop, testcase):
+    def __init__(self, testcase):
         super().__init__()
         self.responses = {}
         self.responses['airplay_playback'] = []

tests/airplay/test_airplay.py

@@ -25,7 +25,7 @@ def setUp(self):
         self.no_of_sleeps = 0
 
     async def get_application(self, loop=None):
-        self.fake_device = FakeAirPlayDevice(self.loop, self)
+        self.fake_device = FakeAirPlayDevice(self)
         self.usecase = AirPlayUseCases(self.fake_device)
         return self.fake_device
 

tests/airplay/test_airplay_auth.py

@@ -25,7 +25,7 @@ async def tearDownAsync(self):
         await self.session.close()
 
     async def get_application(self, loop=None):
-        self.fake_atv = FakeAirPlayDevice(self.loop, self)
+        self.fake_atv = FakeAirPlayDevice(self)
         return self.fake_atv
 
     @unittest_run_loop

tests/dmap/fake_dmap_atv.py

@@ -53,11 +53,11 @@ def __init__(self, revision=0, shuffle=False, **kwargs):
 
 
 class FakeAppleTV(FakeAirPlayDevice):
-    """Implementation of fake Apple TV."""
+    """Implementation of a fake DMAP Apple TV."""
 
-    def __init__(self, loop, hsgid, pairing_guid, session_id, testcase):
+    def __init__(self, hsgid, pairing_guid, session_id, testcase):
         """Initialize a new FakeAppleTV."""
-        super().__init__(loop, testcase)
+        super().__init__(testcase)
 
         self.responses['login'] = [LoginResponse(session_id, 200)]
         self.responses['artwork'] = []
@@ -111,19 +111,21 @@ async def handle_remote_button(self, request):
         self.buttons_press_count += 1
         return web.Response(status=200)
 
-    @staticmethod
-    def _convert_button(data):
+    def _convert_button(self, data):
         value = parser.first(data, 'cmbe')
-        if value == 'touchUp&time=6&point=20,250':
-            return 'up'
-        elif value == 'touchUp&time=6&point=20,275':
-            return 'down'
-        elif value == 'touchUp&time=7&point=50,100':
-            return 'left'
-        elif value == 'touchUp&time=7&point=75,100':
-            return 'right'
-        else:
-            return value
+
+        # Consider navigation buttons if six commands have been received
+        if self.buttons_press_count == 6:
+            if value == 'touchUp&time=6&point=20,250':
+                return 'up'
+            elif value == 'touchUp&time=6&point=20,275':
+                return 'down'
+            elif value == 'touchUp&time=7&point=50,100':
+                return 'left'
+            elif value == 'touchUp&time=7&point=75,100':
+                return 'right'
+
+        return value
 
     async def handle_artwork(self, request):
         """Handle artwork requests."""
@@ -341,15 +343,6 @@ def media_is_loading(self):
         self.device.responses['playing'].insert(0, PlayingResponse(
             playstatus=1))
 
-    def set_property(self, prop, value):
-        """Change value of a property."""
-        # Use "fictional" properties to not tie them to DAP (if some other
-        # protocol is to be supported in the future)
-        if prop == 'shuffle':
-            self.device.properties['dacp.shufflestate'] = value
-        elif prop == 'repeat':
-            self.device.properties['dacp.repeatstate'] = value
-
     def pairing_response(self, remote_name, expected_pairing_code):
         """Reponse when a pairing request is made."""
         self.device.responses['pairing'].insert(0, PairingResponse(

tests/dmap/test_dmap_functional.py

@@ -46,15 +46,15 @@ def tearDown(self):
 
     async def get_application(self, loop=None):
         self.fake_atv = FakeAppleTV(
-            self.loop, HSGID, PAIRING_GUID, SESSION_ID, self)
+            HSGID, PAIRING_GUID, SESSION_ID, self)
         self.usecase = AppleTVUseCases(self.fake_atv)
         return self.fake_atv
 
     def get_connected_device(self, identifier):
-        details = AppleTV('127.0.0.1', 'Apple TV')
-        details.add_service(DmapService(identifier, port=self.server.port))
-        details.add_service(AirPlayService(self.server.port))
-        return connect_to_apple_tv(details, self.loop)
+        conf = AppleTV('127.0.0.1', 'Apple TV')
+        conf.add_service(DmapService(identifier, port=self.server.port))
+        conf.add_service(AirPlayService(self.server.port))
+        return connect_to_apple_tv(conf, self.loop)
 
     @unittest_run_loop
     async def test_not_supportedt(self):

tests/mrp/fake_mrp_atv.py

@@ -0,0 +1,109 @@
+"""Fake MRP Apple TV for tests."""
+
+import asyncio
+import logging
+
+from pyatv.mrp import (messages, protobuf, variant)
+from tests.airplay.fake_airplay_device import (
+    FakeAirPlayDevice, AirPlayUseCases)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class FakeAppleTV(FakeAirPlayDevice, asyncio.Protocol):
+    """Implementation of a fake MRP Apple TV."""
+
+    def __init__(self, testcase):
+        super().__init__(testcase)
+        self.buttons_press_count = 0
+        self.last_button_pressed = None
+
+        self.server = None
+        self.buffer = b''
+        self.transport = None
+        self.mapping = {
+            protobuf.DEVICE_INFO_MESSAGE: self.handle_device_info,
+            protobuf.CRYPTO_PAIRING_MESSAGE: self.handle_crypto_pairing,
+            }
+
+    @asyncio.coroutine
+    def start(self, loop):
+        coro = loop.create_server(lambda: self, '127.0.0.1')
+        self.server = yield from loop.create_task(coro)
+        _LOGGER.info('Started MRP server at port %d', self.port)
+
+    @property
+    def port(self):
+        return self.server.sockets[0].getsockname()[1]
+
+    def connection_made(self, transport):
+        self.transport = transport
+
+    def _send(self, message):
+        data = message.SerializeToString()
+        length = variant.write_variant(len(data))
+        self.transport.write(length + data)
+
+    def data_received(self, data):
+        self.buffer += data
+
+        length, raw = variant.read_variant(self.buffer)
+        if len(raw) < length:
+            return
+
+        data = raw[:length]
+        self.buffer = raw[length:]
+        parsed = protobuf.ProtocolMessage()
+        parsed.ParseFromString(data)
+        _LOGGER.info('Incoming message: %s', parsed)
+
+        try:
+            def unhandled_message(message):
+                _LOGGER.warning('No message handler for %s', message)
+
+            self.mapping.get(parsed.type, unhandled_message)(parsed)
+        except Exception:
+            _LOGGER.exception('Error while dispatching message')
+
+    def handle_device_info(self, message):
+        _LOGGER.debug('Received device info message')
+
+        resp = messages.device_information('Fake MRP ATV', '1234')
+        resp.identifier = message.identifier
+        self._send(resp)
+
+    def handle_crypto_pairing(self, message):
+        _LOGGER.debug('Received crypto pairing message')
+
+
+class AppleTVUseCases(AirPlayUseCases):
+    """Wrapper for altering behavior of a FakeMrpAppleTV instance."""
+
+    def __init__(self, fake_apple_tv):
+        """Initialize a new AppleTVUseCases."""
+        self.device = fake_apple_tv
+
+    def change_artwork(self, artwork):
+        """Call this method to change artwork response."""
+        pass
+
+    def nothing_playing(self):
+        """Call this method to put device in idle state."""
+        pass
+
+    def example_video(self, **kwargs):
+        """Play some example video."""
+        pass
+
+    def video_playing(self, paused, title, total_time, position, **kwargs):
+        """Call this method to change what is currently plaing to video."""
+        pass
+
+    def music_playing(self, paused, artist, album, title, genre,
+                      total_time, position):
+        """Call this method to change what is currently plaing to music."""
+        pass
+
+    def media_is_loading(self):
+        """Call this method to put device in a loading state."""
+        pass

tests/mrp/test_mrp_functional.py

@@ -0,0 +1,39 @@
+"""Functional tests using the API with a fake Apple TV."""
+
+import unittest
+
+from aiohttp.test_utils import unittest_run_loop
+
+from pyatv import connect_to_apple_tv
+from pyatv.conf import (AirPlayService, MrpService, AppleTV)
+
+from tests import common_functional_tests
+from tests.mrp.fake_mrp_atv import (
+    FakeAppleTV, AppleTVUseCases)
+
+
+@unittest.skip('not ready yet')
+class MRPFunctionalTest(common_functional_tests.CommonFunctionalTests):
+
+    def setUp(self):
+        super().setUp()
+        self.atv = self.get_connected_device(self.fake_atv.port)
+
+    def tearDown(self):
+        pass
+
+    async def get_application(self, loop=None):
+        self.fake_atv = FakeAppleTV(self)
+        self.usecase = AppleTVUseCases(self.fake_atv)
+        await self.fake_atv.start(self.loop)
+        return self.fake_atv
+
+    def get_connected_device(self, port):
+        conf = AppleTV('127.0.0.1', 'Test device')
+        conf.add_service(MrpService(port))
+        conf.add_service(AirPlayService(self.server.port))
+        return connect_to_apple_tv(conf, loop=self.loop)
+
+    @unittest_run_loop
+    def test_dummy_test(self):
+        self.assertFalse(True)