Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_connect_invalid_protocol(self):
conf = AppleTV('1.2.3.4', 'Apple TV')
conf.add_service(AirPlayService('airplay_id'))
with self.assertRaises(exceptions.UnsupportedProtocolError):
await pyatv.connect(
conf, self.loop, protocol=Protocol.AirPlay)
async def test_pair_missing_service(self):
conf = AppleTV('1.2.3.4', 'Apple TV')
with self.assertRaises(exceptions.NoServiceError):
await pyatv.pair(conf, Protocol.DMAP, self.loop)
def setUp(self):
AioHTTPTestCase.setUp(self)
self.pairing = None
self.service = DmapService(
'dmap_id', PAIRING_GUID, port=self.server.port)
self.conf = AppleTV('127.0.0.1', 'Apple TV')
self.conf.add_service(self.service)
# TODO: currently stubs internal method, should provide stub
# for netifaces later
pairing._get_private_ip_addresses = \
lambda: [ipaddress.ip_address('10.0.0.1')]
self.zeroconf = zeroconf_stub.stub(pyatv.dmap.pairing)
def create_conf(name, address, *services):
"""Create an Apple TV configuration."""
atv = conf.AppleTV(name, address)
for service in services:
atv.add_service(service)
return atv
async def setUpAsync(self):
await super().setUpAsync()
self.conf = AppleTV('127.0.0.1', 'Test device')
self.conf.add_service(MrpService('mrp_id', self.fake_atv.port))
self.conf.add_service(AirPlayService(
'airplay_id', self.server.port, DEVICE_CREDENTIALS))
self.atv = await self.get_connected_device()
async def setUp(self):
self.service = conf.DmapService(None, None)
self.config = conf.AppleTV('Apple TV', '127.0.0.1')
self.config.add_service(self.service)
self.zeroconf = zeroconf_stub.stub(pairing)
self.pairing = None
# TODO: currently stubs internal method, should provide stub
# for netifaces later
pairing._get_private_ip_addresses = \
lambda: [ipaddress.ip_address('10.0.0.1')]
def _manual_device(args):
config = AppleTV(args.address, args.name)
if args.dmap_credentials or args.protocol == const.Protocol.DMAP:
config.add_service(DmapService(
args.id, args.dmap_credentials, port=args.port))
if args.mrp_credentials or args.protocol == const.Protocol.MRP:
config.add_service(MrpService(
args.id, args.port, credentials=args.mrp_credentials))
if args.airplay_credentials:
config.add_service(AirPlayService(
args.id, credentials=args.airplay_credentials))
return config
async def print_what_is_playing(loop):
"""Connect to device and print what is playing."""
config = conf.AppleTV(ADDRESS, NAME)
config.add_service(conf.DmapService('some_id', HSGID))
print('Connecting to {0}'.format(config.address))
atv = await connect(config, loop)
try:
print(await atv.metadata.playing())
finally:
# Do not forget to close
await atv.close()