Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_unknown_playstate_throws(self):
with self.assertRaises(exceptions.UnknownPlayState):
convert.playstate(99999)
def test_unknown_playstate_throws(self):
with self.assertRaises(exceptions.UnknownPlayState):
convert.playstate(99999)
async def test_user_pair_invalid_pin(flow, mrp_device, pairing_mock):
"""Test pairing with invalid pin."""
pairing_mock.begin.return_value = mock_coro()
pairing_mock.finish.side_effect = exceptions.PairingError()
(await flow().step_user(identifier="MRP Device")).gives_form_confirm(
description_placeholders={"name": "MRP Device"}
)
(await flow().step_confirm()).gives_form_pair_with_pin()
(await flow().step_pair_with_pin(pin=1111)).gives_form_pair_with_pin(
errors={"base": "auth"}
)
def test_main_service_no_service(self):
with self.assertRaises(exceptions.NoServiceError):
self.config.main_service()
async def test_play_video_no_permission(self):
self.usecase.airplay_playback_playing_no_permission()
with self.assertRaises(exceptions.NoCredentialsError):
await self.player.play_url(STREAM, position=START_POSITION)
async def _scan(self):
identifier = self.config_entry.unique_id
address = self.config_entry.data[CONF_ADDRESS]
protocol = Protocol(self.config_entry.data[CONF_PROTOCOL])
self._update_state(message="Discovering device...")
try:
atvs = await scan(
self.hass.loop,
identifier=identifier,
protocol=protocol,
hosts=[address],
)
if atvs:
return atvs[0]
except exceptions.NonLocalSubnetError:
_LOGGER.debug(
"Address %s is on non-local subnet, relying on regular scan", address
)
_LOGGER.debug(
"Failed to find device %s with address %s, trying to scan",
identifier,
address,
)
atvs = await scan(self.hass.loop, identifier=identifier, protocol=protocol)
if atvs:
return atvs[0]
self._update_state("Device not found, trying again later...")
_LOGGER.debug("Failed to find device %s, trying later", identifier)
def main_service(self, protocol=None):
"""Return suggested service used to establish connection."""
protocols = [protocol] if protocol is not None else \
[Protocol.MRP, Protocol.DMAP]
for prot in protocols:
service = self.get_service(prot)
if service is not None:
return service
raise exceptions.NoServiceError(
'no service to connect to')
def media_kind(kind):
"""Convert iTunes media kind to API representation."""
if kind in [1]:
return MediaType.Unknown
if kind in [3, 7, 11, 12, 13, 18, 32]:
return MediaType.Video
if kind in [2, 4, 10, 14, 17, 21, 36]:
return MediaType.Music
if kind in [8, 64]:
return MediaType.TV
raise exceptions.UnknownMediaKind('Unknown media kind: ' + str(kind))
headers=HEADERS,
data=plistlib.dumps(body, fmt=plistlib.FMT_BINARY),
timeout=TIMEOUT)
# Sometimes AirPlay fails with "Internal Server Error", we
# apply a "lets try again"-approach to that
if status == 500:
retry += 1
_LOGGER.debug('Failed to stream %s, retry %d of %d',
url, retry, PLAY_RETRIES)
await asyncio.sleep(1.0, loop=self.loop)
continue
# TODO: Should be more fine-grained
if 400 <= status < 600:
raise exceptions.AuthenticationError(
'Status code: ' + str(status))
await self._wait_for_media_to_end()
return
raise exceptions.PlaybackError('Max retries exceeded')
async def authenticate_with_device(atv):
"""Perform device authentication and print credentials."""
credentials = await atv.airplay.generate_credentials()
await atv.airplay.load_credentials(credentials)
try:
await atv.airplay.start_authentication()
pin = input('PIN Code: ')
await atv.airplay.finish_authentication(pin)
print('Credentials: {0}'.format(credentials))
except exceptions.DeviceAuthenticationError:
print('Failed to authenticate', file=sys.stderr)