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_unload(self):
with OpsDroid() as opsdroid:
mock_connector = Connector({}, opsdroid=opsdroid)
mock_connector.disconnect = amock.CoroutineMock()
opsdroid.connectors = [mock_connector]
mock_database = Database({})
mock_database.disconnect = amock.CoroutineMock()
opsdroid.memory.databases = [mock_database]
mock_skill = amock.Mock(config={"name": "mockskill"})
opsdroid.skills = [mock_skill]
opsdroid.web_server = Web(opsdroid)
opsdroid.web_server.stop = amock.CoroutineMock()
mock_web_server = opsdroid.web_server
opsdroid.cron_task = amock.CoroutineMock()
opsdroid.cron_task.cancel = amock.CoroutineMock()
mock_cron_task = opsdroid.cron_task
async def task():
await asyncio.sleep(0.5)
t = asyncio.Task(task(), loop=self.loop)
await opsdroid.unload()
self.assertTrue(t.cancel())
self.assertTrue(mock_connector.disconnect.called)
async def test_web_stats_handler(self):
"""Check the stats handler."""
with OpsDroid() as opsdroid:
opsdroid.config["web"] = {}
app = web.Web(opsdroid)
self.assertEqual(
type(await app.web_stats_handler(None)), aiohttp.web.Response
)
async def test_web_stop(self):
"""Check the stats handler."""
with OpsDroid() as opsdroid:
app = web.Web(opsdroid)
app.runner = amock.CoroutineMock()
app.runner.cleanup = amock.CoroutineMock()
await app.stop()
self.assertTrue(app.runner.cleanup.called)
async def test_web_build_response(self):
"""Check the response builder."""
with OpsDroid() as opsdroid:
opsdroid.config["web"] = {}
app = web.Web(opsdroid)
response = {"test": "test"}
resp = app.build_response(200, response)
self.assertEqual(type(resp), aiohttp.web.Response)
async def test_web(self):
"""Create a web object and check the config."""
with OpsDroid() as opsdroid:
app = web.Web(opsdroid)
self.assertEqual(app.config, {})
async def test_match_webhook_response(self):
with OpsDroid() as opsdroid:
opsdroid.loader.current_import_config = {"name": "testhook"}
opsdroid.web_server = Web(opsdroid)
opsdroid.web_server.web_app = mock.Mock()
webhook = "test"
decorator = matchers.match_webhook(webhook)
opsdroid.skills.append(decorator(await self.getMockSkill()))
opsdroid.skills[0].config = {"name": "mockedskill"}
opsdroid.web_server.setup_webhooks(opsdroid.skills)
postcalls, _ = opsdroid.web_server.web_app.router.add_post.call_args_list[0]
wrapperfunc = postcalls[1]
webhookresponse = await wrapperfunc(None)
self.assertEqual(type(webhookresponse), aiohttp.web.Response)
async def test_web_get_host(self):
"""Check the host getter."""
with OpsDroid() as opsdroid:
opsdroid.config["web"] = {}
app = web.Web(opsdroid)
self.assertEqual(app.get_host, "0.0.0.0")
opsdroid.config["web"] = {"host": "127.0.0.1"}
app = web.Web(opsdroid)
self.assertEqual(app.get_host, "127.0.0.1")
async def test_web_get_port(self):
"""Check the port getter."""
with OpsDroid() as opsdroid:
opsdroid.config["web"] = {}
app = web.Web(opsdroid)
self.assertEqual(app.get_port, 8080)
opsdroid.config["web"] = {"port": 8000}
app = web.Web(opsdroid)
self.assertEqual(app.get_port, 8000)
async def test_web_index_handler(self):
"""Check the index handler."""
with OpsDroid() as opsdroid:
opsdroid.config["web"] = {}
app = web.Web(opsdroid)
self.assertEqual(
type(await app.web_index_handler(None)), aiohttp.web.Response
)