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 get_ota_mock(*args):
if upgradeable:
img = mock.MagicMock()
img.should_update.return_value = True
img.key.manufacturer_id = mock.sentinel.manufacturer_id
img.key.image_type = mock.sentinel.image_type
img.version = mock.sentinel.image_version
img.header.image_size = mock.sentinel.image_size
elif has_image:
img = mock.MagicMock()
img.should_update.return_value = False
else:
img = None
return img
async def test_respond(self):
"""Test that responding sends a message."""
post_response = amock.Mock()
post_response.status = 200
with OpsDroid() as opsdroid, amock.patch(
"aiohttp.ClientSession.post", new=asynctest.CoroutineMock()
) as patched_request:
self.assertTrue(opsdroid.__class__.instances)
connector = ConnectorFacebook({}, opsdroid=opsdroid)
room = "a146f52c-548a-11e8-a7d1-28cfe949e12d"
test_message = Message(
text="Hello world", user="Alice", target=room, connector=connector
)
patched_request.return_value = asyncio.Future()
patched_request.return_value.set_result(post_response)
await test_message.respond("Response")
self.assertTrue(patched_request.called)
async def test_parse_crontab_timezone(self):
with OpsDroid() as opsdroid:
self.not_first_run_flag = True
opsdroid.eventloop.is_running = self.true_once
opsdroid.run_skill = amock.CoroutineMock()
with amock.patch("asyncio.sleep"):
mock_skill = await self.getMockSkill()
mock_skill.config = {"name": "greetings"}
opsdroid.skills.append(
match_crontab("* * * * *", timezone="Europe/London")(mock_skill)
)
await parse_crontab(opsdroid)
self.assertTrue(opsdroid.run_skill.called)
async def test_parse_always_raises(self):
with OpsDroid() as opsdroid:
mock_skill = await self.getRaisingMockSkill()
mock_skill.config = {"name": "greetings"}
opsdroid.skills.append(match_always()(mock_skill))
self.assertEqual(len(opsdroid.skills), 1)
mock_connector = amock.MagicMock()
mock_connector.send = amock.CoroutineMock()
message = Message(
text="Hello world",
user="user",
target="default",
connector=mock_connector,
)
await parse_always(opsdroid, message)
self.assertLogs("_LOGGER", "exception")
listen.listener_event("non_existing_event", "test2")
listener.event.assert_has_calls(
[mock.call("test1"), mock.call("test1")], any_order=True
)
context_listener.event.assert_has_calls(
[mock.call(listen, "test1")], any_order=True
)
broken_listener.event.assert_has_calls([mock.call("test1")], any_order=True)
assert listener.event.call_count == 2
assert context_listener.event.call_count == 1
assert broken_listener.event.call_count == 1
class Logger(util.LocalLogMixin):
log = mock.MagicMock()
def test_log():
log = Logger()
log.debug("Test debug")
log.exception("Test exception")
log.info("Test info")
log.warning("Test warn")
log.error("Test error")
@pytest.mark.skipif(
sys.version_info < (3, 8), reason="logging stacklevel kwarg was introduced in 3.8"
)
def test_log_stacklevel():
class MockHandler(logging.Handler):
def setUp(self):
self.config = conf.AppleTV('address', 'name')
self.mock_device = asynctest.mock.Mock(MockAppleTV())
async def test_make_filter(self):
with amock.patch(api_string.format("create_filter")) as patched_filter:
patched_filter.return_value = asyncio.Future()
patched_filter.return_value.set_result({"filter_id": "arbitrary string"})
filter_id = await self.connector.make_filter(self.api)
assert filter_id == "arbitrary string"
assert patched_filter.called
assert patched_filter.call_args[1]["user_id"] == "@opsdroid:localhost"
async def test_pass_through(client, parameters, expected):
async with MockTransport() as respx_mock:
request = respx_mock.add(**parameters)
with asynctest.mock.patch(
"asyncio.open_connection",
side_effect=ConnectionRefusedError("test request blocked"),
) as open_connection:
with pytest.raises(NetworkError):
await client.get("https://example.org/")
assert open_connection.called is True
assert request.called is True
assert request.pass_through is expected
with MockTransport() as respx_mock:
request = respx_mock.add(**parameters)
with asynctest.mock.patch(
"socket.socket.connect", side_effect=socket.error("test request blocked"),
) as connect:
@asynctest.mock.patch("mbddns.update", new=mbddns_update_mock)
async def test_update_fails_if_invalid_host(hass):
"""Run with invalid characters in host and check false is returned."""
result = await async_setup_component(
hass,
mythicbeastsdns.DOMAIN,
{
mythicbeastsdns.DOMAIN: {
"domain": "example.org",
"password": "correct",
"host": "$hass",
}
},
)
assert not result