Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_create_join_leave_destroy_lifecycle(testbot): # noqa
rooms = testbot.bot.rooms()
assert len(rooms) == 1
r1 = rooms[0]
assert str(r1) == "testroom"
assert issubclass(r1.__class__, errbot.backends.base.Room)
r2 = testbot.bot.query_room('testroom2')
assert not r2.exists
r2.create()
assert r2.exists
rooms = testbot.bot.rooms()
assert r2 not in rooms
assert not r2.joined
r2.destroy()
rooms = testbot.bot.rooms()
assert r2 not in rooms
r2.join()
assert r2.exists
node_domain, resource = hipchat_user['xmpp_jid'].split('/')
else:
node_domain = hipchat_user['xmpp_jid']
resource = hipchat_user['name']
node, domain = node_domain.split('@')
self._aclattr = aclattr
super().__init__(node, domain, resource, room)
@property
def aclattr(self):
return self._aclattr
class HipChatRoom(Room):
"""
This class represents a Multi-User Chatroom.
"""
def __init__(self, name, bot):
"""
:param name:
The name of the room
"""
self.name = name
self.hypchat = bot.conn.hypchat
self.xep0045 = bot.conn.client.plugin['xep_0045']
self._bot = bot
@property
def room(self):
fullname += " " + self.last_name
return fullname
@property
def username(self):
return self._username
@property
def client(self):
return None
person = id
nick = username
class TelegramRoom(TelegramIdentifier, Room):
def __init__(self, id, title=None):
super().__init__(id)
self._title = title
@property
def id(self):
return self._id
@property
def title(self):
"""Return the groupchat title (only applies to groupchats)"""
return self._title
def join(self, username: str = None, password: str = None):
raise RoomsNotSupportedError()
avatarMedium=json_user['avatarUrlMedium'])
def __unicode__(self):
if self.url == self._room._uri:
return self.username # this is a 1 to 1 MUC
return self.username + '@' + self._room.name
def __eq__(self, other):
if hasattr(other, 'person'):
return self.person == other.person
return str(self) == str(other)
__str__ = __unicode__
class GitterRoom(Room):
def invite(self, *args) -> None:
pass
def __init__(self, backend, idd, uri, name):
self._backend = backend
self._name = name
self._uri = uri
self._idd = idd
self._joined = False
def join(self, username=None, password=None):
log.debug("Joining room %s (%s)" % (self._uri, self._idd))
try:
response = self._backend.writeAPIRequest('rooms', {'uri': self._uri})
log.debug("Response: %s" % response)
def is_group(self) -> bool:
return isinstance(self.to, Room)
@property
def aclattr(self):
return self.nick
@property
def fullname(self):
return self._sparkpy_person.displayName
def __getattr__(self, attr):
return getattr(self._sparkpy_person, attr)
def __repr__(self):
return ''.format(person=self.person)
class ErrSparkRoom(Room):
'''
This object represents a Spark Room
args: a string of the roomId or a sparpy.models.room.SparkRoom
'''
def __init__(self, room, session):
if isinstance(room, sparkpy.SparkRoom):
self._sparkpy_room = room
if sparkpy.utils.is_api_id(room, 'rooms'):
self._sparkpy_room = sparkpy.SparkRoom(room, parent=session)
else:
raise TypeError('Invalid Cisco Spark Room id: ' + room)
def join(self):
raise NotImplemented('Cannot join rooms. Must be added instead')
def get_members(self, space_name: str) -> Iterable[dict]:
return self._list('spaces/{}/members'.format(space_name.lstrip('spaces/')), 'memberships')
def get_member(self, space_name: str, name: str) -> Optional[dict]:
return self._request('spaces/{}/members/{}'.format(space_name.lstrip('spaces/'), name))
def create_message(self, space_name: str, body: dict, thread_key: str = None) -> Optional[dict]:
url = 'spaces/{}/messages'.format(space_name.lstrip('spaces/'))
if thread_key is None:
return self._request(url, body=json.dumps(body), method='POST')
else:
return self._request(url, body=json.dumps(body), method='POST',
query_string='threadKey={}'.format(thread_key))
class HangoutsChatRoom(Room):
"""
Represents a 'Space' in Google-Hangouts-Chat terminology
"""
def __init__(self, space_id, chat_api):
super().__init__()
self.space_id = space_id
self.chat_api = chat_api
self._load()
def _load(self):
space = self.chat_api.get_space(self.space_id)
self.does_exist = bool(space)
self.display_name = space['displayName'] if self.does_exist else ''
def join(self, username=None, password=None):
raise RoomsNotSupportedError()
def aclattr(self):
return str(self)
def __str__(self):
return '@' + self._person
def __eq__(self, other):
if not isinstance(other, Person):
return False
return self.person == other.person
def __hash__(self):
return self.person.__hash__()
class TextRoom(Room):
def __init__(self, name, bot):
self._topic = ''
self._joined = False
self.name = name
self._bot = bot
# fill up the room with a coherent set of identities.
self._occupants = [TextOccupant('somebody', self),
TextOccupant(TextPerson(bot.bot_config.BOT_ADMINS[0]), self),
TextOccupant(bot.bot_identifier, self)]
def join(self, username=None, password=None):
self._joined = True
def leave(self, reason=None):
@property
def room(self):
return self._room
def __unicode__(self):
return "%s" % self._nickmask
def __str__(self):
return self.__unicode__()
def __repr__(self):
return "<{} - {}>".format(self.__unicode__(), super().__repr__())
class IRCRoom(Room):
"""
Represent the specifics of a IRC Room/Channel.
This lifecycle of this object is:
- Created in IRCConnection.on_join
- The joined status change in IRCConnection on_join/on_part
- Deleted/destroyed in IRCConnection.on_disconnect
"""
def __init__(self, room, bot):
self._bot = bot
self.room = room
self.connection = self._bot.conn.connection
self._topic_lock = threading.Lock()
self._topic = None
def check_identifier(self, identifier: Identifier):
is_room = isinstance(self.requestor, Room)
is_room = is_room and isinstance(identifier, RoomOccupant)
is_room = is_room and self.requestor == identifier.room
return is_room or self.requestor == identifier