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_train_rasanlu(self):
with OpsDroid() as opsdroid:
opsdroid.config["parsers"] = {"rasanlu": {"enabled": True}}
with amock.patch("opsdroid.parsers.rasanlu.train_rasanlu"):
await opsdroid.train_parsers({})
async def test_parse_witai_raises(self):
with OpsDroid() as opsdroid:
opsdroid.config["parsers"] = [
{"name": "witai", "token": "test", "min-score": 0.3}
]
mock_skill = await self.getRaisingMockSkill()
mock_skill.config = {"name": "mocked-skill"}
opsdroid.skills.append(match_witai("get_weather")(mock_skill))
mock_connector = amock.MagicMock()
mock_connector.send = amock.CoroutineMock()
message = Message(
text="how's the weather outside",
user="user",
target="default",
connector=mock_connector,
)
def test_start_from_path(self):
runner = CliRunner()
with mock.patch.object(OpsDroid, "run") as mock_run:
runner.invoke(
opsdroid.cli.start,
["-f", os.path.abspath("tests/configs/full_valid.yaml")],
)
assert mock_run.called
async def test_connect_failure(self):
result = amock.MagicMock()
result.status = 401
with OpsDroid() as opsdroid, amock.patch(
"aiohttp.ClientSession.get"
) as patched_request:
patched_request.return_value = asyncio.Future()
patched_request.return_value.set_result(result)
await self.connector.connect()
self.assertLogs("_LOGGER", "error")
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")
async def test_parse_witai_entities(self):
with OpsDroid() as opsdroid:
opsdroid.config["parsers"] = [
{"name": "witai", "token": "test", "min-score": 0.3}
]
mock_skill = await self.getMockSkill()
opsdroid.skills.append(match_witai("get_weather")(mock_skill))
mock_connector = amock.CoroutineMock()
message = Message(
text="what is the weather in london",
user="user",
target="default",
connector=mock_connector,
)
with amock.patch.object(witai, "call_witai") as mocked_call_witai:
mocked_call_witai.return_value = {
def setUp(self):
configure_lang({})
self.connector = RocketChat(
{
"name": "rocket.chat",
"token": "test",
"user-id": "userID",
"default_target": "test",
},
opsdroid=OpsDroid(),
)
self.connector.latest_update = "2018-10-08T12:57:37.126Z"
with amock.patch("aiohttp.ClientSession") as mocked_session:
self.connector.session = mocked_session
def test_core(self):
with OpsDroid() as opsdroid:
self.assertIsInstance(opsdroid, OpsDroid)
you that the path doesn't exist. Also, the file needs to be either a json or a yaml file.
Args:
ctx (:obj:`click.Context`): The current click cli context.
path (string): a string representing the path to load the config,
obtained from `ctx.obj`.
value (string): the value of this parameter after invocation.
It is either "config" or "log" depending on the program
calling this function.
Returns:
int: the exit code. Always returns 0 in this case.
"""
with OpsDroid() as opsdroid:
loader = Loader(opsdroid)
config = load_config_file([path] if path else DEFAULT_CONFIG_LOCATIONS)
loader.load_modules_from_config(config)
click.echo("Configuration validated - No errors founds!")
ctx.exit(0)