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_parse_regex_named_groups_entities(self):
with OpsDroid() as opsdroid:
regex = r"Hello (?P.*)"
mock_skill_named_groups = await self.getMockSkill()
opsdroid.skills.append(match_regex(regex)(mock_skill_named_groups))
mock_connector = amock.CoroutineMock()
message = Message("Hello opsdroid", "user", "default", mock_connector)
[skill] = await opsdroid.get_ranked_skills(opsdroid.skills, message)
parsed_message = skill["message"]
self.assertEqual(len(parsed_message.entities.keys()), 1)
self.assertTrue("name" in parsed_message.entities.keys())
self.assertEqual(parsed_message.entities["name"]["value"], "opsdroid")
async def test_parse_regex_priority_low(self):
with OpsDroid() as opsdroid:
regex = r"(.*)"
mock_skill_low = await self.getMockSkill()
opsdroid.skills.append(match_regex(regex, score_factor=0.6)(mock_skill_low))
mock_connector = amock.CoroutineMock()
message = Message("Hello world", "user", "default", mock_connector)
skills = await opsdroid.get_ranked_skills(opsdroid.skills, message)
self.assertEqual(mock_skill_low, skills[0]["skill"])
async def test_parse_regex_raises(self):
with OpsDroid() as opsdroid:
mock_skill = await self.getRaisingMockSkill()
opsdroid.skills.append(match_regex(r"(.*)")(mock_skill))
self.assertEqual(len(opsdroid.skills), 1)
mock_connector = amock.MagicMock()
mock_connector.send = amock.CoroutineMock()
message = Message("Hello world", "user", "default", mock_connector)
skills = await parse_regex(opsdroid, opsdroid.skills, message)
self.assertEqual(mock_skill, skills[0]["skill"])
async def test_parse_regex_method_skill(self):
with OpsDroid() as opsdroid:
regex = r"Hello .*"
mock_connector = Connector({}, opsdroid=opsdroid)
mock_connector.send = amock.CoroutineMock()
skill = await self.getMockMethodSkill()
opsdroid.skills.append(match_regex(regex)(skill))
message = Message(
text="Hello world",
user="user",
target="default",
connector=mock_connector,
)
await opsdroid.parse(message)
self.assertTrue(mock_connector.send.called)
async def test_parse_regex_priority_high(self):
with OpsDroid() as opsdroid:
regex = r"(.*)"
mock_skill_high = await self.getMockSkill()
opsdroid.skills.append(match_regex(regex, score_factor=1)(mock_skill_high))
mock_connector = amock.CoroutineMock()
message = Message("Hello world", "user", "default", mock_connector)
skills = await opsdroid.get_ranked_skills(opsdroid.skills, message)
self.assertEqual(mock_skill_high, skills[0]["skill"])
@match_regex("!inviteall")
@constrain_connectors("matrix")
@ignore_appservice_users
async def on_invite_all(self, message):
await message.respond("Inviting you to all rooms...")
rooms = await self.get_all_community_rooms()
for r in rooms:
# If the room is archived don't invite people to it.
# with self.memory[r]:
# archived = await self.opsdroid.memory.get("is_archived")
# _LOGGER.debug(f"Room {r} reports archived is {archived}")
# if archived:
# continue
await message.respond(UserInvite(user=message.raw_event['sender'],
target=r,
connector=self.matrix_connector))
@match_regex("!autoinvite disable")
@constrain_connectors("matrix")
@ignore_appservice_users
async def on_disable_auto_invite(self, message):
sender = message.raw_event['sender']
users = await self.opsdroid.memory.get("autoinvite_users") or []
if sender not in users:
return await message.respond("You do not have autoinvite enabled.")
users.remove(sender)
await self.opsdroid.memory.put("autoinvite_users", users)
return await message.respond("Autoinvite disabled.")
@match_regex("!autoinvite")
@constrain_connectors("matrix")
@ignore_appservice_users
async def on_auto_invite(self, message):
sender = message.raw_event['sender']
users = await self.opsdroid.memory.get("autoinvite_users") or []
if sender in users:
return await message.respond("You already have autoinvite enabled.")
users.append(sender)
await self.opsdroid.memory.put("autoinvite_users", users)
return await message.respond(
"You will be invited to all future rooms. Use !inviteall to get invites to existing rooms.")