Add test for play_url and AirPlay

pyatv/airplay.py

@@ -82,4 +82,4 @@ def _wait_for_media_to_end(self, port):
             finally:
                 yield from info.release()
 
-            yield from asyncio.sleep(1, self.loop)
+            yield from asyncio.sleep(1, loop=self.loop)

pyatv/internal/apple_tv.py

@@ -146,11 +146,9 @@ def play_url(self, url, start_position, **kwargs):
         """
         # AirPlay is separate from DAAP, so to make it easier to test the port
         # can be overriden to something else. NOT part of the public API!
-        if 'port' in kwargs:
-            port = kwargs['port']
-        else:
-            import pyatv.airplay
-            port = pyatv.airplay.AIRPLAY_PORT
+        import pyatv.airplay
+        port_override = 'port' in kwargs
+        port = kwargs['port'] if port_override else pyatv.airplay.AIRPLAY_PORT
         return self.airplay.play_url(url, int(start_position), port)
 
 

tests/fake_apple_tv.py

@@ -6,7 +6,9 @@
 information is correct and headers are present.
 """
 
+import re
 import asyncio
+import plistlib
 from collections import namedtuple
 
 from aiohttp import web
@@ -26,6 +28,7 @@
 
 LoginResponse = namedtuple('LoginResponse', 'session, status')
 ArtworkResponse = namedtuple('ArtworkResponse', 'content, status')
+AirPlayPlaybackResponse = namedtuple('AirPlayPlaybackResponse', 'content')
 
 
 class PlayingResponse:
@@ -59,13 +62,15 @@ def __init__(self, loop, hsgid, pairing_guid, session_id, testcase):
         self.responses['login'] = [LoginResponse(session_id, 200)]
         self.responses['artwork'] = []
         self.responses['playing'] = []
+        self.responses['airplay_playback'] = []
         self.hsgid = hsgid
         self.pairing_guid = pairing_guid
         self.session = None
         self.last_button_pressed = None
         self.properties = {}  # setproperty
         self.tc = testcase
 
+        # Regular DAAP stuff
         self.router.add_get('/login', self.handle_login)
         self.router.add_get(
             '/ctrl-int/1/playstatusupdate', self.handle_playstatus)
@@ -79,6 +84,11 @@ def __init__(self, loop, hsgid, pairing_guid, session_id, testcase):
             self.router.add_post('/ctrl-int/1/' + button,
                                  self.handle_playback_button)
 
+        # AirPlay stuff
+        self.router.add_post('/play', self.handle_airplay_play)
+        self.router.add_get('/playback-info',
+                            self.handle_airplay_playback_info)
+
     # This method will retrieve the next response for a certain type.
     # If there are more than one response, it "pop" the last one and
     # return it. When only one response remains, that response will be
