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(self):
opsdroid = amock.CoroutineMock()
database = RedisDatabase({}, opsdroid=opsdroid)
with amock.patch.object(aioredis, "create_pool") as mocked_connection:
mocked_connection.side_effect = NotImplementedError
with suppress(NotImplementedError):
await database.connect()
self.assertTrue(mocked_connection.called)
self.assertLogs("_LOGGER", "info")
@patch('pyatv.connect')
def test_auto_connect_with_device(self, connect_func, scan_func):
scan_func.return_value = [self.config]
connect_func.return_value = self.mock_device
self.found_device = None
async def found_handler(atv):
self.found_device = atv
helpers.auto_connect(found_handler)
self.assertEqual(self.found_device, self.mock_device)
self.assertTrue(self.mock_device.logout.called)
async def test_web_start(self):
"""Check the stats handler."""
with OpsDroid() as opsdroid:
with amock.patch("aiohttp.web.AppRunner.setup") as mock_runner, amock.patch(
"aiohttp.web.TCPSite.__init__"
) as mock_tcpsite, amock.patch(
"aiohttp.web.TCPSite.start"
) as mock_tcpsite_start:
mock_tcpsite.return_value = None
app = web.Web(opsdroid)
await app.start()
self.assertTrue(mock_runner.called)
self.assertTrue(mock_tcpsite.called)
self.assertTrue(mock_tcpsite_start.called)
state = hass.states.get(ent1.entity_id)
assert STATE_ON == state.state
assert state.attributes.get("color_temp") is None
test_time = dt_util.utcnow().replace(hour=8, minute=30, second=0)
sunset_time = test_time.replace(hour=17, minute=0, second=0)
sunrise_time = test_time.replace(hour=5, minute=0, second=0)
def event_date(hass, event, now=None):
if event == SUN_EVENT_SUNRISE:
return sunrise_time
return sunset_time
with patch(
"homeassistant.components.flux.switch.dt_utcnow", return_value=test_time
), patch(
"homeassistant.components.flux.switch.get_astral_event_date",
side_effect=event_date,
):
assert await async_setup_component(
hass,
switch.DOMAIN,
{
switch.DOMAIN: {
"platform": "flux",
"name": "flux",
"lights": [ent1.entity_id],
"mode": "mired",
}
},
)
turn_on_calls = async_mock_service(hass, light.DOMAIN, SERVICE_TURN_ON)
async def test_connect(self):
with amock.patch("aiohttp.ClientSession.get") as patched_request:
mockresponse = amock.CoroutineMock()
mockresponse.status = 200
mockresponse.json = amock.CoroutineMock(return_value={"login": "opsdroid"})
patched_request.return_value = asyncio.Future()
patched_request.return_value.set_result(mockresponse)
await self.connector.connect()
@patch('os.urandom')
async def test_pairing_with_device_new_credentials(self, rand_func):
rand_func.side_effect = predetermined_key
await self.do_pairing()
async def test_web_start(self):
"""Check the stats handler."""
with OpsDroid() as opsdroid:
with amock.patch("aiohttp.web.AppRunner.setup") as mock_runner, amock.patch(
"aiohttp.web.TCPSite.__init__"
) as mock_tcpsite, amock.patch(
"aiohttp.web.TCPSite.start"
) as mock_tcpsite_start:
mock_tcpsite.return_value = None
app = web.Web(opsdroid)
await app.start()
self.assertTrue(mock_runner.called)
self.assertTrue(mock_tcpsite.called)
self.assertTrue(mock_tcpsite_start.called)
async def test_get_config_flows(hass):
"""Verify that custom components with config_flow are available."""
test_1_integration = _get_test_integration(hass, "test_1", False)
test_2_integration = _get_test_integration(hass, "test_2", True)
with patch("homeassistant.loader.async_get_custom_components") as mock_get:
mock_get.return_value = {
"test_1": test_1_integration,
"test_2": test_2_integration,
}
flows = await loader.async_get_config_flows(hass)
assert "test_2" in flows
assert "test_1" not in flows
@mock.patch("aiosfstream.client.Client.create_replay_storage")
def test_init_creates_replay_storage(self, create_replay_storage):
replay_param = object()
create_replay_storage.return_value = mock.create_autospec(
ReplayMarkerStorage
)()
client = Client(self.authenticator,
replay=replay_param)
self.assertEqual(client.auth, self.authenticator)
self.assertEqual(client.extensions,
[create_replay_storage.return_value])
self.assertEqual(client.replay_storage,
create_replay_storage.return_value)
self.assertIsNone(client.replay_fallback)
create_replay_storage.assert_called_with(replay_param)
@patch.object(ActivityCount, 'get_json')
async def test_get_all_local(self, _mocked_func):
_mocked_func.side_effect = self.mocked_get_json
for metrics in self.activity.responses:
provider_data = [p for p in providers if p["provider"] == metrics["provider"]]
test_data = getattr(self, metrics["provider"], None)
test_metric_count = jmespath.search(provider_data[0]["metrics"][0]["path"], test_data)
returned_metric_count = metrics["metrics"][0]["count"]
assert test_metric_count == returned_metric_count