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_cache(self):
cache = Cache()
cache.set_data("name", CCInfo("192.168.0.6", 8009, "Fake Factory Inc.", "Fakecast", "fake"))
self.assertEqual(
cache.get_data("name").all_info,
{
"ip": "192.168.0.6",
"port": 8009,
"manufacturer": "Fake Factory Inc.",
"model_name": "Fakecast",
"cast_type": "fake",
},
)
cache.clear()
cache = Cache()
self.assertEqual(cache.get_data("name"), None)
cache.clear()
:param device: Can be an ip-address or a name.
:type device: str
:returns: Chromecast object for use in a CastController,
and CCInfo object for use in setup_cast and StreamInfo
:rtype: (pychromecast.Chromecast, CCInfo)
"""
cast = None
if device and is_ipaddress(device):
cast = get_chromecast_with_ip(device, DEFAULT_PORT)
if not cast:
msg = "No device found at {}".format(device)
raise CastError(msg)
cc_info = CCInfo(cast.host, cast.port, None, None, cast.cast_type)
else:
cache = Cache()
cc_info = cache.get_data(device)
if cc_info:
cast = get_chromecast_with_ip(cc_info.ip, cc_info.port)
if not cast:
cast = get_chromecast(device)
if not cast:
msg = 'Specified device "{}" not found'.format(device) if device else "No devices found"
raise CastError(msg)
cc_info = CCInfo(cast.host, cast.port, cast.device.manufacturer, cast.model_name, cast.cast_type)
cache.set_data(cast.name, cc_info)
cast.wait()
return (cast, cc_info)
if not cast:
msg = "No device found at {}".format(device)
raise CastError(msg)
cc_info = CCInfo(cast.host, cast.port, None, None, cast.cast_type)
else:
cache = Cache()
cc_info = cache.get_data(device)
if cc_info:
cast = get_chromecast_with_ip(cc_info.ip, cc_info.port)
if not cast:
cast = get_chromecast(device)
if not cast:
msg = 'Specified device "{}" not found'.format(device) if device else "No devices found"
raise CastError(msg)
cc_info = CCInfo(cast.host, cast.port, cast.device.manufacturer, cast.model_name, cast.cast_type)
cache.set_data(cast.name, cc_info)
cast.wait()
return (cast, cc_info)
def __init__(self):
vhash = hashlib.sha1(__version__.encode()).hexdigest()[:8]
cache_path = Path(tempfile.gettempdir(), "catt_{}_cache".format(vhash), "chromecast_hosts")
super(Cache, self).__init__(cache_path)
self._create_store_dir()
if not self.store_path.is_file():
devices = get_chromecasts()
cache_data = {
d.name: CCInfo(d.host, d.port, d.device.manufacturer, d.model_name, d.cast_type).all_info
for d in devices
}
self._write_store(cache_data)
def get_data(self, name: str): # type: ignore
data = self._read_store()
# In the case that cache has been initialized with no cc's on the
# network, we need to ensure auto-discovery.
if not data:
return None
if name:
fetched = data.get(name)
else:
# When the user does not specify a device, we need to make an attempt
# to consistently return the same IP, thus the alphabetical sorting.
fetched = data[min(data, key=str)]
return CCInfo(**fetched) if fetched else None