@@ -209,6 +219,30 @@ def _verify_auth_parameters(self,
             self.tc.assertEqual(int(params['session-id']), self.session,
                                 msg='session id does not match')
 
+    @asyncio.coroutine
+    def handle_airplay_play(self, request):
+        """Handler for AirPlay play requests."""
+        headers = request.headers
+
+        # Verify headers first
+        self.tc.assertEqual(headers['User-Agent'], 'MediaControl/1.0')
+        self.tc.assertEqual(headers['Content-Type'], 'text/parameters')
+
+        body = yield from request.text()
+
+        self.last_airplay_url = re.search(
+            r'Content-Location: (.*)', body).group(1)
+        self.last_airplay_start = float(re.search(
+            r'Start-Position: (.*)', body).group(1))
+
+        return web.Response(status=200)
+
+    @asyncio.coroutine
+    def handle_airplay_playback_info(self, request):
+        """Handler for AirPlay playback-info requests."""
+        response = self._get_response('airplay_playback')
+        return web.Response(body=response.content, status=200)
+
 
 class AppleTVUseCases:
     """Wrapper for altering behavior of a FakeAppleTV instance.
@@ -270,3 +304,16 @@ def media_is_loading(self):
         """Calling this method puts device in a loading state."""
         self.device.responses['playing'].insert(0, PlayingResponse(
             playstatus=1))
+
+    def airplay_playback_idle(self):
+        """Make playback-info return idle info."""
+        plist = dict(readyToPlay=False, uuid=123)
+        self.device.responses['airplay_playback'].insert(
+            0, AirPlayPlaybackResponse(plistlib.dumps(plist)))
+
+    def airplay_playback_playing(self):
+        """Make playback-info return that something is playing."""
+        # This is _not_ complete, currently not needed
+        plist = dict(duration=0.8)
+        self.device.responses['airplay_playback'].insert(
+            0, AirPlayPlaybackResponse(plistlib.dumps(plist)))

tests/test_airplay.py

@@ -0,0 +1,59 @@
+"""Functional tests using the API with a fake Apple TV."""
+
+import asyncio
+
+from tests.log_output_handler import LogOutputHandler
+from aiohttp.test_utils import (AioHTTPTestCase, unittest_run_loop)
+
+from pyatv import airplay
+from tests.fake_apple_tv import (FakeAppleTV, AppleTVUseCases)
+
+
+STREAM = 'http://airplaystream'
+START_POSITION = 0.8
+
+
+class AirPlayTest(AioHTTPTestCase):
+
+    def setUp(self):
+        AioHTTPTestCase.setUp(self)
+        self.log_handler = LogOutputHandler(self)
+
+        # This is a hack that overrides asyncio.sleep to avoid making the test
+        # slow. It also counts number of calls, since this is quite important
+        # to the general function.
+        airplay.asyncio.sleep = self.fake_asyncio_sleep
+        self.no_of_sleeps = 0
+
+    def tearDown(self):
+        AioHTTPTestCase.tearDown(self)
+        self.log_handler.tearDown()
+
+    def get_app(self, loop):
+        self.fake_atv = FakeAppleTV(loop, 0, 0, 0, self)
+        self.usecase = AppleTVUseCases(self.fake_atv)
+
+        # Import TestServer here and not globally, otherwise py.test will
+        # complain when running:
+        #
+        #   test_functional.py cannot collect test class 'TestServer'
+        #   because it has a __init__ constructor
+        from aiohttp.test_utils import TestServer
+        return TestServer(self.fake_atv)
+
+    @asyncio.coroutine
+    def fake_asyncio_sleep(self, time, loop):
+        self.no_of_sleeps += 1
+
+    @unittest_run_loop
+    def test_play_video(self):
+        self.usecase.airplay_playback_idle()
+        self.usecase.airplay_playback_playing()
+        self.usecase.airplay_playback_idle()
+
+        aplay = airplay.AirPlay(self.loop, '127.0.0.1')
+        yield from aplay.play_url(STREAM, START_POSITION, self.app.port)
+
+        self.assertEqual(self.fake_atv.last_airplay_url, STREAM)
+        self.assertEqual(self.fake_atv.last_airplay_start, START_POSITION)
+        self.assertEqual(self.no_of_sleeps, 2)  # playback + idle = 3

tests/test_functional.py

@@ -19,6 +19,7 @@
 PIN_CODE = 1234
 
 EXPECTED_ARTWORK = b'1234'
+AIRPLAY_STREAM = 'http://stream'
 
 # This is valid for the PAIR in the pairing module and pin 1234
 # (extracted form a real device)
@@ -107,6 +108,17 @@ def test_pairing_with_device(self):
         yield from handler.stop()
         yield from self.atv.logout()
 
+    @unittest_run_loop
+    def test_play_url(self):
+        self.usecase.airplay_playback_idle()
+        self.usecase.airplay_playback_playing()
+        self.usecase.airplay_playback_idle()
+
+        yield from self.atv.remote_control.play_url(
+            AIRPLAY_STREAM, 0, port=self.app.port)
+
+        self.assertEqual(self.fake_atv.last_airplay_url, AIRPLAY_STREAM)
+
     @unittest_run_loop
     def test_login_failed(self):
         # Twice since the client will retry one time