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_instance_creation():
assert isinstance(Cache(), TimedCache)
from soco import config
config.CACHE_ENABLED = False
assert isinstance(Cache(), NullCache)
config.CACHE_ENABLED = True
*args):
"""Test a advertise address config'd by the HASS config file."""
discover_mock.return_value = {SoCoMock('192.0.2.1')}
config = {
DOMAIN: {
CONF_PLATFORM: 'sonos',
CONF_ADVERTISE_ADDR: '192.0.1.1',
}
}
assert setup_component(self.hass, DOMAIN, config)
self.assertEqual(len(sonos.DEVICES), 1)
self.assertEqual(discover_mock.call_count, 1)
self.assertEqual(soco.config.EVENT_ADVERTISE_IP, '192.0.1.1')
def test_instance_creation():
assert isinstance(Cache(), TimedCache)
from soco import config
config.CACHE_ENABLED = False
assert isinstance(Cache(), NullCache)
config.CACHE_ENABLED = True
def __init__(self):
#: `bool`: Indicates whether the server is currently running
self.is_running = False
self._start_lock = threading.Lock()
#: `tuple`: The address (ip, port) on which the server is
#: configured to listen.
# Empty for the moment. (It is set in `start`)
self.address = ()
#: `int`: Port on which to listen.
self.requested_port_number = config.EVENT_LISTENER_PORT
service = self.service
# The Event Listener must be running, so start it if not
# pylint: disable=no-member
if not self.event_listener.is_running:
self.event_listener.start(service.soco)
# an event subscription looks like this:
# SUBSCRIBE publisher path HTTP/1.1
# HOST: publisher host:publisher port
# CALLBACK:
# NT: upnp:event
# TIMEOUT: Second-requested subscription duration (optional)
# pylint: disable=unbalanced-tuple-unpacking
ip_address, port = self.event_listener.address
if config.EVENT_ADVERTISE_IP:
ip_address = config.EVENT_ADVERTISE_IP
headers = {
'Callback': ''.format(ip_address, port),
'NT': 'upnp:event'
}
if requested_timeout is not None:
headers["TIMEOUT"] = "Second-{}".format(requested_timeout)
# pylint: disable=missing-docstring
def success(headers):
self.sid = headers['sid']
timeout = headers['timeout']
# According to the spec, timeout can be "infinite" or "second-123"
# where 123 is a number of seconds. Sonos uses "Second-123"
# (with a capital letter)
(default)
Make sure that your firewall allows connections to this port
Args:
any_zone (SoCo): Any Sonos device on the network. It does not
matter which device. It is used only to find a local IP address
reachable by the Sonos net.
"""
# Find our local network IP address which is accessible to the
# Sonos net, see http://stackoverflow.com/q/166506
temp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
temp_sock.connect((any_zone.ip_address, config.EVENT_LISTENER_PORT))
ip_address = temp_sock.getsockname()[0]
temp_sock.close()
# Start the event listener server in a separate thread.
self.address = (ip_address, config.EVENT_LISTENER_PORT)
self._listener_thread = EventServerThread(self.address)
self._listener_thread.daemon = True
self._listener_thread.start()
self.is_running = True
log.info("Event listener started")
Args:
any_zone (SoCo): Any Sonos device on the network. It does not
matter which device. It is used only to find a local IP address
reachable by the Sonos net.
"""
# Find our local network IP address which is accessible to the
# Sonos net, see http://stackoverflow.com/q/166506
temp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
temp_sock.connect((any_zone.ip_address, config.EVENT_LISTENER_PORT))
ip_address = temp_sock.getsockname()[0]
temp_sock.close()
# Start the event listener server in a separate thread.
self.address = (ip_address, config.EVENT_LISTENER_PORT)
self._listener_thread = EventServerThread(self.address)
self._listener_thread.daemon = True
self._listener_thread.start()
self.is_running = True
log.info("Event listener started")