Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
connector=mock_connector,
raw_event=raw_message,
)
self.assertEqual(message.text, "Hello world")
self.assertEqual(message.user_id, "user_id")
self.assertEqual(message.user, "user")
self.assertEqual(message.target, "default")
self.assertEqual(message.raw_event["timestamp"], "01/01/2000 19:23:00")
self.assertEqual(message.raw_event["messageId"], "101")
with self.assertRaises(TypeError):
await message.respond("Goodbye world")
# Also try responding with just some empty Event
with self.assertRaises(TypeError):
await message.respond(
events.Event(message.user, message.target, message.connector)
)
def test_unique_subclasses(self):
with self.assertRaises(NameError):
class Message(events.Event):
pass
class Message(events.Event): # noqa
_no_register = True
pass
# If we don't have the event call the unknown event coroutine
cls.events = collections.defaultdict(lambda: cls._unknown_event)
for event_method in event_methods:
for event_type in event_method.__opsdroid_events__:
if not issubclass(event_type, Event):
err_msg = (
"The event type {event_type} is "
"not a valid OpsDroid event type"
)
raise TypeError(err_msg.format(event_type=event_type))
if event_method.__opsdroid_match_subclasses__:
# Register all event types which are a subclass of this
# one.
for event in Event.event_registry.values():
if issubclass(event, event_type):
cls.events[event] = event_method
else:
cls.events[event_type] = event_method
return super().__new__(cls)
"""Event class to represent the naming of a room."""
def __init__(self, name, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
self.name = name
class RoomAddress(Event):
"""Event class to represent a room's address being changed."""
def __init__(self, address, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
self.address = address
class RoomImage(Event):
"""Event class to represent a room's display image being changed."""
def __init__(self, room_image, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
if not isinstance(room_image, Image):
raise TypeError(
"Room image must be an opsdroid.events.Image instance"
) # pragma: no cover
self.room_image = room_image
class RoomDescription(Event):
"""Event class to represent a room's description being changed."""
def __init__(self, description, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
async def update_entity(self, name, value, confidence=None):
"""Add or update an entitiy.
Adds or updates an entitiy entry for an event.
Args:
name (string): String name of entity
value (string): String value of entity
confidence (float, optional): Confidence that entity is correct
"""
self.entities[name] = {"value": value, "confidence": confidence}
class OpsdroidStarted(Event):
"""An event to indicate that Opsdroid has loaded."""
class Message(Event):
"""A message object.
Stores messages in a format that allows OpsDroid to respond or react with
delays for thinking and typing as defined in configuration YAML file.
Args:
text (string): String text of message
room (string, optional): String name of the room or chat channel in
which message was sent
connector (Connector, optional): Connector object used to interact with
given chat service
raw_event (dict, optional): Raw message as provided by chat service.
matched. E.g. a regular expression or natural language intent
responded_to: Boolean initialized as False. True if event has been
responded to
"""
def __init__(self, blocks, *args, **kwargs):
"""Create object with minimum properties."""
super().__init__("", *args, **kwargs)
self.blocks = blocks
if isinstance(self.blocks, list):
self.blocks = json.dumps(self.blocks)
class InteractiveAction(Event):
"""Super class to represent Slack interactive actions."""
def __init__(self, payload, *args, **kwargs):
"""Create object with minimum properties."""
super().__init__(*args, **kwargs)
self.payload = payload
self.ssl_context = ssl.create_default_context(cafile=certifi.where())
async def respond(self, response_event):
"""Respond to this message using the response_url field in the payload."""
if isinstance(response_event, str):
if "response_url" in self.payload:
async with aiohttp.ClientSession() as session:
headers = {"Content-Type": "application/json"}
response = await session.post(
confidence = results[0].confidence
results = filter(lambda x: x.confidence == confidence, results)
results = list(filter(lambda x: bool(x.mime_type), results))
return results[0].mime_type
class Image(File):
"""Event class specifically for image files."""
async def get_dimensions(self):
"""Return the image dimensions `(w,h)`."""
fbytes = await self.get_file_bytes()
return get_image_size_from_bytesio(io.BytesIO(fbytes), len(fbytes))
class NewRoom(Event):
"""Event class to represent the creation of a new room."""
def __init__(self, name=None, params=None, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
self.name = name
self.room_params = params or {}
class RoomName(Event):
"""Event class to represent the naming of a room."""
def __init__(self, name, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
self.name = name
self.address = address
class RoomImage(Event):
"""Event class to represent a room's display image being changed."""
def __init__(self, room_image, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
if not isinstance(room_image, Image):
raise TypeError(
"Room image must be an opsdroid.events.Image instance"
) # pragma: no cover
self.room_image = room_image
class RoomDescription(Event):
"""Event class to represent a room's description being changed."""
def __init__(self, description, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
self.description = description
class JoinRoom(Event):
"""Event class to represent a user joining a room."""
class UserInvite(Event):
"""Event class to represent a user being invited to a room."""
class UserRole(Event):
async def match_event(event, event_opts):
"""Filter and matches the event."""
event_type = event_opts.get("type", None)
if event_type:
# The event type can be specified with a string
if isinstance(event_type, str):
# pylint: disable=invalid-name
et = Event.event_registry.get(event_type, None)
if et is None:
raise ValueError(
"{event_type} is not a valid opsdroid"
" event representation.".format(event_type=event_type)
)
event_type = et
# TODO: Add option to match all subclasses as well
# if isinstance(event, event_type):
# pylint: disable=unidiomatic-typecheck
if type(event) is event_type:
for key in event_opts:
if key != "type":
event_value = event_opts.get(key, None)
entity_value = event.entities.get(key, {}).get("value", None)
"""Event class to represent a room's description being changed."""
def __init__(self, description, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
self.description = description
class JoinRoom(Event):
"""Event class to represent a user joining a room."""
class UserInvite(Event):
"""Event class to represent a user being invited to a room."""
class UserRole(Event):
"""Event class to represent a user's role or powers in a room being changed."""
def __init__(self, role, *args, **kwargs): # noqa: D107
super().__init__(*args, **kwargs)
self.role = role
class JoinGroup(Event):
"""
Event to represent joining a group (not a room).