PK!|``slack/__init__.pyfrom .methods import HOOK_URL, ROOT_URL # noQa from .methods import Methods as methods # noQa PK!6slack/actions.pyimport json import typing import logging from typing import Any, Dict, Iterator, Optional from collections import defaultdict from collections.abc import MutableMapping from . import exceptions LOG = logging.getLogger(__name__) class Action(MutableMapping): """ MutableMapping representing a response to an interactive message, a dialog submission or a message action. Args: raw_action: Decoded body of the HTTP request verification_token: Slack verification token used to verify the request came from slack team_id: Verify the event is for the correct team Raises: :class:`slack.exceptions.FailedVerification`: when `verification_token` or `team_id` does not match the incoming event's """ def __init__( self, raw_action: typing.MutableMapping, verification_token: Optional[str] = None, team_id: Optional[str] = None, ) -> None: self.action = raw_action if verification_token and self.action["token"] != verification_token: raise exceptions.FailedVerification( self.action["token"], self.action["team"]["id"] ) if team_id and self.action["team"]["id"] != team_id: raise exceptions.FailedVerification( self.action["token"], self.action["team"]["id"] ) def __getitem__(self, item): return self.action[item] def __setitem__(self, key, value): self.action[key] = value def __delitem__(self, key): del self.action[key] def __iter__(self): return iter(self.action) def __len__(self): return len(self.action) def __repr__(self): return str(self.action) @classmethod def from_http( cls, payload: typing.MutableMapping, verification_token: Optional[str] = None, team_id: Optional[str] = None, ) -> "Action": action = json.loads(payload["payload"]) return cls(action, verification_token=verification_token, team_id=team_id) class Router: """ When creating a slack applications you can only set one action url. This provide a routing mechanism for the incoming actions, based on their `callback_id` and the action name, to one or more handlers. """ def __init__(self): self._routes: Dict[str, Dict] = defaultdict(dict) def register(self, callback_id: str, handler: Any, name: str = "*") -> None: """ Register a new handler for a specific :class:`slack.actions.Action` `callback_id`. Optional routing based on the action name too. The name argument is useful for actions of type `interactive_message` to provide a different handler for each individual action. Args: callback_id: Callback_id the handler is interested in handler: Callback name: Name of the action (optional). """ LOG.info("Registering %s, %s to %s", callback_id, name, handler) if name not in self._routes[callback_id]: self._routes[callback_id][name] = [] self._routes[callback_id][name].append(handler) def dispatch(self, action: Action) -> Any: """ Yields handlers matching the incoming :class:`slack.actions.Action` `callback_id`. Args: action: :class:`slack.actions.Action` Yields: handler """ LOG.debug("Dispatching action %s, %s", action["type"], action["callback_id"]) if action["type"] == "interactive_message": yield from self._dispatch_interactive_message(action) elif action["type"] in ("dialog_submission", "message_action"): yield from self._dispatch_action(action) else: raise UnknownActionType(action) def _dispatch_action(self, action: Action) -> Iterator[Any]: yield from self._routes[action["callback_id"]].get("*", []) def _dispatch_interactive_message(self, action: Action) -> Iterator[Any]: if action["actions"][0]["name"] in self._routes[action["callback_id"]]: yield from self._routes[action["callback_id"]][action["actions"][0]["name"]] else: yield from self._routes[action["callback_id"]].get("*", []) class UnknownActionType(Exception): """ Raised for incoming action with unknown type Attributes: action: The incoming action """ def __init__(self, action: Action) -> None: self.action = action PK!\c slack/commands.pyimport typing import logging from typing import Any, Dict, Iterator, Optional from collections import defaultdict from collections.abc import MutableMapping from . import exceptions LOG = logging.getLogger(__name__) class Command(MutableMapping): """ MutableMapping representing a slack slash command. Args: raw_command: Decoded body of the webhook HTTP request Raises: :class:`slack.exceptions.FailedVerification`: when `verification_token` or `team_id` does not match the incoming command's """ def __init__( self, raw_command: typing.MutableMapping, verification_token: Optional[str] = None, team_id: Optional[str] = None, ) -> None: self.command = raw_command if verification_token and self.command["token"] != verification_token: raise exceptions.FailedVerification( self.command["token"], self.command["team_id"] ) if team_id and self.command["team_id"] != team_id: raise exceptions.FailedVerification( self.command["token"], self.command["team_id"] ) def __getitem__(self, item): return self.command[item] def __setitem__(self, key, value): self.command[key] = value def __delitem__(self, key): del self.command[key] def __iter__(self): return iter(self.command) def __len__(self): return len(self.command) def __repr__(self): return str(self.command) class Router: """ When creating slash command for your applications each one can have a custom webhook url. For ease of configuration this class provide a routing mechanisms based on the command so that each command can define the same webhook url. """ def __init__(self): self._routes: Dict[str, list] = defaultdict(list) def register(self, command: str, handler: Any): """ Register a new handler for a specific slash command Args: command: Slash command handler: Callback """ if not command.startswith("/"): command = f"/{command}" LOG.info("Registering %s to %s", command, handler) self._routes[command].append(handler) def dispatch(self, command: Command) -> Iterator[Any]: """ Yields handlers matching the incoming :class:`slack.actions.Command`. Args: command: :class:`slack.actions.Command` Yields: handler """ LOG.debug("Dispatching command %s", command["command"]) for callback in self._routes[command["command"]]: yield callback PK!<݀&&slack/events.pyimport re import copy import json import logging import itertools from typing import Any, Dict, Iterator, Optional, MutableMapping from collections import defaultdict from collections.abc import MutableMapping from . import exceptions LOG = logging.getLogger(__name__) class Event(MutableMapping): """ MutableMapping representing a slack event coming from the RTM API or the Event API. Attributes: metadata: Metadata dispatched with the event when using the Event API (see `slack event API documentation `_) """ def __init__( self, raw_event: MutableMapping, metadata: Optional[MutableMapping] = None ) -> None: self.event = raw_event self.metadata = metadata def __getitem__(self, item): return self.event[item] def __setitem__(self, key, value): self.event[key] = value def __delitem__(self, key): del self.event[key] def __iter__(self): return iter(self.event) def __len__(self): return len(self.event) def __repr__(self): return "Slack Event: " + str(self.event) def clone(self) -> "Event": """ Clone the event Returns: :class:`slack.events.Event` """ return self.__class__(copy.deepcopy(self.event), copy.deepcopy(self.metadata)) @classmethod def from_rtm(cls, raw_event: MutableMapping) -> "Event": """ Create an event with data coming from the RTM API. If the event type is a message a :class:`slack.events.Message` is returned. Args: raw_event: JSON decoded data from the RTM API Returns: :class:`slack.events.Event` or :class:`slack.events.Message` """ if raw_event["type"].startswith("message"): return Message(raw_event) else: return Event(raw_event) @classmethod def from_http( cls, raw_body: MutableMapping, verification_token: Optional[str] = None, team_id: Optional[str] = None, ) -> "Event": """ Create an event with data coming from the HTTP Event API. If the event type is a message a :class:`slack.events.Message` is returned. Args: raw_body: Decoded body of the Event API request verification_token: Slack verification token used to verify the request came from slack team_id: Verify the event is for the correct team Returns: :class:`slack.events.Event` or :class:`slack.events.Message` Raises: :class:`slack.exceptions.FailedVerification`: when `verification_token` or `team_id` does not match the incoming event's. """ if verification_token and raw_body["token"] != verification_token: raise exceptions.FailedVerification(raw_body["token"], raw_body["team_id"]) if team_id and raw_body["team_id"] != team_id: raise exceptions.FailedVerification(raw_body["token"], raw_body["team_id"]) if raw_body["event"]["type"].startswith("message"): return Message(raw_body["event"], metadata=raw_body) else: return Event(raw_body["event"], metadata=raw_body) class Message(Event): """ Type of :class:`slack.events.Event` corresponding to a message event type """ def __init__( self, msg: Optional[MutableMapping] = None, metadata: Optional[MutableMapping] = None, ) -> None: if not msg: msg = {} super().__init__(msg, metadata) def __repr__(self) -> str: return "Slack Message: " + str(self.event) def response(self, in_thread: Optional[bool] = None) -> "Message": """ Create a response message. Depending on the incoming message the response can be in a thread. By default the response follow where the incoming message was posted. Args: in_thread (boolean): Overwrite the `threading` behaviour Returns: a new :class:`slack.event.Message` """ data = {"channel": self["channel"]} if in_thread: if "message" in self: data["thread_ts"] = ( self["message"].get("thread_ts") or self["message"]["ts"] ) else: data["thread_ts"] = self.get("thread_ts") or self["ts"] elif in_thread is None: if "message" in self and "thread_ts" in self["message"]: data["thread_ts"] = self["message"]["thread_ts"] elif "thread_ts" in self: data["thread_ts"] = self["thread_ts"] return Message(data) def serialize(self) -> dict: """ Serialize the message for sending to slack API Returns: serialized message """ data = {**self} if "attachments" in self: data["attachments"] = json.dumps(self["attachments"]) return data def to_json(self) -> str: return json.dumps({**self}) class EventRouter: """ When receiving an event from the RTM API or the slack API it is useful to have a routing mechanisms for dispatching event to individual function/coroutine. This class provide such mechanisms for any :class:`slack.events.Event`. """ def __init__(self): self._routes: Dict[str, Dict] = defaultdict(dict) def register(self, event_type: str, handler: Any, **detail: Any) -> None: """ Register a new handler for a specific :class:`slack.events.Event` `type` (See `slack event types documentation `_ for a list of event types). The arbitrary keyword argument is used as a key/value pair to compare against what is in the incoming :class:`slack.events.Event` Args: event_type: Event type the handler is interested in handler: Callback **detail: Additional key for routing """ LOG.info("Registering %s, %s to %s", event_type, detail, handler) if len(detail) > 1: raise ValueError("Only one detail can be provided for additional routing") elif not detail: detail_key, detail_value = "*", "*" else: detail_key, detail_value = detail.popitem() if detail_key not in self._routes[event_type]: self._routes[event_type][detail_key] = {} if detail_value not in self._routes[event_type][detail_key]: self._routes[event_type][detail_key][detail_value] = [] self._routes[event_type][detail_key][detail_value].append(handler) def dispatch(self, event: Event) -> Iterator[Any]: """ Yields handlers matching the routing of the incoming :class:`slack.events.Event`. Args: event: :class:`slack.events.Event` Yields: handler """ LOG.debug('Dispatching event "%s"', event.get("type")) if event["type"] in self._routes: for detail_key, detail_values in self._routes.get( event["type"], {} ).items(): event_value = event.get(detail_key, "*") yield from detail_values.get(event_value, []) else: return class MessageRouter: """ When receiving an event of type message from the RTM API or the slack API it is useful to have a routing mechanisms for dispatching the message to individual function/coroutine. This class provide such mechanisms for any :class:`slack.events.Message`. The routing is based on regex pattern matching of the message text and the receiving channel. """ def __init__(self): self._routes: Dict[str, Dict] = defaultdict(dict) def register( self, pattern: str, handler: Any, flags: int = 0, channel: str = "*", subtype: Optional[str] = None, ) -> None: """ Register a new handler for a specific :class:`slack.events.Message`. The routing is based on regex pattern matching the message text and the incoming slack channel. Args: pattern: Regex pattern matching the message text. handler: Callback flags: Regex flags. channel: Slack channel ID. Use * for any. subtype: Message subtype """ LOG.debug('Registering message endpoint "%s: %s"', pattern, handler) match = re.compile(pattern, flags) if subtype not in self._routes[channel]: self._routes[channel][subtype] = dict() if match in self._routes[channel][subtype]: self._routes[channel][subtype][match].append(handler) else: self._routes[channel][subtype][match] = [handler] def dispatch(self, message: Message) -> Iterator[Any]: """ Yields handlers matching the routing of the incoming :class:`slack.events.Message` Args: message: :class:`slack.events.Message` Yields: handler """ if "text" in message: text = message["text"] or "" elif "message" in message: text = message["message"].get("text", "") else: text = "" msg_subtype = message.get("subtype") for subtype, matchs in itertools.chain( self._routes[message["channel"]].items(), self._routes["*"].items() ): if msg_subtype == subtype or subtype is None: for match, endpoints in matchs.items(): if match.search(text): yield from endpoints PK!J  slack/exceptions.pyimport http from typing import MutableMapping class HTTPException(Exception): """ Raised on non 200 status code Attributes: headers: Response headers data: Response data status: Response status """ def __init__( self, status: int, headers: MutableMapping, data: MutableMapping ) -> None: self.headers = headers self.data = data self.status = http.HTTPStatus(status) def __str__(self): return "{}, {}".format(self.status.value, self.status.phrase) class SlackAPIError(Exception): """ Raised for errors return by the Slack API Attributes: headers: Response headers data: Response data error: Slack API error """ def __init__( self, error: str, headers: MutableMapping, data: MutableMapping ) -> None: self.headers = headers self.data = data self.error = error def __str__(self): return str(self.error) class RateLimited(HTTPException, SlackAPIError): """ Raised when rate limited. Attributes: retry_after: Timestamp when the rate limitation ends """ def __init__( self, retry_after: int, error: str, status: int, headers: MutableMapping, data: MutableMapping, ) -> None: HTTPException.__init__(self, status=status, headers=headers, data=data) SlackAPIError.__init__(self, error=error, headers=headers, data=data) self.retry_after = retry_after def __str__(self): return HTTPException.__str__(self) + ", retry in {}s".format(self.retry_after) class InvalidRequest(Exception): """ Base class for all exception raised due to an invalid verification """ class FailedVerification(InvalidRequest): """ Raised when incoming data from Slack webhooks fail verification Attributes: token: Token that failed verification team_id: Team id that failed verification """ def __init__(self, token: str, team_id: str) -> None: self.token = token self.team_id = team_id class InvalidSlackSignature(InvalidRequest): """ Raised when the incoming request fails signature check Attributes: slack_signature: Signature sent by slack calculated_singature: Calculated signature """ def __init__(self, slack_signature: str, calculated_signature: str) -> None: self.slack_signature = slack_signature self.calculated_signature = calculated_signature class InvalidTimestamp(InvalidRequest): """ Raised when the incoming request is too old Attributes: timestamp: Timestamp of the incoming request """ def __init__(self, timestamp: float) -> None: self.timestamp = timestamp PK!slack/io/__init__.pyPK!0-B<<slack/io/abc.pyimport json import time import logging from typing import Tuple, Union, Optional, AsyncIterator, MutableMapping from .. import events, sansio, methods, exceptions LOG = logging.getLogger(__name__) class SlackAPI: """ :py:term:`abstract base class` abstracting the HTTP library used to call Slack API. Built with the functions of :mod:`slack.sansio`. Args: session: HTTP session token: Slack API token headers: Default headers for all request """ def __init__(self, *, token: str, headers: Optional[MutableMapping] = None) -> None: self._token = token self._headers = headers or {} async def _request( self, method: str, url: str, headers: Optional[MutableMapping], body: Optional[Union[str, MutableMapping]], ) -> Tuple[int, bytes, MutableMapping]: raise NotImplementedError() async def _rtm(self, url: str) -> AsyncIterator[str]: yield "" raise NotImplementedError() async def sleep(self, seconds: Union[int, float]): raise NotImplementedError() async def _make_query( self, url: str, body: Optional[Union[str, MutableMapping]], headers: Optional[MutableMapping], ) -> dict: LOG.debug("Querying %s with %s, %s", url, headers, body) status, rep_body, rep_headers = await self._request("POST", url, headers, body) LOG.debug("Response from %s: %s, %s, %s", url, status, rep_body, rep_headers) response_data = sansio.decode_response(status, rep_headers, rep_body) return response_data async def query( self, url: Union[str, methods], data: Optional[MutableMapping] = None, headers: Optional[MutableMapping] = None, as_json: Optional[bool] = None, ) -> dict: """ Query the slack API When using :class:`slack.methods` the request is made `as_json` if available Args: url: :class:`slack.methods` or url string data: JSON encodable MutableMapping headers: Custom headers as_json: Post JSON to the slack API Returns: dictionary of slack API response data """ url, body, headers = sansio.prepare_request( url=url, data=data, headers=headers, as_json=as_json, global_headers=self._headers, token=self._token, ) return await self._make_query(url, body, headers) async def iter( self, url: Union[str, methods], data: Optional[MutableMapping] = None, headers: Optional[MutableMapping] = None, *, limit: int = 200, iterkey: Optional[str] = None, itermode: Optional[str] = None, minimum_time: Optional[int] = None, as_json: Optional[bool] = None ) -> AsyncIterator[dict]: """ Iterate over a slack API method supporting pagination When using :class:`slack.methods` the request is made `as_json` if available Args: url: :class:`slack.methods` or url string data: JSON encodable MutableMapping headers: limit: Maximum number of results to return per call. iterkey: Key in response data to iterate over (required for url string). itermode: Iteration mode (required for url string) (one of `cursor`, `page` or `timeline`) minimum_time: Minimum elapsed time (in seconds) between two calls to the Slack API (default to 0). If not reached the client will sleep for the remaining time. as_json: Post JSON to the slack API Returns: Async iterator over `response_data[key]` """ itervalue = None if not data: data = {} last_request_time = None while True: current_time = time.time() if ( minimum_time and last_request_time and last_request_time + minimum_time > current_time ): await self.sleep(last_request_time + minimum_time - current_time) data, iterkey, itermode = sansio.prepare_iter_request( url, data, iterkey=iterkey, itermode=itermode, limit=limit, itervalue=itervalue, ) last_request_time = time.time() response_data = await self.query(url, data, headers, as_json) itervalue = sansio.decode_iter_request(response_data) for item in response_data[iterkey]: yield item if not itervalue: break async def rtm( self, url: Optional[str] = None, bot_id: Optional[str] = None ) -> AsyncIterator[events.Event]: """ Iterate over event from the RTM API Args: url: Websocket connection url bot_id: Connecting bot ID Returns: :class:`slack.events.Event` or :class:`slack.events.Message` """ while True: bot_id = bot_id or await self._find_bot_id() url = url or await self._find_rtm_url() async for event in self._incoming_from_rtm(url, bot_id): yield event url = None async def _find_bot_id(self) -> str: """ Find the bot ID to discard incoming message from the bot itself. Returns: The bot ID """ auth = await self.query(methods.AUTH_TEST) user_info = await self.query(methods.USERS_INFO, {"user": auth["user_id"]}) bot_id = user_info["user"]["profile"]["bot_id"] LOG.info("BOT_ID is %s", bot_id) return bot_id async def _find_rtm_url(self) -> str: """ Call `rtm.connect` to find the websocket url. Returns: Url for websocket connection """ response = await self.query(methods.RTM_CONNECT) return response["url"] async def _incoming_from_rtm( self, url: str, bot_id: str ) -> AsyncIterator[events.Event]: """ Connect and discard incoming RTM event if necessary. :param url: Websocket url :param bot_id: Bot ID :return: Incoming events """ async for data in self._rtm(url): event = events.Event.from_rtm(json.loads(data)) if sansio.need_reconnect(event): break elif sansio.discard_event(event, bot_id): continue else: yield event PK!ZvWWslack/io/aiohttp.pyimport asyncio from typing import Tuple, Union, Optional, AsyncIterator, MutableMapping import aiohttp from . import abc from .. import methods class SlackAPI(abc.SlackAPI): """ `aiohttp` implementation of :class:`slack.io.abc.SlackAPI` Args: session: HTTP session """ def __init__(self, *, session: aiohttp.ClientSession, **kwargs) -> None: self._session = session super().__init__(**kwargs) async def _request( self, method: str, url: str, headers: Optional[MutableMapping], body: Optional[Union[str, MutableMapping]], ) -> Tuple[int, bytes, MutableMapping]: async with self._session.request( method, url, headers=headers, data=body ) as response: return response.status, await response.read(), response.headers async def _rtm(self, url: str) -> AsyncIterator[str]: async with self._session.ws_connect(url) as ws: async for data in ws: if data.type == aiohttp.WSMsgType.TEXT: yield data.data elif data.type == aiohttp.WSMsgType.CLOSED: break elif data.type == aiohttp.WSMsgType.ERROR: break async def sleep(self, seconds: Union[int, float]) -> None: await asyncio.sleep(seconds) PK!K1++slack/io/curio.pyfrom typing import Tuple, Union, Optional, AsyncIterator, MutableMapping import asks import curio from . import abc class SlackAPI(abc.SlackAPI): """ `asks curio` implementation of :class:`slack.io.abc.SlackAPI` Args: session: HTTP session """ def __init__(self, *, session: asks.Session, **kwargs) -> None: self._session = session super().__init__(**kwargs) async def _request( self, method: str, url: str, headers: Optional[MutableMapping], body: Optional[Union[str, MutableMapping]], ) -> Tuple[int, bytes, MutableMapping]: response = await self._session.request(method, url, headers=headers, data=body) return response.status_code, response.content, response.headers async def rtm(self, url=None, bot_id=None): raise NotImplementedError async def _rtm(self, url: str) -> AsyncIterator[str]: yield "" raise NotImplementedError async def sleep(self, seconds: float) -> None: await curio.sleep(seconds) PK!0gslack/io/requests.pyimport json import time import logging from typing import Tuple, Union, Iterator, Optional, MutableMapping import requests import websocket from . import abc from .. import events, sansio, methods, exceptions LOG = logging.getLogger(__name__) class SlackAPI(abc.SlackAPI): """ `requests` implementation of :class:`slack.io.abc.SlackAPI` Args: session: HTTP session """ def __init__(self, *, session: Optional[requests.Session] = None, **kwargs) -> None: self._session = session or requests.session() super().__init__(**kwargs) def _request( # type: ignore self, method: str, url: str, headers: Optional[MutableMapping], body: Optional[Union[str, MutableMapping]], ) -> Tuple[int, bytes, MutableMapping]: response = self._session.request( # type: ignore method, url, headers=headers, data=body ) return response.status_code, response.content, response.headers def _rtm(self, url: str) -> Iterator[str]: # type: ignore ws = websocket.create_connection(url) while True: event = ws.recv() if event: yield event else: self.sleep(0.5) def sleep(self, seconds: Union[int, float]) -> None: # type: ignore time.sleep(seconds) def _make_query( # type: ignore self, url: str, body: Optional[Union[str, MutableMapping]], headers: Optional[MutableMapping], ) -> dict: status, rep_body, rep_headers = self._request("POST", url, headers, body) response_data = sansio.decode_response(status, rep_headers, rep_body) return response_data def query( # type: ignore self, url: Union[str, methods], data: Optional[MutableMapping] = None, headers: Optional[MutableMapping] = None, as_json: Optional[bool] = None, ) -> dict: """ Query the slack API When using :class:`slack.methods` the request is made `as_json` if available Args: url: :class:`slack.methods` or url string data: JSON encodable MutableMapping headers: Custom headers as_json: Post JSON to the slack API Returns: dictionary of slack API response data """ url, body, headers = sansio.prepare_request( url=url, data=data, headers=headers, global_headers=self._headers, token=self._token, ) return self._make_query(url, body, headers) def iter( # type: ignore self, url: Union[str, methods], data: Optional[MutableMapping] = None, headers: Optional[MutableMapping] = None, *, limit: int = 200, iterkey: Optional[str] = None, itermode: Optional[str] = None, minimum_time: Optional[int] = None, as_json: Optional[bool] = None ) -> Iterator[dict]: """ Iterate over a slack API method supporting pagination When using :class:`slack.methods` the request is made `as_json` if available Args: url: :class:`slack.methods` or url string data: JSON encodable MutableMapping headers: limit: Maximum number of results to return per call. iterkey: Key in response data to iterate over (required for url string). itermode: Iteration mode (required for url string) (one of `cursor`, `page` or `timeline`) minimum_time: Minimum elapsed time (in seconds) between two calls to the Slack API (default to 0). If not reached the client will sleep for the remaining time. as_json: Post JSON to the slack API Returns: Async iterator over `response_data[key]` """ itervalue = None if not data: data = {} last_request_time = None while True: current_time = time.time() if ( minimum_time and last_request_time and last_request_time + minimum_time > current_time ): self.sleep(last_request_time + minimum_time - current_time) data, iterkey, itermode = sansio.prepare_iter_request( url, data, iterkey=iterkey, itermode=itermode, limit=limit, itervalue=itervalue, ) last_request_time = time.time() response_data = self.query(url, data, headers, as_json) itervalue = sansio.decode_iter_request(response_data) for item in response_data[iterkey]: yield item if not itervalue: break def rtm( # type: ignore self, url: Optional[str] = None, bot_id: Optional[str] = None ) -> Iterator[events.Event]: """ Iterate over event from the RTM API Args: url: Websocket connection url bot_id: Connecting bot ID Returns: :class:`slack.events.Event` or :class:`slack.events.Message` """ while True: bot_id = bot_id or self._find_bot_id() url = url or self._find_rtm_url() for event in self._incoming_from_rtm(url, bot_id): yield event url = None def _find_bot_id(self) -> str: # type: ignore auth = self.query(methods.AUTH_TEST) user_info = self.query(methods.USERS_INFO, {"user": auth["user_id"]}) bot_id = user_info["user"]["profile"]["bot_id"] LOG.info("BOT_ID is %s", bot_id) return bot_id def _find_rtm_url(self) -> str: # type: ignore response = self.query(methods.RTM_CONNECT) return response["url"] def _incoming_from_rtm( # type: ignore self, url: str, bot_id: str ) -> Iterator[events.Event]: for data in self._rtm(url): event = events.Event.from_rtm(json.loads(data)) if sansio.need_reconnect(event): break elif sansio.discard_event(event, bot_id): continue else: yield event PK!,A))slack/io/trio.pyfrom typing import Tuple, Union, Optional, AsyncIterator, MutableMapping import asks import trio from . import abc class SlackAPI(abc.SlackAPI): """ `asks curio` implementation of :class:`slack.io.abc.SlackAPI` Args: session: HTTP session """ def __init__(self, *, session: asks.Session, **kwargs) -> None: self._session = session super().__init__(**kwargs) async def _request( self, method: str, url: str, headers: Optional[MutableMapping], body: Optional[Union[str, MutableMapping]], ) -> Tuple[int, bytes, MutableMapping]: response = await self._session.request(method, url, headers=headers, data=body) return response.status_code, response.content, response.headers async def rtm(self, url=None, bot_id=None): raise NotImplementedError async def _rtm(self, url: str) -> AsyncIterator[str]: yield "" raise NotImplementedError async def sleep(self, seconds: float) -> None: await trio.sleep(seconds) PK!;((slack/methods.pyfrom enum import Enum ROOT_URL: str = "https://slack.com/api/" HOOK_URL: str = "https://hooks.slack.com" class Methods(Enum): """ Enumeration of available slack methods. Provides `iterkey` and `itermod` for :func:`SlackAPI.iter() `. """ # method = (ROOT_URL + method_url, iterkey, itermode, as_json) # api API_TEST = (ROOT_URL + "api.test", None, None, True) # apps.permissions APPS_PERMISSIONS_INFO = (ROOT_URL + "apps.permissions.info", None, None, False) APPS_PERMISSIONS_REQUEST = ( ROOT_URL + "apps.permissions.request", None, None, False, ) # auth AUTH_REVOKE = (ROOT_URL + "auth.revoke", None, None, False) AUTH_TEST = (ROOT_URL + "auth.test", None, None, True) # bots BOTS_INFO = (ROOT_URL + "bots.info", None, None, False) # channels CHANNELS_ARCHIVE = (ROOT_URL + "channels.archive", None, None, True) CHANNELS_CREATE = (ROOT_URL + "channels.create", None, None, True) CHANNELS_HISTORY = (ROOT_URL + "channels.history", "timeline", "messages", False) CHANNELS_INFO = (ROOT_URL + "channels.info", None, None, False) CHANNELS_INVITE = (ROOT_URL + "channels.invite", None, None, True) CHANNELS_JOIN = (ROOT_URL + "channels.join", None, None, True) CHANNELS_KICK = (ROOT_URL + "channels.kick", None, None, True) CHANNELS_LEAVE = (ROOT_URL + "channels.leave", None, None, True) CHANNELS_LIST = (ROOT_URL + "channels.list", "cursor", "channels", False) CHANNELS_MARK = (ROOT_URL + "channels.mark", None, None, True) CHANNELS_RENAME = (ROOT_URL + "channels.rename", None, None, True) CHANNELS_REPLIES = (ROOT_URL + "channels.replies", None, None, False) CHANNELS_SET_PURPOSE = (ROOT_URL + "channels.setPurpose", None, None, True) CHANNELS_SET_TOPIC = (ROOT_URL + "channels.setTopic", None, None, True) CHANNELS_UNARCHIVE = (ROOT_URL + "channels.unarchive", None, None, True) # chat CHAT_DELETE = (ROOT_URL + "chat.delete", None, None, True) CHAT_ME_MESSAGE = (ROOT_URL + "chat.meMessage", None, None, True) CHAT_POST_EPHEMERAL = (ROOT_URL + "chat.postEphemeral", None, None, True) CHAT_POST_MESSAGE = (ROOT_URL + "chat.postMessage", None, None, True) CHAT_UNFURL = (ROOT_URL + "chat.unfurl", None, None, True) CHAT_UPDATE = (ROOT_URL + "chat.update", None, None, True) # conversations CONVERSATIONS_ARCHIVE = (ROOT_URL + "conversations.archive", None, None, True) CONVERSATIONS_CLOSE = (ROOT_URL + "conversations.close", None, None, True) CONVERSATIONS_CREATE = (ROOT_URL + "conversations.create", None, None, True) CONVERSATIONS_HISTORY = ( ROOT_URL + "conversations.history", "cursor", "messages", False, ) CONVERSATIONS_INFO = (ROOT_URL + "conversations.info", None, None, False) CONVERSATIONS_INVITE = (ROOT_URL + "conversations.invite", None, None, True) CONVERSATIONS_JOIN = (ROOT_URL + "conversations.join", None, None, True) CONVERSATIONS_KICK = (ROOT_URL + "conversations.kick", None, None, True) CONVERSATIONS_LEAVE = (ROOT_URL + "conversations.leave", None, None, True) CONVERSATIONS_LIST = (ROOT_URL + "conversations.list", "cursor", "channels", False) CONVERSATIONS_MEMBERS = ( ROOT_URL + "conversations.members", "cursor", "members", False, ) CONVERSATIONS_OPEN = (ROOT_URL + "conversations.open", None, None, True) CONVERSATIONS_RENAME = (ROOT_URL + "conversations.rename", None, None, True) CONVERSATIONS_REPLIES = ( ROOT_URL + "conversations.replies", "cursor", "messages", False, ) CONVERSATIONS_SET_PURPOSE = ( ROOT_URL + "conversations.setPurpose", None, None, True, ) CONVERSATIONS_SET_TOPIC = (ROOT_URL + "conversations.setTopic", None, None, True) CONVERSATIONS_UNARCHIVE = (ROOT_URL + "conversations.unarchive", None, None, True) # dialog DIALOG_OPEN = (ROOT_URL + "dialog.open", None, None, True) # dnd DND_END_DND = (ROOT_URL + "dnd.endDnd", None, None, True) DND_END_SNOOZE = (ROOT_URL + "dnd.endSnooze", None, None, True) DND_INFO = (ROOT_URL + "dnd.info", None, None, False) DND_SET_SNOOZE = (ROOT_URL + "dnd.setSnooze", None, None, False) DND_TEAM_INFO = (ROOT_URL + "dnd.teamInfo", None, None, False) # emoji EMOJI_LIST = (ROOT_URL + "emoji.list", None, None, False) # files.comments FILES_COMMENTS_ADD = (ROOT_URL + "files.comments.add", None, None, True) FILES_COMMENTS_DELETE = (ROOT_URL + "files.comments.delete", None, None, True) FILES_COMMENTS_EDIT = (ROOT_URL + "files.comments.edit", None, None, True) # files FILES_DELETE = (ROOT_URL + "files.delete", None, None, True) FILES_INFO = (ROOT_URL + "files.info", None, None, False) FILES_LIST = (ROOT_URL + "files.list", "page", "files", False) FILES_REVOKE_PUBLIC_URL = (ROOT_URL + "files.revokePublicURL", None, None, True) FILES_SHARED_PUBLIC_URL = (ROOT_URL + "files.sharedPublicURL", None, None, True) FILES_UPLOAD = (ROOT_URL + "files.upload", None, None, False) # groups GROUPS_ARCHIVE = (ROOT_URL + "groups.archive", None, None, True) GROUPS_CLOSE = (ROOT_URL + "groups.close", None, None, False) GROUPS_CREATE = (ROOT_URL + "groups.create", None, None, True) GROUPS_CREATE_CHILD = (ROOT_URL + "groups.createChild", None, None, False) GROUPS_HISTORY = (ROOT_URL + "groups.history", "timeline", "messages", False) GROUPS_INFO = (ROOT_URL + "groups.info", None, None, False) GROUPS_INVITE = (ROOT_URL + "groups.invite", None, None, True) GROUPS_KICK = (ROOT_URL + "groups.kick", None, None, True) GROUPS_LEAVE = (ROOT_URL + "groups.leave", None, None, True) GROUPS_LIST = (ROOT_URL + "groups.list", None, None, False) GROUPS_MARK = (ROOT_URL + "groups.mark", None, None, True) GROUPS_OPEN = (ROOT_URL + "groups.open", None, None, True) GROUPS_RENAME = (ROOT_URL + "groups.rename", None, None, True) GROUPS_REPLIES = (ROOT_URL + "groups.replies", None, None, False) GROUPS_SET_PURPOSE = (ROOT_URL + "groups.setPurpose", None, None, True) GROUPS_SET_TOPIC = (ROOT_URL + "groups.setTopic", None, None, True) GROUPS_UNARCHIVE = (ROOT_URL + "groups.unarchive", None, None, True) # im IM_CLOSE = (ROOT_URL + "im.close", None, None, True) IM_HISTORY = (ROOT_URL + "im.history", "timeline", "messages", False) IM_LIST = (ROOT_URL + "im.list", None, None, False) IM_MARK = (ROOT_URL + "im.mark", None, None, True) IM_OPEN = (ROOT_URL + "im.open", None, None, True) IM_REPLIES = (ROOT_URL + "im.replies", None, None, False) # mpim MPIM_CLOSE = (ROOT_URL + "mpim.close", None, None, True) MPIM_HISTORY = (ROOT_URL + "mpim.history", "timeline", "messages", False) MPIM_LIST = (ROOT_URL + "mpim.list", None, None, False) MPIM_MARK = (ROOT_URL + "mpim.mark", None, None, True) MPIM_OPEN = (ROOT_URL + "mpim.open", None, None, True) MPIM_REPLIES = (ROOT_URL + "mpim.replies", None, None, False) # oauth OAUTH_ACCESS = (ROOT_URL + "oauth.access", None, None, False) OAUTH_TOKEN = (ROOT_URL + "oauth.token", None, None, False) # pins PINS_ADD = (ROOT_URL + "pins.add", None, None, True) PINS_LIST = (ROOT_URL + "pins.list", None, None, False) PINS_REMOVE = (ROOT_URL + "pins.remove", None, None, True) # reactions REACTIONS_ADD = (ROOT_URL + "reactions.add", None, None, True) REACTIONS_GET = (ROOT_URL + "reactions.get", None, None, False) REACTIONS_LIST = (ROOT_URL + "reactions.list", "page", "items", False) REACTIONS_REMOVE = (ROOT_URL + "reactions.remove", None, None, True) # reminders REMINDERS_ADD = (ROOT_URL + "reminders.add", None, None, True) REMINDERS_COMPLETE = (ROOT_URL + "reminders.complete", None, None, True) REMINDERS_DELETE = (ROOT_URL + "reminders.delete", None, None, True) REMINDERS_INFO = (ROOT_URL + "reminders.info", None, None, False) REMINDERS_LIsT = (ROOT_URL + "reminders.list", None, None, False) # rtm RTM_CONNECT = (ROOT_URL + "rtm.connect", None, None, False) RTM_START = (ROOT_URL + "rtm.start", None, None, False) # search SEARCH_ALL = (ROOT_URL + "search.all", "page", "messages", False) SEARCH_FILES = (ROOT_URL + "search.files", "page", "files", False) SEARCH_MESSAGES = (ROOT_URL + "search.messages", "page", "messages", False) # starts STARS_ADD = (ROOT_URL + "stars.add", None, None, True) STARS_LIST = (ROOT_URL + "stars.list", "page", "items", False) STARS_REMOVE = (ROOT_URL + "stars.remove", None, None, True) # team TEAM_ACCESS_LOGS = (ROOT_URL + "teams.accessLogs", None, None, False) TEAM_BILLABLE_INFO = (ROOT_URL + "teams.billableInfo", None, None, False) TEAM_INFO = (ROOT_URL + "teams.info", None, None, False) TEAM_INTEGRATION_LOGS = (ROOT_URL + "teams.integrationLogs", None, None, False) # team profile TEAM_PROFILE_GET = (ROOT_URL + "teams.profile.get", None, None, False) # usergroups USERGROUPS_CREATE = (ROOT_URL + "usergroups.create", None, None, True) USERGROUPS_DISABLE = (ROOT_URL + "usergroups.disable", None, None, True) USERGROUPS_ENABLE = (ROOT_URL + "usergroups.enable", None, None, True) USERGROUPS_LIST = (ROOT_URL + "usergroups.list", None, None, False) USERGROUPS_UPDATE = (ROOT_URL + "usergroups.update", None, None, True) # usergroups users USERGROUPS_USERS_LIST = (ROOT_URL + "usergroups.users.list", None, None, False) USERGROUPS_USERS_UPDATE = (ROOT_URL + "usergroups.users.update", None, None, True) # users USERS_DELETE_PHOTO = (ROOT_URL + "users.deletePhoto", None, None, False) USERS_GET_PRESENCE = (ROOT_URL + "users.getPresence", None, None, False) USERS_IDENTITY = (ROOT_URL + "users.identity", None, None, False) USERS_INFO = (ROOT_URL + "users.info", None, None, False) USERS_LIST = (ROOT_URL + "users.list", "cursor", "members", False) USERS_SET_ACTIVE = (ROOT_URL + "users.setActive", None, None, True) USERS_SET_PHOTO = (ROOT_URL + "users.setPhoto", None, None, False) USERS_SET_PRESENCE = (ROOT_URL + "users.setPresence", None, None, True) # users profile USERS_PROFILE_GET = (ROOT_URL + "users.profile.get", None, None, False) USERS_PROFILE_SET = (ROOT_URL + "users.profile.set", None, None, True) PK!slack/py.typedPK!X;+;+slack/sansio.py""" Collection of functions for sending and decoding request to or from the slack API """ import cgi import hmac import json import time import base64 import hashlib import logging from typing import Tuple, Union, Optional, MutableMapping from . import HOOK_URL, ROOT_URL, events, methods, exceptions LOG = logging.getLogger(__name__) RECONNECT_EVENTS = ("team_migration_started", "goodbye") """Events type preceding a disconnection""" SKIP_EVENTS = ("reconnect_url",) """Events that do not need to be dispatched""" ITERMODE = ("cursor", "page", "timeline") """Supported pagination mode""" def raise_for_status( status: int, headers: MutableMapping, data: MutableMapping ) -> None: """ Check request response status Args: status: Response status headers: Response headers data: Response data Raises: :class:`slack.exceptions.RateLimited`: For 429 status code :class:`slack.exceptions:HTTPException`: """ if status != 200: if status == 429: if isinstance(data, str): error = data else: error = data.get("error", "ratelimited") try: retry_after = int(headers.get("Retry-After", 1)) except ValueError: retry_after = 1 raise exceptions.RateLimited(retry_after, error, status, headers, data) else: raise exceptions.HTTPException(status, headers, data) def raise_for_api_error(headers: MutableMapping, data: MutableMapping) -> None: """ Check request response for Slack API error Args: headers: Response headers data: Response data Raises: :class:`slack.exceptions.SlackAPIError` """ if not data["ok"]: raise exceptions.SlackAPIError(data.get("error", "unknow_error"), headers, data) if "warning" in data: LOG.warning("Slack API WARNING: %s", data["warning"]) def decode_body(headers: MutableMapping, body: bytes) -> dict: """ Decode the response body For 'application/json' content-type load the body as a dictionary Args: headers: Response headers body: Response body Returns: decoded body """ type_, encoding = parse_content_type(headers) decoded_body = body.decode(encoding) # There is one api that just returns `ok` instead of json. In order to have a consistent API we decided to modify the returned payload into a dict. if type_ == "application/json": payload = json.loads(decoded_body) else: if decoded_body == "ok": payload = {"ok": True} else: payload = {"ok": False, "data": decoded_body} return payload def parse_content_type(headers: MutableMapping) -> Tuple[Optional[str], str]: """ Find content-type and encoding of the response Args: headers: Response headers Returns: :py:class:`tuple` (content-type, encoding) """ content_type = headers.get("content-type") if not content_type: return None, "utf-8" else: type_, parameters = cgi.parse_header(content_type) encoding = parameters.get("charset", "utf-8") return type_, encoding def prepare_request( url: Union[str, methods], data: Optional[MutableMapping], headers: Optional[MutableMapping], global_headers: MutableMapping, token: str, as_json: Optional[bool] = None, ) -> Tuple[str, Union[str, MutableMapping], MutableMapping]: """ Prepare outgoing request Create url, headers, add token to the body and if needed json encode it Args: url: :class:`slack.methods` item or string of url data: Outgoing data headers: Custom headers global_headers: Global headers token: Slack API token as_json: Post JSON to the slack API Returns: :py:class:`tuple` (url, body, headers) """ if isinstance(url, methods): as_json = as_json or url.value[3] real_url = url.value[0] else: real_url = url as_json = False if not headers: headers = {**global_headers} else: headers = {**global_headers, **headers} payload: Optional[Union[str, MutableMapping]] = None if real_url.startswith(HOOK_URL) or (real_url.startswith(ROOT_URL) and as_json): payload, headers = _prepare_json_request(data, token, headers) elif real_url.startswith(ROOT_URL) and not as_json: payload = _prepare_form_encoded_request(data, token) else: real_url = ROOT_URL + real_url payload = _prepare_form_encoded_request(data, token) return real_url, payload, headers def _prepare_json_request( data: Optional[MutableMapping], token: str, headers: MutableMapping ) -> Tuple[str, MutableMapping]: headers["Authorization"] = f"Bearer {token}" headers["Content-type"] = "application/json; charset=utf-8" if isinstance(data, events.Message): payload = data.to_json() else: payload = json.dumps(data or {}) return payload, headers def _prepare_form_encoded_request( data: Optional[MutableMapping], token: str ) -> MutableMapping: if isinstance(data, events.Message): data = data.serialize() if not data: data = {"token": token} elif "token" not in data: data["token"] = token return data def decode_response(status: int, headers: MutableMapping, body: bytes) -> dict: """ Decode incoming response Args: status: Response status headers: Response headers body: Response body Returns: Response data """ data = decode_body(headers, body) raise_for_status(status, headers, data) raise_for_api_error(headers, data) return data def find_iteration( url: Union[methods, str], itermode: Optional[str] = None, iterkey: Optional[str] = None, ) -> Tuple[str, str]: """ Find iteration mode and iteration key for a given :class:`slack.methods` Args: url: :class:`slack.methods` or string url itermode: Custom iteration mode iterkey: Custom iteration key Returns: :py:class:`tuple` (itermode, iterkey) """ if isinstance(url, methods): if not itermode: itermode = url.value[1] if not iterkey: iterkey = url.value[2] if not iterkey or not itermode: raise ValueError("Iteration not supported for: {}".format(url)) elif itermode not in ITERMODE: raise ValueError("Iteration not supported for: {}".format(itermode)) return itermode, iterkey def prepare_iter_request( url: Union[methods, str], data: MutableMapping, *, iterkey: Optional[str] = None, itermode: Optional[str] = None, limit: int = 200, itervalue: Optional[Union[str, int]] = None, ) -> Tuple[MutableMapping, str, str]: """ Prepare outgoing iteration request Args: url: :class:`slack.methods` item or string of url data: Outgoing data limit: Maximum number of results to return per call. iterkey: Key in response data to iterate over (required for url string). itermode: Iteration mode (required for url string) (one of `cursor`, `page` or `timeline`) itervalue: Value for current iteration (cursor hash, page or timestamp depending on the itermode) Returns: :py:class:`tuple` (data, iterkey, itermode) """ itermode, iterkey = find_iteration(url, itermode, iterkey) if itermode == "cursor": data["limit"] = limit if itervalue: data["cursor"] = itervalue elif itermode == "page": data["count"] = limit if itervalue: data["page"] = itervalue elif itermode == "timeline": data["count"] = limit if itervalue: data["latest"] = itervalue return data, iterkey, itermode def decode_iter_request(data: dict) -> Optional[Union[str, int]]: """ Decode incoming response from an iteration request Args: data: Response data Returns: Next itervalue """ if "response_metadata" in data: return data["response_metadata"].get("next_cursor") elif "paging" in data: current_page = int(data["paging"].get("page", 1)) max_page = int(data["paging"].get("pages", 1)) if current_page < max_page: return current_page + 1 elif "has_more" in data and data["has_more"] and "latest" in data: return data["messages"][-1]["ts"] return None def discard_event(event: events.Event, bot_id: str = None) -> bool: """ Check if the incoming event needs to be discarded Args: event: Incoming :class:`slack.events.Event` bot_id: Id of connected bot Returns: boolean """ if event["type"] in SKIP_EVENTS: return True elif bot_id and isinstance(event, events.Message): if event.get("bot_id") == bot_id: LOG.debug("Ignoring event: %s", event) return True elif "message" in event and event["message"].get("bot_id") == bot_id: LOG.debug("Ignoring event: %s", event) return True return False def need_reconnect(event: events.Event) -> bool: """ Check if RTM needs reconnecting Args: event: Incoming :class:`slack.events.Event` Returns: boolean """ if event["type"] in RECONNECT_EVENTS: return True else: return False def validate_request_signature( body: str, headers: MutableMapping, signing_secret: str ) -> None: """ Validate incoming request signature using the application signing secret. Contrary to the ``team_id`` and ``verification_token`` verification this method is not called by ``slack-sansio`` when creating object from incoming HTTP request. Because the body of the request needs to be provided as text and not decoded as json beforehand. Args: body: Raw request body headers: Request headers signing_secret: Application signing_secret Raise: :class:`slack.exceptions.InvalidSlackSignature`: when provided and calculated signature do not match :class:`slack.exceptions.InvalidTimestamp`: when incoming request timestamp is more than 5 minutes old """ request_timestamp = int(headers["X-Slack-Request-Timestamp"]) if (int(time.time()) - request_timestamp) > (60 * 5): raise exceptions.InvalidTimestamp(timestamp=request_timestamp) slack_signature = headers["X-Slack-Signature"] calculated_signature = ( "v0=" + hmac.new( signing_secret.encode("utf-8"), f"""v0:{headers["X-Slack-Request-Timestamp"]}:{body}""".encode("utf-8"), digestmod=hashlib.sha256, ).hexdigest() ) if not hmac.compare_digest(slack_signature, calculated_signature): raise exceptions.InvalidSlackSignature(slack_signature, calculated_signature) PK!slack/tests/__init__.pyPK! _??slack/tests/conftest.pyimport copy import json import time import functools from unittest.mock import Mock import pytest import requests import asynctest from slack.events import Event, EventRouter, MessageRouter from slack.io.abc import SlackAPI from slack.actions import Action from slack.actions import Router as ActionRouter from slack.commands import Router as CommandRouter from slack.commands import Command from . import data try: from slack.io.requests import SlackAPI as SlackAPIRequest except ImportError: SlackAPIRequest = None # type: ignore TOKEN = "abcdefg" class FakeIO(SlackAPI): async def _request(self, method, url, headers, body): pass async def sleep(self, seconds): time.sleep(seconds) async def _rtm(self, url): pass @pytest.fixture(params=(data.RTMEvents.__members__,)) def rtm_iterator(request): async def events(url): for key in request.param: yield data.RTMEvents[key].value return events @pytest.fixture(params=(data.RTMEvents.__members__,)) def rtm_iterator_non_async(request): def events(url): for key in request.param: yield data.RTMEvents[key].value return events @pytest.fixture(params=(FakeIO,)) def io_client(request): return request.param @pytest.fixture(params=({"token": TOKEN},)) def client(request, io_client): default_request = { "status": 200, "body": {"ok": True}, "headers": {"content-type": "application/json; charset=utf-8"}, } if "_request" not in request.param: request.param["_request"] = default_request elif isinstance(request.param["_request"], dict): request.param["_request"] = _default_response(request.param["_request"]) elif isinstance(request.param["_request"], list): for index, item in enumerate(request.param["_request"]): request.param["_request"][index] = _default_response(item) else: raise ValueError("Invalid `_request` parameters: %s", request.param["_request"]) if "token" not in request.param: request.param["token"] = TOKEN slackclient = io_client( **{k: v for k, v in request.param.items() if not k.startswith("_")} ) if isinstance(request.param["_request"], dict): return_value = ( request.param["_request"]["status"], json.dumps(request.param["_request"]["body"]).encode(), request.param["_request"]["headers"], ) if isinstance(slackclient, SlackAPIRequest): slackclient._request = Mock(return_value=return_value) else: slackclient._request = asynctest.CoroutineMock(return_value=return_value) else: responses = [ ( response["status"], json.dumps(response["body"]).encode(), response["headers"], ) for response in request.param["_request"] ] if isinstance(slackclient, SlackAPIRequest): slackclient._request = Mock(side_effect=responses) else: slackclient._request = asynctest.CoroutineMock(side_effect=responses) return slackclient def _default_response(response): default_response = { "status": 200, "body": {"ok": True}, "headers": {"content-type": "application/json; charset=utf-8"}, } response = {**default_response, **response} if "content-type" not in response["headers"]: response["headers"]["content-type"] = default_response["headers"][ "content-type" ] if isinstance(response["body"], str): response["body"] = copy.deepcopy(data.Methods[response["body"]].value) return response @pytest.fixture( params={**data.Events.__members__, **data.Messages.__members__} # type: ignore ) def event(request): if isinstance(request.param, str): try: payload = copy.deepcopy(data.Events[request.param].value) except KeyError: payload = copy.deepcopy(data.Messages[request.param].value) else: payload = copy.deepcopy(request.param) return payload @pytest.fixture(params={**data.Messages.__members__}) # type: ignore def message(request): if isinstance(request.param, str): payload = copy.deepcopy(data.Messages[request.param].value) else: payload = copy.deepcopy(request.param) return payload @pytest.fixture() def token(): return copy.copy(TOKEN) @pytest.fixture() def itercursor(): return "wxyz" @pytest.fixture() def event_router(): return EventRouter() @pytest.fixture() def message_router(): return MessageRouter() @pytest.fixture( params={ **data.InteractiveMessage.__members__, # type: ignore **data.DialogSubmission.__members__, # type: ignore **data.MessageAction.__members__, # type: ignore } ) def action(request): if isinstance(request.param, str): try: payload = copy.deepcopy(data.InteractiveMessage[request.param].value) except KeyError: try: payload = copy.deepcopy(data.DialogSubmission[request.param].value) except KeyError: payload = copy.deepcopy(data.MessageAction[request.param].value) else: payload = copy.deepcopy(request.param) return payload # @pytest.fixture(params={**data.InteractiveMessage.__members__}) # def interactive_message(request): # return Action.from_http(raw_action(request)) # @pytest.fixture(params={**data.DialogSubmission.__members__}) # def dialog_submission(request): # return Action.from_http(raw_action(request)) # @pytest.fixture(params={**data.MessageAction.__members__}) # def message_action(request): # return Action.from_http(raw_action(request)) @pytest.fixture() def action_router(): return ActionRouter() @pytest.fixture(params={**data.Commands.__members__}) def command(request): if isinstance(request.param, str): payload = copy.deepcopy(data.Commands[request.param].value) else: payload = copy.deepcopy(request.param) return payload @pytest.fixture() def command_router(): return CommandRouter() PK!Hslack/tests/data/__init__.pyfrom .events import Events, Messages, RTMEvents # noQa F401 from .actions import MessageAction, DialogSubmission, InteractiveMessage # noQa F401 from .methods import Methods # noQa F401 from .commands import Commands # noQa F401 PK!A)! ! slack/tests/data/actions.pyimport json from enum import Enum button_ok = { "type": "interactive_message", "actions": [{"name": "ok", "type": "button", "value": "hello"}], "callback_id": "test_action", "team": {"id": "T000AAA0A", "domain": "team"}, "channel": {"id": "C00000A00", "name": "general"}, "user": {"id": "U000AA000", "name": "username"}, "action_ts": "987654321.000001", "message_ts": "123456789.000001", "attachment_id": "1", "token": "supersecuretoken", "is_app_unfurl": False, "response_url": "https://hooks.slack.com/actions/T000AAA0A/123456789123/YTC81HsJRuuGSLVFbSnlkJlh", "trigger_id": "000000000.0000000000.e1bb750705a2f472e4476c4228cf4784", } button_cancel = { "type": "interactive_message", "actions": [{"name": "cancel", "type": "button", "value": "hello"}], "callback_id": "test_action", "team": {"id": "T000AAA0A", "domain": "team"}, "channel": {"id": "C00000A00", "name": "general"}, "user": {"id": "U000AA000", "name": "username"}, "action_ts": "987654321.000001", "message_ts": "123456789.000001", "attachment_id": "1", "token": "supersecuretoken", "is_app_unfurl": False, "response_url": "https://hooks.slack.com/actions/T000AAA0A/123456789123/YTC81HsJRuuGSLVFbSnlkJlh", "trigger_id": "000000000.0000000000.e1bb750705a2f472e4476c4228cf4784", } dialog_submission = { "type": "dialog_submission", "submission": {"foo": "bar"}, "callback_id": "test_action", "team": {"id": "T000AAA0A", "domain": "team"}, "user": {"id": "U000AA000", "name": "username"}, "channel": {"id": "C00000A00", "name": "general"}, "action_ts": "987654321.000001", "token": "supersecuretoken", "response_url": "https://hooks.slack.com/actions/T000AAA0A/123456789123/YTC81HsJRuuGSLVFbSnlkJlh", } message_action = { "type": "message_action", "token": "supersecuretoken", "action_ts": "987654321.000001", "team": {"id": "T000AAA0A", "domain": "team"}, "user": {"id": "U000AA000", "name": "username"}, "channel": {"id": "C00000A00", "name": "general"}, "callback_id": "test_action", "trigger_id": "418799722116.77329528181.9c7441638716b0b9b698f3d8ae73d9c1", "message_ts": "1534605601.000100", "message": { "type": "message", "user": "U000AA000", "text": "test message", "client_msg_id": "904f281d-338e-4621-a56f-afbfc80b3c59", "ts": "1534605601.000100", }, "response_url": "https://hooks.slack.com/actions/T000AAA0A/123456789123/YTC81HsJRuuGSLVFbSnlkJlh", } raw_button_ok = {"payload": json.dumps(button_ok)} raw_button_cancel = {"payload": json.dumps(button_cancel)} raw_dialog_submission = {"payload": json.dumps(dialog_submission)} raw_message_action = {"payload": json.dumps(message_action)} class InteractiveMessage(Enum): """ List of available interactive message action for testing - button_ok - button_cancel """ button_ok = raw_button_ok button_cancel = raw_button_cancel class DialogSubmission(Enum): """ List of available dialog submission action for testing - dialog_submission """ dialog_submission = raw_dialog_submission class MessageAction(Enum): """ List of available message action submission for testing - action """ action = raw_message_action PK!='..slack/tests/data/commands.pyfrom enum import Enum no_text = { "token": "supersecuretoken", "team_id": "T000AAA0A", "team_domain": "teamdomain", "channel_id": "C00000A00", "channel_name": "general", "user_id": "U000AA000", "user_name": "myuser", "command": "/test", "text": "", "response_url": "https://hooks.slack.com/actions/T000AAA0A/123456789123/YTC81HsJRuuGSLVFbSnlkJlh", "trigger_id": "000000000.0000000000.e1bb750705a2f472e4476c4228cf4784", } text = { "token": "supersecuretoken", "team_id": "T000AAA0A", "team_domain": "teamdomain", "channel_id": "C00000A00", "channel_name": "general", "user_id": "U000AA000", "user_name": "myuser", "command": "/test", "text": "foo bar", "response_url": "https://hooks.slack.com/actions/T000AAA0A/123456789123/YTC81HsJRuuGSLVFbSnlkJlh", "trigger_id": "000000000.0000000000.e1bb750705a2f472e4476c4228cf4784", } class Commands(Enum): """ List of available command for testing - text - no_text """ text = text no_text = no_text PK!;<3<3slack/tests/data/events.pyimport json from enum import Enum CHANNEL_DELETED = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "channel_deleted", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } PIN_ADDED = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "pin_added", "user": "U000AA000", "channel": "C00000A00", "item": { "type": "message", "channel": "C00000A00", "message": { "type": "message", "user": "U000AA000", "text": "hello world", "ts": "123456789.000001", "permalink": "https://team.slack.com/archives/C00000A00/p123456789000001", "pinned_to": ["C00000A00"], }, "created": 1513860592, "created_by": "U000AA000", }, "item_user": "U000AA000", "pin_count": 1, "event_ts": "1513860592.000014", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } REACTION_ADDED = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "reaction_added", "user": "U000AA000", "item": {"type": "message", "channel": "C00000A00", "ts": "123456789.000001"}, "reaction": "sirbot", "item_user": "U000AA000", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_SIMPLE = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "user": "U000AA000", "text": "hello world", "ts": "123456789.000001", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_MENTION = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "user": "U000AA000", "text": "<@U0AAA0A00> hello world", "ts": "123456789.000001", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_SNIPPET = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "user": "U000AA000", "text": "```\nhello world\n```", "ts": "123456789.000001", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_SHARED = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "user": "U000AA000", "text": "hello", "attachments": [ { "fallback": "[December 1st, 2000 1:01 PM] ovv: <@U000AA000> hello", "ts": "123456789.000001", "author_id": "U000AA000", "author_subname": "Ovv", "channel_id": "C00000A00", "channel_name": "general", "is_msg_unfurl": True, "text": "hello", "author_name": "Ovv", "author_link": "https://team.slack.com/team/U000AA000", "author_icon": "https://avatars.slack-edge.com/2000-01-01/111111111_11111111111_48.jpg", "mrkdwn_in": ["text"], "color": "D0D0D0", "from_url": "https://team.slack.com/archives/C00000A00/p123456789000001", "is_share": True, "footer": "Posted in #hello", } ], "ts": "123456789.000001", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_THREADED = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "user": "U000AA000", "text": "hello world", "thread_ts": "123456789.000001", "parent_user_id": "U000AA001", "ts": "987654321.000001", "channel": "C00000A00", "event_ts": "987654321.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_BOT = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "user": "U000AA000", "text": "hello world", "bot_id": "B0AAA0A00", "ts": "123456789.000001", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_ATTACHMENTS = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "user": "U000AA000", "text": "hello", "attachments": [ { "fallback": "Required plain-text summary of the attachment.", "text": "hello world", "id": 1, "color": "36a64f", } ], "ts": "123456789.000001", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_EDIT = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "message": { "type": "message", "user": "U000AA000", "text": "hello world", "edited": {"user": "U000AA000", "ts": "1513882449.000000"}, "ts": "123456789.000001", }, "subtype": "message_changed", "hidden": True, "channel": "C00000A00", "previous_message": { "type": "message", "user": "U000AA000", "text": "foo bar", "ts": "123456789.000001", }, "event_ts": "123456789.000002", "ts": "123456789.000002", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_BOT_EDIT = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "message": { "type": "message", "user": "U000AA000", "text": "hello world", "bot_id": "B0AAA0A00", "edited": {"user": "U000AA000", "ts": "1513882449.000000"}, "ts": "123456789.000001", }, "subtype": "message_changed", "hidden": True, "channel": "C00000A00", "previous_message": { "type": "message", "user": "U000AA000", "text": "foo bar", "ts": "123456789.000001", "bot_id": "B0AAA0A00", }, "event_ts": "123456789.000002", "ts": "123456789.000002", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_EDIT_THREADED = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "type": "message", "message": { "type": "message", "user": "U000AA000", "text": "hello", "edited": {"user": "U000AA000", "ts": "1513882759.000000"}, "thread_ts": "123456789.000001", "parent_user_id": "U000AA000", "ts": "1513882746.000279", }, "subtype": "message_changed", "hidden": True, "channel": "C00000A00", "previous_message": { "type": "message", "user": "U000AA000", "text": "foo bar", "thread_ts": "123456789.000001", "parent_user_id": "U000AA000", "ts": "123456789.000001", }, "event_ts": "123456789.000002", "ts": "123456789.000002", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_NONE_TEXT = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "bot_id": "B092ZCBCY", "attachments": [ { "fallback": """: JupyterCon 2018: Call For Proposals: """ """. """ """""" "", "ts": 1516989711, "author_name": "Python Software", "author_link": "https://twitter.com/ThePSF/status/956950323523932160", "author_icon": "https://pbs.twimg.com/profile_images/439154912719413248/pUBY5pVj_normal.png", "author_subname": "@ThePSF", "pretext": "", "text": """JupyterCon 2018: Call For Proposals: """ """. """ """""", "service_name": "twitter", "service_url": "https://twitter.com/", "from_url": "https://twitter.com/ThePSF/status/956950323523932160", "id": 1, "footer": "Twitter", "footer_icon": "https://a.slack-edge.com/6e067/img/services/twitter_pixel_snapped_32.png", } ], "text": None, "type": "message", "subtype": "bot_message", "ts": "123456789.000001", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } MESSAGE_CHANNEL_TOPIC = { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": { "user": "U299GM524", "topic": "Company-wide announcements and work-based matter hello", "text": "<@U299GM524> set the channel topic: Company-wide announcements and work-based matter hello", "type": "message", "subtype": "channel_topic", "ts": "123456789.000001", "channel": "C00000A00", "event_ts": "123456789.000001", }, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } GOODBYE = {"type": "goodbye"} RECONNECT_URL = { "type": "reconnect_url", "url": "wss://testteam.slack.com/012345678910", } class Events(Enum): """ List of available event for testing - channel_deleted - pin_added - reaction_added """ channel_deleted = CHANNEL_DELETED pin_added = PIN_ADDED reaction_added = REACTION_ADDED non_text = MESSAGE_NONE_TEXT class RTMEvents(Enum): """ List of available rtm event for testing - channel_deleted - pin_added - goodbye - message_bot - reconnect_url """ channel_deleted = json.dumps(CHANNEL_DELETED["event"]) pin_added = json.dumps(PIN_ADDED["event"]) goodbye = json.dumps(GOODBYE) message_bot = json.dumps(MESSAGE_BOT["event"]) reconnect_url = json.dumps(RECONNECT_URL) class Messages(Enum): """ List of available message for testing - simple - snippet - shared - threaded - bot - bot_edit - attachment - edit - edit_threaded - mention - none_text - channel_topic """ simple = MESSAGE_SIMPLE snippet = MESSAGE_SNIPPET shared = MESSAGE_SHARED threaded = MESSAGE_THREADED bot = MESSAGE_BOT bot_edit = MESSAGE_BOT_EDIT attachment = MESSAGE_ATTACHMENTS edit = MESSAGE_EDIT edit_threaded = MESSAGE_EDIT_THREADED mention = MESSAGE_MENTION channel_topic = MESSAGE_CHANNEL_TOPIC PK!ppslack/tests/data/methods.pyfrom enum import Enum CHANNELS = { "ok": True, "channels": [ { "id": "C00000001", "name": "fun", "created": 1360782804, "creator": "U000AA000", "is_archived": False, "is_member": False, "num_members": 6, "topic": { "value": "Fun times", "creator": "U000AA000", "last_set": 1369677212, }, "purpose": { "value": "This channel is for fun", "creator": "U000AA000", "last_set": 1360782804, }, }, { "id": "C00000002", "name": "fun", "created": 1360782804, "creator": "U000AA000", "is_archived": False, "is_member": False, "num_members": 6, "topic": { "value": "Fun times", "creator": "U000AA000", "last_set": 1369677212, }, "purpose": { "value": "This channel is for fun", "creator": "U000AA000", "last_set": 1360782804, }, }, ], } CHANNELS_ITER = { "ok": True, "channels": CHANNELS["channels"], "response_metadata": {"next_cursor": "wxyz"}, } USERS_INFO = { "ok": True, "user": { "id": "W012A3CDE", "team_id": "T012AB3C4", "name": "sirbotalotr", "deleted": True, "color": "9f69e7", "real_name": "episod", "tz": "America/Los_Angeles", "tz_label": "Pacific Daylight Time", "tz_offset": -25200, "profile": { "avatar_hash": "ge3b51ca72de", "status_text": "Print is dead", "status_emoji": ":books:", "real_name": "Egon Spengler", "display_name": "spengler", "real_name_normalized": "Egon Spengler", "display_name_normalized": "spengler", "email": "spengler@ghostbusters.example.com", "image_24": "https://.../avatar/e3b51ca72dee4ef87916ae2b9240df50.jpg", "image_32": "https://.../avatar/e3b51ca72dee4ef87916ae2b9240df50.jpg", "image_48": "https://.../avatar/e3b51ca72dee4ef87916ae2b9240df50.jpg", "image_72": "https://.../avatar/e3b51ca72dee4ef87916ae2b9240df50.jpg", "image_192": "https://.../avatar/e3b51ca72dee4ef87916ae2b9240df50.jpg", "image_512": "https://.../avatar/e3b51ca72dee4ef87916ae2b9240df50.jpg", "team": "T012AB3C4", "bot_id": "B0AAA0A00", }, "is_admin": True, "is_owner": False, "is_primary_owner": False, "is_restricted": False, "is_ultra_restricted": False, "is_bot": True, "updated": 1502138686, "is_app_user": False, "has_2fa": False, }, } AUTH_TEST = { "ok": True, "url": "https://testteam.slack.com/", "team": "TestTeam Workspace", "user": "sirbotalot", "team_id": "T12345678", "user_id": "W12345678", } RTM_CONNECT = { "ok": True, "self": {"id": "W012A3CDE", "name": "sirbotalot"}, "team": {"domain": "testteam", "id": "T12345678", "name": "testteam"}, "url": "wss://testteam.slack.com/012345678910", } class Methods(Enum): """ List of available methods for testing - channels - channels_iter (channels with a cursor) - users_info - auth_test - rtm_connect """ channels_iter = CHANNELS_ITER channels = CHANNELS users_info = USERS_INFO auth_test = AUTH_TEST rtm_connect = RTM_CONNECT PK!ed&+slack/tests/plugin.pyimport pytest from . import data from .conftest import event as slack_event from .conftest import action as slack_action from .conftest import client as slack_client from .conftest import command as slack_command from .conftest import message as slack_message PK!uslack/tests/test.sh#!/bin/sh set -e flake8 . black --check --diff . isort --recursive --check-only . mypy slack/ pytest test --verbose --cov sphinx-build docs/ docs/_build -W python setup.py sdist python setup.py bdist_wheel PK!]{iNNslack/tests/test_actions.pyimport pytest import slack from slack.actions import Action from . import data class TestActions: def test_from_http(self, action): act = Action.from_http(action) assert isinstance(act, slack.actions.Action) def test_parsing_token(self, action): slack.actions.Action.from_http(action, verification_token="supersecuretoken") def test_parsing_team_id(self, action): slack.actions.Action.from_http(action, team_id="T000AAA0A") def test_parsing_wrong_token(self, action): with pytest.raises(slack.exceptions.FailedVerification): slack.actions.Action.from_http(action, verification_token="xxx") def test_parsing_wrong_team_id(self, action): with pytest.raises(slack.exceptions.FailedVerification): slack.actions.Action.from_http(action, team_id="xxx") def test_mapping_access(self, action): act = Action.from_http(action) assert act["callback_id"] == "test_action" def test_mapping_delete(self, action): act = Action.from_http(action) assert act["callback_id"] == "test_action" del act["callback_id"] with pytest.raises(KeyError): print(act["callback_id"]) def test_mapping_set(self, action): act = Action.from_http(action) assert act["callback_id"] == "test_action" act["callback_id"] = "foo" assert act["callback_id"] == "foo" class TestActionRouter: def test_register(self, action_router): def handler(): pass action_router.register("test_action", handler) assert len(action_router._routes["test_action"]["*"]) == 1 assert action_router._routes["test_action"]["*"][0] is handler def test_register_name(self, action_router): def handler(): pass action_router.register("test_action", handler, name="aaa") assert len(action_router._routes["test_action"]["aaa"]) == 1 assert action_router._routes["test_action"]["aaa"][0] is handler def test_multiple_register(self, action_router): def handler(): pass def handler_bis(): pass action_router.register("test_action", handler) action_router.register("test_action", handler_bis) assert len(action_router._routes["test_action"]["*"]) == 2 assert action_router._routes["test_action"]["*"][0] is handler assert action_router._routes["test_action"]["*"][1] is handler_bis def test_dispath(self, action, action_router): def handler(): pass act = Action.from_http(action) action_router.register("test_action", handler) handlers = list() for h in action_router.dispatch(act): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler def test_no_dispatch(self, action, action_router): def handler(): pass act = Action.from_http(action) action_router.register("xxx", handler) for h in action_router.dispatch(act): assert False @pytest.fixture(params={**data.InteractiveMessage.__members__}) def test_dispatch_details(self, action, action_router): def handler(): pass act = Action.from_http(action) action_router.register("test_action", handler, name="ok") action_router.register("test_action", handler, name="cancel") handlers = list() for h in action_router.dispatch(act): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler def test_multiple_dispatch(self, action, action_router): def handler(): pass def handler_bis(): pass act = Action.from_http(action) action_router.register("test_action", handler) action_router.register("test_action", handler_bis) handlers = list() for h in action_router.dispatch(act): handlers.append(h) assert len(handlers) == 2 assert handlers[0] is handler assert handlers[1] is handler_bis def test_dispatch_unhandle_type(self, action_router): action = {"type": "unhandled_type", "callback_id": "test_action"} with pytest.raises(slack.actions.UnknownActionType): for _ in action_router.dispatch(action): pass PK!2 slack/tests/test_commands.pyimport pytest import slack from slack.commands import Command class TestCommand: def test_fixture(self, command): com = Command(command) assert isinstance(com, slack.commands.Command) def test_parsing_token(self, command): slack.commands.Command(command, verification_token="supersecuretoken") def test_parsing_team_id(self, command): slack.commands.Command(command, team_id="T000AAA0A") def test_parsing_wrong_token(self, command): with pytest.raises(slack.exceptions.FailedVerification): slack.commands.Command(command, verification_token="xxx") def test_parsing_wrong_team_id(self, command): with pytest.raises(slack.exceptions.FailedVerification): slack.commands.Command(command, team_id="xxx") def test_mapping_access(self, command): com = Command(command) assert com["user_id"] == "U000AA000" def test_mapping_delete(self, command): com = Command(command) assert com["user_id"] == "U000AA000" del com["user_id"] with pytest.raises(KeyError): print(com["user_id"]) def test_mapping_set(self, command): com = Command(command) assert com["user_id"] == "U000AA000" com["user_id"] = "foo" assert com["user_id"] == "foo" class TestCommandRouter: def test_register(self, command_router): def handler(): pass command_router.register("/test", handler) assert len(command_router._routes["/test"]) == 1 assert command_router._routes["/test"][0] is handler def test_register_no_slash(self, command_router): def handler(): pass command_router.register("test", handler) assert len(command_router._routes["/test"]) == 1 assert command_router._routes["/test"][0] is handler def test_multiple_register(self, command_router): def handler(): pass def handler_bis(): pass command_router.register("/test", handler) command_router.register("/test", handler_bis) assert len(command_router._routes["/test"]) == 2 assert command_router._routes["/test"][0] is handler assert command_router._routes["/test"][1] is handler_bis def test_dispath(self, command, command_router): def handler(): pass handlers = list() command_router.register("/test", handler) for h in command_router.dispatch(command): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler def test_no_dispatch(self, command, command_router): def handler(): pass command_router.register("/xxx", handler) for h in command_router.dispatch(command): assert False def test_multiple_dispatch(self, command, command_router): def handler(): pass def handler_bis(): pass handlers = list() command_router.register("/test", handler) command_router.register("/test", handler_bis) for h in command_router.dispatch(command): handlers.append(h) assert len(handlers) == 2 assert handlers[0] is handler assert handlers[1] is handler_bis PK!'o55slack/tests/test_events.pyimport re import pytest import slack from slack.events import Event from . import data class TestEvents: def test_clone_event(self, event): ev = Event.from_http(event) clone = ev.clone() assert clone == ev def test_modify_clone(self, event): ev = Event.from_http(event) clone = ev.clone() clone["text"] = "aaaaa" assert clone != ev def test_parsing(self, event): http_event = slack.events.Event.from_http(event) rtm_event = slack.events.Event.from_rtm(event["event"]) assert isinstance(http_event, slack.events.Event) assert isinstance(rtm_event, slack.events.Event) assert http_event.event == rtm_event.event == event["event"] assert rtm_event.metadata is None assert http_event.metadata == { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": http_event.event, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } def test_parsing_token(self, event): slack.events.Event.from_http(event, verification_token="supersecuretoken") def test_parsing_team_id(self, event): slack.events.Event.from_http(event, team_id="T000AAA0A") def test_parsing_wrong_token(self, event): with pytest.raises(slack.exceptions.FailedVerification): slack.events.Event.from_http(event, verification_token="xxx") def test_parsing_wrong_team_id(self, event): with pytest.raises(slack.exceptions.FailedVerification): slack.events.Event.from_http(event, team_id="xxx") @pytest.mark.parametrize( "event", [ "pin_added", "reaction_added", "simple", "snippet", "shared", "threaded", "bot", "attachment", ], indirect=True, ) def test_mapping_access(self, event): ev = Event.from_http(event) assert ev["user"] == "U000AA000" @pytest.mark.parametrize( "event", [ "pin_added", "reaction_added", "simple", "snippet", "shared", "threaded", "bot", "attachment", ], indirect=True, ) def test_mapping_delete(self, event): ev = Event.from_http(event) assert ev["user"] == "U000AA000" del ev["user"] with pytest.raises(KeyError): print(ev["user"]) @pytest.mark.parametrize( "event", [ "pin_added", "reaction_added", "simple", "snippet", "shared", "threaded", "bot", "attachment", ], indirect=True, ) def test_mapping_set(self, event): ev = Event.from_http(event) assert ev["user"] == "U000AA000" ev["user"] = "foo" assert ev["user"] == "foo" class TestMessage: @pytest.mark.parametrize("event", {**data.Messages.__members__}, indirect=True) def test_parsing(self, event): http_event = slack.events.Event.from_http(event) rtm_event = slack.events.Event.from_rtm(event["event"]) assert isinstance(http_event, slack.events.Message) assert isinstance(rtm_event, slack.events.Message) assert http_event.event == rtm_event.event assert rtm_event.metadata is None assert http_event.metadata == { "token": "supersecuretoken", "team_id": "T000AAA0A", "api_app_id": "A0AAAAAAA", "event": http_event.event, "type": "event_callback", "authed_teams": ["T000AAA0A"], "event_id": "AAAAAAA", "event_time": 123456789, } def test_serialize(self): msg = slack.events.Message() msg["channel"] = "C00000A00" msg["text"] = "Hello world" assert msg.serialize() == {"channel": "C00000A00", "text": "Hello world"} def test_serialize_attachments(self): msg = slack.events.Message() msg["channel"] = "C00000A00" msg["text"] = "Hello world" msg["attachments"] = {"hello": "world"} assert msg.serialize() == { "channel": "C00000A00", "text": "Hello world", "attachments": '{"hello": "world"}', } def test_response(self, message): msg = Event.from_http(message) rep = msg.response() assert isinstance(rep, slack.events.Message) assert rep["channel"] == "C00000A00" def test_response_not_in_thread(self, message): msg = Event.from_http(message) rep = msg.response(in_thread=False) assert rep == {"channel": "C00000A00"} def test_response_in_thread(self, message): msg = Event.from_http(message) rep = msg.response(in_thread=True) assert rep == {"channel": "C00000A00", "thread_ts": "123456789.000001"} def test_response_thread_default(self, message): msg = Event.from_http(message) rep = msg.response() if "thread_ts" in msg or "thread_ts" in msg.get("message", {}): assert rep == {"channel": "C00000A00", "thread_ts": "123456789.000001"} else: assert rep == {"channel": "C00000A00"} class TestEventRouter: def test_register(self, event_router): def handler(): pass event_router.register("channel_deleted", handler) assert len(event_router._routes["channel_deleted"]["*"]["*"]) == 1 assert event_router._routes["channel_deleted"]["*"]["*"][0] is handler def test_register_details(self, event_router): def handler(): pass event_router.register("channel_deleted", handler, hello="world") assert len(event_router._routes["channel_deleted"]["hello"]["world"]) == 1 assert event_router._routes["channel_deleted"]["hello"]["world"][0] is handler def test_register_two_details(self, event_router): def handler(): pass with pytest.raises(ValueError): event_router.register("channel_deleted", handler, hello="world", foo="bar") def test_multiple_register(self, event_router): def handler(): pass def handler_bis(): pass event_router.register("channel_deleted", handler) event_router.register("channel_deleted", handler_bis) assert len(event_router._routes["channel_deleted"]["*"]["*"]) == 2 assert event_router._routes["channel_deleted"]["*"]["*"][0] is handler assert event_router._routes["channel_deleted"]["*"]["*"][1] is handler_bis @pytest.mark.parametrize("event", {**data.Events.__members__}, indirect=True) def test_dispatch(self, event, event_router): def handler(): pass ev = Event.from_http(event) handlers = list() event_router.register("channel_deleted", handler) event_router.register("pin_added", handler) event_router.register("reaction_added", handler) event_router.register("message", handler) for h in event_router.dispatch(ev): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler @pytest.mark.parametrize("event", {**data.Events.__members__}, indirect=True) def test_no_dispatch(self, event, event_router): def handler(): pass ev = Event.from_http(event) event_router.register("xxx", handler) for h in event_router.dispatch(ev): assert False @pytest.mark.parametrize("event", {**data.Events.__members__}, indirect=True) def test_dispatch_details(self, event, event_router): def handler(): pass ev = Event.from_http(event) handlers = list() event_router.register("channel_deleted", handler, channel="C00000A00") event_router.register("pin_added", handler, channel="C00000A00") event_router.register("reaction_added", handler, reaction="sirbot") event_router.register("message", handler, text=None) for h in event_router.dispatch(ev): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler @pytest.mark.parametrize("event", {**data.Events.__members__}, indirect=True) def test_multiple_dispatch(self, event, event_router): def handler(): pass def handler_bis(): pass ev = Event.from_http(event) handlers = list() event_router.register("channel_deleted", handler) event_router.register("pin_added", handler) event_router.register("reaction_added", handler) event_router.register("channel_deleted", handler_bis) event_router.register("pin_added", handler_bis) event_router.register("reaction_added", handler_bis) event_router.register("message", handler) event_router.register("message", handler_bis) for h in event_router.dispatch(ev): handlers.append(h) assert len(handlers) == 2 assert handlers[0] is handler assert handlers[1] is handler_bis class TestMessageRouter: def test_register(self, message_router): def handler(): pass message_router.register(".*", handler) assert len(message_router._routes["*"][None][re.compile(".*")]) == 1 assert message_router._routes["*"][None][re.compile(".*")][0] is handler def test_register_channel(self, message_router): def handler(): pass message_router.register(".*", handler, channel="C00000A00") assert len(message_router._routes["C00000A00"][None][re.compile(".*")]) == 1 assert message_router._routes["C00000A00"][None][re.compile(".*")][0] is handler def test_register_subtype(self, message_router): def handler(): pass message_router.register(".*", handler, subtype="bot_message") assert len(message_router._routes["*"]["bot_message"][re.compile(".*")]) == 1 assert ( message_router._routes["*"]["bot_message"][re.compile(".*")][0] is handler ) def test_multiple_register(self, message_router): def handler(): pass def handler_bis(): pass message_router.register(".*", handler) message_router.register(".*", handler_bis) assert len(message_router._routes["*"][None][re.compile(".*")]) == 2 assert message_router._routes["*"][None][re.compile(".*")][0] is handler assert message_router._routes["*"][None][re.compile(".*")][1] is handler_bis def test_dispatch(self, message_router, message): def handler(): pass msg = Event.from_http(message) message_router.register(".*", handler) handlers = list() for h in message_router.dispatch(msg): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler def test_no_dispatch(self, message_router, message): def handler(): pass msg = Event.from_http(message) message_router.register("xxx", handler) for h in message_router.dispatch(msg): assert False def test_dispatch_pattern(self, message_router, message): def handler(): pass msg = Event.from_http(message) message_router.register("hello", handler) handlers = list() for h in message_router.dispatch(msg): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler def test_multiple_dispatch(self, message_router, message): def handler(): pass def handler_bis(): pass msg = Event.from_http(message) message_router.register(".*", handler) message_router.register(".*", handler_bis) handlers = list() for h in message_router.dispatch(msg): handlers.append(h) assert len(handlers) == 2 assert handlers[0] is handler assert handlers[1] is handler_bis def test_multiple_dispatch_pattern(self, message_router, message): def handler(): pass def handler_bis(): pass msg = Event.from_http(message) message_router.register("hello", handler) message_router.register("hello", handler_bis) handlers = list() for h in message_router.dispatch(msg): handlers.append(h) assert len(handlers) == 2 assert handlers[0] is handler assert handlers[1] is handler_bis def test_dispatch_channel(self, message_router, message): def handler(): pass msg = Event.from_http(message) message_router.register("hello", handler, channel="C00000A00") handlers = list() for h in message_router.dispatch(msg): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler @pytest.mark.parametrize("message", ("channel_topic",), indirect=True) def test_dispatch_subtype(self, message_router, message): def handler(): pass msg = Event.from_http(message) message_router.register(".*", handler, subtype="channel_topic") handlers = list() for h in message_router.dispatch(msg): handlers.append(h) assert len(handlers) == 1 assert handlers[0] is handler PK!@@slack/tests/test_io.pyimport json import time import asyncio import datetime import asks import trio import curio import pytest import aiohttp import requests import asynctest import slack from slack import methods, exceptions from slack.io.trio import SlackAPI as SlackAPITrio from slack.io.curio import SlackAPI as SlackAPICurio from slack.io.aiohttp import SlackAPI as SlackAPIAiohttp from slack.io.requests import SlackAPI as SlackAPIRequest @pytest.mark.asyncio class TestABC: async def test_query(self, client, token): rep = await client.query(methods.AUTH_TEST) client._request.assert_called_once() assert client._request.call_args[0][0] == "POST" assert client._request.call_args[0][1] == "https://slack.com/api/auth.test" assert client._request.call_args[0][2] == { "Content-type": "application/json; charset=utf-8", "Authorization": f"Bearer {token}", } assert client._request.call_args[0][3] == "{}" assert rep == {"ok": True} async def test_query_url(self, client): await client.query("auth.test") assert client._request.call_args[0][1] == "https://slack.com/api/auth.test" async def test_query_long_url(self, client): await client.query("https://slack.com/api/auth.test") assert client._request.call_args[0][1] == "https://slack.com/api/auth.test" async def test_query_webhook_url(self, client): await client.query("https://hooks.slack.com/abcdef") assert client._request.call_args[0][1] == "https://hooks.slack.com/abcdef" async def test_query_data(self, client, token): data = {"hello": "world"} called_with = json.dumps(data.copy()) await client.query(methods.AUTH_TEST, data) assert client._request.call_args[0][3] == called_with async def test_query_data_webhook(self, client, token): data = {"hello": "world"} called_with = data.copy() await client.query("https://hooks.slack.com/abcdef", data) assert client._request.call_args[0][3] == json.dumps(called_with) async def test_query_headers(self, client, token): custom_headers = { "hello": "world", "Content-type": "application/json; charset=utf-8", "Authorization": f"Bearer {token}", } called_headers = custom_headers.copy() await client.query("https://hooks.slack.com/abcdef", headers=custom_headers) assert client._request.call_args[0][2] == called_headers @pytest.mark.parametrize( "client", ({"_request": [{"body": "channels_iter"}, {"body": "channels"}]},), indirect=True, ) async def test_iter(self, client, token, itercursor): channels = 0 async for _ in client.iter(methods.CHANNELS_LIST): # noQa: F841 channels += 1 assert channels == 4 assert client._request.call_count == 2 client._request.assert_called_with( "POST", "https://slack.com/api/channels.list", {}, {"limit": 200, "token": token, "cursor": itercursor}, ) @pytest.mark.parametrize( "client", ({"_request": [{"body": "channels_iter"}, {"body": "channels"}]},), indirect=True, ) async def test_iter_itermode_iterkey(self, client, token, itercursor): channels = 0 async for _ in client.iter( "channels.list", itermode="cursor", iterkey="channels" ): # noQa: F841 channels += 1 assert channels == 4 assert client._request.call_count == 2 client._request.assert_called_with( "POST", "https://slack.com/api/channels.list", {}, {"limit": 200, "token": token, "cursor": itercursor}, ) async def test_iter_not_supported(self, client): with pytest.raises(ValueError): async for _ in client.iter(methods.AUTH_TEST): # noQa: F841 pass with pytest.raises(ValueError): async for _ in client.iter("channels.list"): # noQa: F841 pass with pytest.raises(ValueError): async for _ in client.iter( "https://slack.com/api/channels.list" ): # noQa: F841 pass @pytest.mark.parametrize( "client", ({"_request": [{"body": "channels_iter"}, {"body": "channels"}]},), indirect=True, ) async def test_iter_wait(self, client): client.sleep = asynctest.CoroutineMock() channels = 0 async for _ in client.iter(methods.CHANNELS_LIST, minimum_time=2): # noQa: F841 channels += 1 assert channels == 4 assert client._request.call_count == 2 assert client.sleep.call_count == 1 assert 2 > client.sleep.call_args[0][0] > 1.9 @pytest.mark.parametrize( "client", ({"_request": [{"body": "channels_iter"}, {"body": "channels"}]},), indirect=True, ) async def test_iter_no_wait(self, client): client.sleep = asynctest.CoroutineMock() channels = 0 async for _ in client.iter(methods.CHANNELS_LIST, minimum_time=1): # noQa: F841 channels += 1 await asyncio.sleep(0.5) assert channels == 4 assert client._request.call_count == 2 assert client.sleep.call_count == 0 @pytest.mark.parametrize( "client", ({"_request": [{"body": "auth_test"}, {"body": "users_info"}]},), indirect=True, ) async def test_find_bot_id(self, client): bot_id = await client._find_bot_id() assert bot_id == "B0AAA0A00" @pytest.mark.parametrize( "client", ({"_request": {"body": "rtm_connect"}},), indirect=True ) async def test_find_rtm_url(self, client): url = await client._find_rtm_url() assert url == "wss://testteam.slack.com/012345678910" async def test_incoming_rtm(self, client, rtm_iterator): client._rtm = rtm_iterator events = [] async for event in client._incoming_from_rtm( "wss://testteam.slack.com/012345678910", "B0AAA0A00" ): assert isinstance(event, slack.events.Event) events.append(event) assert len(events) > 0 @pytest.mark.parametrize("rtm_iterator", (("goodbye",),), indirect=True) async def test_incoming_rtm_reconnect(self, client, rtm_iterator): client._rtm = rtm_iterator events = [] async for event in client._incoming_from_rtm( "wss://testteam.slack.com/012345678910", "B0AAA0A00" ): events.append(event) assert len(events) == 0 @pytest.mark.parametrize("rtm_iterator", (("message_bot",),), indirect=True) async def test_incoming_rtm_discard_bot_id(self, client, rtm_iterator): client._rtm = rtm_iterator events = [] async for event in client._incoming_from_rtm( "wss://testteam.slack.com/012345678910", "B0AAA0A00" ): events.append(event) assert len(events) == 0 @pytest.mark.parametrize("rtm_iterator", (("reconnect_url",),), indirect=True) async def test_incoming_rtm_skip(self, client, rtm_iterator): client._rtm = rtm_iterator events = [] async for event in client._incoming_from_rtm( "wss://testteam.slack.com/012345678910", "B0AAA0A00" ): events.append(event) assert len(events) == 0 @pytest.mark.parametrize("io_client", (SlackAPIRequest,), indirect=True) class TestNoAsync: def test_query(self, client, token): rep = client.query(methods.AUTH_TEST) client._request.assert_called_once() assert client._request.call_args[0][0] == "POST" assert client._request.call_args[0][1] == "https://slack.com/api/auth.test" assert client._request.call_args[0][2] == { "Content-type": "application/json; charset=utf-8", "Authorization": f"Bearer {token}", } assert client._request.call_args[0][3] == "{}" assert rep == {"ok": True} def test_query_data(self, client): data = {"hello": "world"} called_with = json.dumps(data.copy()) client.query(methods.AUTH_TEST, data) assert client._request.call_args[0][3] == called_with def test_query_headers(self, client, token): custom_headers = { "hello": "world", "Content-type": "application/json; charset=utf-8", "Authorization": f"Bearer {token}", } called_headers = custom_headers.copy() client.query("https://hooks.slack.com/abcdef", headers=custom_headers) assert client._request.call_args[0][2] == called_headers @pytest.mark.parametrize( "client", ({"_request": [{"body": "channels_iter"}, {"body": "channels"}]},), indirect=True, ) def test_iter(self, client, token, itercursor): channels = 0 for _ in client.iter(methods.CHANNELS_LIST): # noQa: F841 channels += 1 assert channels == 4 assert client._request.call_count == 2 client._request.assert_called_with( "POST", "https://slack.com/api/channels.list", {}, {"limit": 200, "token": token, "cursor": itercursor}, ) @pytest.mark.parametrize( "client", ({"_request": [{"body": "channels_iter"}, {"body": "channels"}]},), indirect=True, ) def test_iter_wait(self, client): client.sleep = asynctest.CoroutineMock() channels = 0 for _ in client.iter(methods.CHANNELS_LIST, minimum_time=2): # noQa: F841 channels += 1 assert channels == 4 assert client._request.call_count == 2 assert client.sleep.call_count == 1 assert 2 > client.sleep.call_args[0][0] > 1.9 @pytest.mark.parametrize( "client", ({"_request": [{"body": "channels_iter"}, {"body": "channels"}]},), indirect=True, ) def test_iter_no_wait(self, client): client.sleep = asynctest.CoroutineMock() channels = 0 for _ in client.iter(methods.CHANNELS_LIST, minimum_time=1): # noQa: F841 channels += 1 time.sleep(0.5) assert channels == 4 assert client._request.call_count == 2 assert client.sleep.call_count == 0 @pytest.mark.parametrize( "client", ({"_request": [{"body": "auth_test"}, {"body": "users_info"}]},), indirect=True, ) def test_find_bot_id(self, client): bot_id = client._find_bot_id() assert bot_id == "B0AAA0A00" @pytest.mark.parametrize( "client", ({"_request": {"body": "rtm_connect"}},), indirect=True ) def test_find_rtm_url(self, client): url = client._find_rtm_url() assert url == "wss://testteam.slack.com/012345678910" def test_incoming_rtm(self, client, rtm_iterator_non_async): client._rtm = rtm_iterator_non_async events = [] for event in client._incoming_from_rtm( "wss://testteam.slack.com/012345678910", "B0AAA0A00" ): assert isinstance(event, slack.events.Event) events.append(event) assert len(events) > 0 @pytest.mark.parametrize("rtm_iterator_non_async", (("goodbye",),), indirect=True) def test_incoming_rtm_reconnect(self, client, rtm_iterator_non_async): client._rtm = rtm_iterator_non_async events = [] for event in client._incoming_from_rtm( "wss://testteam.slack.com/012345678910", "B0AAA0A00" ): events.append(event) assert len(events) == 0 @pytest.mark.parametrize( "rtm_iterator_non_async", (("message_bot",),), indirect=True ) def test_incoming_rtm_discard_bot_id(self, client, rtm_iterator_non_async): client._rtm = rtm_iterator_non_async events = [] for event in client._incoming_from_rtm( "wss://testteam.slack.com/012345678910", "B0AAA0A00" ): events.append(event) assert len(events) == 0 @pytest.mark.parametrize( "rtm_iterator_non_async", (("reconnect_url",),), indirect=True ) def test_incoming_rtm_skip(self, client, rtm_iterator_non_async): client._rtm = rtm_iterator_non_async events = [] for event in client._incoming_from_rtm( "wss://testteam.slack.com/012345678910", "B0AAA0A00" ): events.append(event) assert len(events) == 0 class TestRequest: def test_sleep(self, token): delay = 1 start = datetime.datetime.now() with requests.Session() as session: client = SlackAPIRequest(session=session, token=token) client.sleep(delay) stop = datetime.datetime.now() assert ( datetime.timedelta(seconds=delay + 1) > (stop - start) > datetime.timedelta(seconds=delay) ) def test__request(self, token): with requests.Session() as session: client = SlackAPIRequest(session=session, token=token) response = client._request( "POST", "https://slack.com/api//api.test", {}, {"token": token} ) assert response[0] == 200 assert response[1] == b'{"ok":false,"error":"invalid_auth"}' class TestAiohttp: @pytest.mark.asyncio async def test_sleep(self, token): delay = 1 start = datetime.datetime.now() async with aiohttp.ClientSession() as session: client = SlackAPIAiohttp(session=session, token=token) await client.sleep(delay) stop = datetime.datetime.now() assert ( datetime.timedelta(seconds=delay + 1) > (stop - start) > datetime.timedelta(seconds=delay) ) @pytest.mark.asyncio async def test__request(self, token): async with aiohttp.ClientSession() as session: client = SlackAPIAiohttp(session=session, token=token) response = await client._request( "POST", "https://slack.com/api//api.test", {}, {"token": token} ) assert response[0] == 200 assert response[1] == b'{"ok":false,"error":"invalid_auth"}' class TestTrio: def test_sleep(self, token): asks.init("trio") async def test_function(): delay = 1 start = datetime.datetime.now() session = asks.Session() client = SlackAPITrio(session=session, token=token) await client.sleep(delay) stop = datetime.datetime.now() return ( datetime.timedelta(seconds=delay + 1) > (stop - start) > datetime.timedelta(seconds=delay) ) assert trio.run(test_function) def test__request(self, token): asks.init("trio") async def test_function(): session = asks.Session() client = SlackAPITrio(session=session, token=token) response = await client._request( "POST", "https://slack.com/api//api.test", {}, {"token": token} ) return response[0], response[1] assert trio.run(test_function) == (200, b'{"ok":false,"error":"invalid_auth"}') class TestCurio: def test_sleep(self, token): asks.init("curio") async def test_function(): delay = 1 start = datetime.datetime.now() session = asks.Session() client = SlackAPICurio(session=session, token=token) await client.sleep(delay) stop = datetime.datetime.now() return ( datetime.timedelta(seconds=delay + 1) > (stop - start) > datetime.timedelta(seconds=delay) ) assert curio.run(test_function) def test__request(self, token): asks.init("curio") async def test_function(): session = asks.Session() client = SlackAPICurio(session=session, token=token) response = await client._request( "POST", "https://slack.com/api//api.test", {}, {"token": token} ) return response[0], response[1] assert curio.run(test_function) == (200, b'{"ok":false,"error":"invalid_auth"}') PK!Lf==slack/tests/test_sansio.pyimport json import time import logging import mock import pytest from slack import sansio, methods, exceptions from slack.events import Event from . import data class TestRequest: def test_prepare_request(self, token): url, body, headers = sansio.prepare_request( methods.AUTH_TEST, {}, {}, {}, token ) assert url == "https://slack.com/api/auth.test" assert body == "{}" assert "Authorization" in headers assert "Content-type" in headers assert headers["Content-type"] == "application/json; charset=utf-8" url, body, headers = sansio.prepare_request( methods.AUTH_REVOKE, {}, {}, {}, token ) assert url == "https://slack.com/api/auth.revoke" assert body == {"token": token} assert headers == {} @pytest.mark.parametrize( "url", (methods.AUTH_TEST, "auth.test", "https://slack.com/api/auth.test") ) def test_prepare_request_urls(self, url): clean_url, _, _ = sansio.prepare_request(url, {}, {}, {}, "") assert clean_url == "https://slack.com/api/auth.test" def test_prepare_request_url_hook(self): clean_url, _, _ = sansio.prepare_request( "https://hooks.slack.com/T0000000/aczvrfver", {}, {}, {}, "" ) assert clean_url == "https://hooks.slack.com/T0000000/aczvrfver" @pytest.mark.parametrize( "payload,result", ( ({"foo": "bar"}, {"foo": "bar"}), ( {"foo": "bar", "attachements": [{"a": "b"}]}, {"foo": "bar", "attachements": [{"a": "b"}]}, ), ), ) def test_prepare_request_body(self, token, payload, result): _, body, headers = sansio.prepare_request( methods.AUTH_TEST, payload, {}, {}, token ) assert isinstance(body, str) assert body == json.dumps(result) assert "Authorization" in headers assert "Content-type" in headers assert headers["Content-type"] == "application/json; charset=utf-8" _, body, headers = sansio.prepare_request( methods.AUTH_REVOKE, payload, {}, {}, token ) result["token"] = token assert isinstance(body, dict) assert body == result @pytest.mark.parametrize( "payload,result", ( ({"foo": "bar"}, '{"foo": "bar"}'), ( {"foo": "bar", "attachements": [{"a": "b"}]}, '{"foo": "bar", "attachements": [{"a": "b"}]}', ), ), ) def test_prepare_request_body_hook(self, token, payload, result): _, body, headers = sansio.prepare_request( "https://hooks.slack.com/abcdefg", payload, {}, {}, token ) assert body == result assert "Authorization" in headers assert "Content-type" in headers assert headers["Content-type"] == "application/json; charset=utf-8" def test_prepare_request_body_message(self, token, message): msg = Event.from_http(message) _, body, headers = sansio.prepare_request(methods.AUTH_TEST, msg, {}, {}, token) assert isinstance(body, str) assert "Authorization" in headers assert "Content-type" in headers assert headers["Content-type"] == "application/json; charset=utf-8" _, body, headers = sansio.prepare_request( methods.AUTH_REVOKE, msg, {}, {}, token ) assert isinstance(body, dict) assert isinstance(body.get("attachments", ""), str) assert body["token"] == token def test_prepare_request_body_message_force_json(self, token, message): msg = Event.from_http(message) _, body, headers = sansio.prepare_request( methods.AUTH_REVOKE, msg, {}, {}, token, as_json=True ) assert isinstance(body, str) assert "Authorization" in headers assert "Content-type" in headers assert headers["Content-type"] == "application/json; charset=utf-8" def test_prepare_request_message_hook(self, token, message): msg = Event.from_http(message) _, body, headers = sansio.prepare_request( "https://hooks.slack.com/abcdefg", msg, {}, {}, token ) assert isinstance(body, str) data = json.loads(body) assert isinstance(data.get("attachments", []), list) assert "Authorization" in headers assert "Content-type" in headers assert headers["Content-type"] == "application/json; charset=utf-8" @pytest.mark.parametrize( "headers,global_headers,result", ( ({"foo": "bar", "py": "3.7"}, {}, {"foo": "bar", "py": "3.7"}), ( {"foo": "bar", "py": "3.7"}, {"sans": "I/O"}, {"foo": "bar", "py": "3.7", "sans": "I/O"}, ), ( {"foo": "bar", "py": "3.7"}, {"foo": "baz", "sans": "I/O"}, {"foo": "bar", "py": "3.7", "sans": "I/O"}, ), ), ) def test_prepare_request_headers(self, headers, global_headers, result): _, _, headers = sansio.prepare_request("", {}, headers, global_headers, "") assert headers == result def test_find_iteration(self): itermode, iterkey = sansio.find_iteration(methods.CHANNELS_LIST) assert itermode == methods.CHANNELS_LIST.value[1] assert iterkey == methods.CHANNELS_LIST.value[2] def test_find_iteration_custom_itermode(self): itermode, iterkey = sansio.find_iteration( methods.CHANNELS_LIST, itermode="timeline" ) assert itermode == "timeline" assert iterkey == methods.CHANNELS_LIST.value[2] def test_find_iteration_custom_iterkey(self): itermode, iterkey = sansio.find_iteration( methods.CHANNELS_LIST, iterkey="users" ) assert itermode == methods.CHANNELS_LIST.value[1] assert iterkey == "users" def test_find_iteration_not_found(self): with pytest.raises(ValueError): _, _ = sansio.find_iteration("") def test_find_iteration_wrong_mode(self): with pytest.raises(ValueError): _, _ = sansio.find_iteration("", itermode="python", iterkey="users") def test_prepare_iter_request(self): data, iterkey, itermode = sansio.prepare_iter_request(methods.CHANNELS_LIST, {}) assert data == {"limit": 200} assert itermode == methods.CHANNELS_LIST.value[1] assert iterkey == methods.CHANNELS_LIST.value[2] def test_prepare_iter_request_no_iterkey(self): data, iterkey, itermode = sansio.prepare_iter_request(methods.CHANNELS_LIST, {}) assert data == {"limit": 200} assert itermode == methods.CHANNELS_LIST.value[1] assert iterkey == methods.CHANNELS_LIST.value[2] def test_prepare_iter_request_cursor(self): data1, _, _ = sansio.prepare_iter_request( "", {}, itermode="cursor", iterkey="channels", itervalue="abcdefg" ) assert data1 == {"limit": 200, "cursor": "abcdefg"} data2, _, _ = sansio.prepare_iter_request( "", {}, itermode="cursor", itervalue="abcdefg", iterkey="channels", limit=300, ) assert data2 == {"limit": 300, "cursor": "abcdefg"} def test_prepare_iter_request_page(self): data1, _, _ = sansio.prepare_iter_request( "", {}, itermode="page", iterkey="channels", itervalue="abcdefg" ) assert data1 == {"count": 200, "page": "abcdefg"} data2, _, _ = sansio.prepare_iter_request( "", {}, itermode="page", itervalue="abcdefg", iterkey="channels", limit=300 ) assert data2 == {"count": 300, "page": "abcdefg"} def test_prepare_iter_request_timeline(self): data1, _, _ = sansio.prepare_iter_request( "", {}, itermode="timeline", iterkey="channels", itervalue="abcdefg" ) assert data1 == {"count": 200, "latest": "abcdefg"} data2, _, _ = sansio.prepare_iter_request( "", {}, itermode="timeline", itervalue="abcdefg", iterkey="channels", limit=300, ) assert data2 == {"count": 300, "latest": "abcdefg"} class TestResponse: def test_raise_for_status_200(self): try: sansio.raise_for_status(200, {}, {}) except Exception as exc: raise pytest.fail("RAISE {}".format(exc)) def test_raise_for_status_400(self): with pytest.raises(exceptions.HTTPException): sansio.raise_for_status(400, {}, {}) def test_raise_for_status_400_httpexception(self): with pytest.raises(exceptions.HTTPException) as exc: sansio.raise_for_status( 400, {"test-header": "hello"}, {"test-data": "world"} ) assert exc.type == exceptions.HTTPException assert exc.value.status == 400 assert exc.value.headers == {"test-header": "hello"} assert exc.value.data == {"test-data": "world"} def test_raise_for_status_429(self): with pytest.raises(exceptions.RateLimited) as exc: sansio.raise_for_status(429, {}, {}) assert exc.type == exceptions.RateLimited assert exc.value.retry_after == 1 def test_raise_for_status_429_headers(self): headers = {"Retry-After": "10"} with pytest.raises(exceptions.RateLimited) as exc: sansio.raise_for_status(429, headers, {}) assert exc.type == exceptions.RateLimited assert exc.value.retry_after == 10 def test_raise_for_status_429_wrong_headers(self): headers = {"Retry-After": "aa"} with pytest.raises(exceptions.RateLimited) as exc: sansio.raise_for_status(429, headers, {}) assert exc.type == exceptions.RateLimited assert exc.value.retry_after == 1 def test_raise_for_api_error_ok(self): try: sansio.raise_for_api_error({}, {"ok": True}) except Exception as exc: raise pytest.fail("RAISE {}".format(exc)) def test_raise_for_api_error_nok(self): data = {"ok": False} headers = {"test-header": "hello"} with pytest.raises(exceptions.SlackAPIError) as exc: sansio.raise_for_api_error(headers, data) assert exc.type == exceptions.SlackAPIError assert exc.value.headers == {"test-header": "hello"} assert exc.value.data == {"ok": False} assert exc.value.error == "unknow_error" def test_raise_for_api_error_nok_with_error(self): data = {"ok": False, "error": "test_error"} with pytest.raises(exceptions.SlackAPIError) as exc: sansio.raise_for_api_error({}, data) assert exc.type == exceptions.SlackAPIError assert exc.value.error == "test_error" def test_raise_for_api_error_warning(self, caplog): caplog.set_level(logging.WARNING) data = {"ok": True, "warning": "test warning"} sansio.raise_for_api_error({}, data) assert len(caplog.records) == 1 assert caplog.records[0].msg == "Slack API WARNING: %s" assert caplog.records[0].args == ("test warning",) def test_decode_body_string_ok(self): body = b"ok" decoded_body = sansio.decode_body({}, body) assert decoded_body == {"ok": True} def test_decode_body_string_nok(self): body = b"hello world" decoded_body = sansio.decode_body({}, body) assert decoded_body == {"ok": False, "data": "hello world"} def test_decode_body_json(self): body = b'{"test-string":"hello","test-bool":true}' headers = {"content-type": "application/json; charset=utf-8"} decoded_body = sansio.decode_body(headers, body) assert decoded_body == {"test-string": "hello", "test-bool": True} def test_decode_body_json_no_charset(self): body = b'{"test-string":"hello","test-bool":true}' headers = {"content-type": "application/json"} decoded_body = sansio.decode_body(headers, body) assert decoded_body == {"test-string": "hello", "test-bool": True} def test_decode_response(self): headers = {"content-type": "application/json; charset=utf-8"} data = b'{"ok": true, "hello": "world"}' try: data = sansio.decode_response(200, headers, data) except Exception as exc: pytest.fail("RAISE {}".format(exc)) else: assert data == {"ok": True, "hello": "world"} def test_decode_iter_request_cursor(self): data = {"response_metadata": {"next_cursor": "abcdefg"}} cursor = sansio.decode_iter_request(data) assert cursor == "abcdefg" def test_decode_iter_request_paging(self): data = {"paging": {"page": 2, "pages": 4}} page = sansio.decode_iter_request(data) assert page == 3 def test_decode_iter_request_timeline(self): timestamp = time.time() latest = timestamp - 1000 data = {"has_more": True, "latest": timestamp, "messages": [{"ts": latest}]} next_ = sansio.decode_iter_request(data) assert next_ == latest @mock.patch("time.time", mock.MagicMock(return_value=1534688291)) def test_validate_request_signature_ok(self): headers = { "X-Slack-Request-Timestamp": "1534688291", "X-Slack-Signature": "v0=ac720e09cb1ecb0baa17bea5638fa3d11fc177576dd364e05475d6dbc620c696", } body = """{"token":"abcdefghijkl","team_id":"T000000","api_app_id":"A000000","event":{},"type":"event_callback","authed_teams":["T000000"],"event_id":"AAAAAAA","event_time":1111111111}""" sansio.validate_request_signature( body=body, headers=headers, signing_secret="mysupersecret" ) @mock.patch("time.time", mock.MagicMock(return_value=1534688291)) def test_validate_request_signature_nok(self): headers = { "X-Slack-Request-Timestamp": "1534688291", "X-Slack-Signature": "v0=ac720e09cb1ecb0baa17bea5638fa3d11fc177576dd364e05475d6dbc620c697", } body = """{"token":"abcdefghijkl","team_id":"T000000","api_app_id":"A000000","event":{},"type":"event_callback","authed_teams":["T000000"],"event_id":"AAAAAAA","event_time":1111111111}""" with pytest.raises(exceptions.InvalidSlackSignature): sansio.validate_request_signature( body=body, headers=headers, signing_secret="mysupersecret" ) def test_validate_request_signature_too_old(self): headers = { "X-Slack-Request-Timestamp": "1534688291", "X-Slack-Signature": "v0=ac720e09cb1ecb0baa17bea5638fa3d11fc177576dd364e05475d6dbc620c696", } body = """{"token":"abcdefghijkl","team_id":"T000000","api_app_id":"A000000","event":{},"type":"event_callback","authed_teams":["T000000"],"event_id":"AAAAAAA","event_time":1111111111}""" with pytest.raises(exceptions.InvalidTimestamp): sansio.validate_request_signature( body=body, headers=headers, signing_secret="mysupersecret" ) class TestIncomingEvent: @pytest.mark.parametrize("event", ("bot", "bot_edit"), indirect=True) def test_discard_event(self, event): ev = Event.from_http(event) assert sansio.discard_event(ev, "B0AAA0A00") is True def test_not_discard_event(self, event): ev = Event.from_http(event) assert sansio.discard_event(ev, "B0AAA0A01") is False def test_no_need_reconnect(self, event): ev = Event.from_http(event) assert sansio.need_reconnect(ev) is False PK!;++$slack_sansio-0.7.0.dist-info/LICENSEMIT License Copyright (c) 2017 pyslackers Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. PK!HnHTU"slack_sansio-0.7.0.dist-info/WHEEL A н#Z;/"d&F[xzw@Zpy3Fv]\fi4WZ^EgM_-]#0(q7PK!Hˈn\%slack_sansio-0.7.0.dist-info/METADATA]O0+N5;V05ץѰ{[Dxםyل$oLi.&jx!I5gIBlgR@NP; g#BY($)ޘS&-LF/0ag 1*!a*"IEc"ub;jWv<Үi@\4SW=FƩOڜPuWlX 0@WJ>Dzuf窀Tjnx:LPP\d;rd5uZ 3>9N_PK!H @F #slack_sansio-0.7.0.dist-info/RECORDǖVཟa1 @ 3l8(rzQ{*{wW0- T0(HD ~^)|uPp6b:9V\sٵ z>u`|ՅQfU-<ɦG(FPPnaJF7Jg(}LFr[-7t$$(:cN!kίNCk"]\&̠+51>}kv{=K^uUdUNhoˆb:xϻQ Ȑ]SĽnN@Sv<lw ?ltiя,Hr=ػj}N@NiXv/&tPb)taInאAtqI h톓"*鰤ӗdC=q-L _FQF/F4(T ƨH%T 70ҹ辌fKn'N$uxnSvlH|29m9'Uާ?)l/YY碭OByjg@IWvfڊ%>y)\6z2-vkup)#^Qf9TN[9peu9zcrGsy?x` EUCA5SV\; CS<{1?`vwv<_8#⺊9`ũf -e҃=ѷkN sWWK@ 5AV4u4j>@|݊['j(9)ø}l3WBFld.Wol_c loQoym*ؕĠƙbeZK/ LTV %t'23W{fH?Xw@ 'Q(7,ĿB^;*'Z̖(!$ PovUofdx]ӥRBdQзZA^y[]rӈvuZr\}7斨>?!o/gZ݁6ըq3yqa!M5Pkc7vIy(HAn6tmt`W@4:?Cfduv)4X8>*\h(USܮ_>;g?ief@Md8v:cllw6)0v gGڤ?ڧ"AC![Jzq9%jW]8α~ PK!|``slack/__init__.pyPK!6slack/actions.pyPK!\c slack/commands.pyPK!<݀&&slack/events.pyPK!J  =Dslack/exceptions.pyPK!uOslack/io/__init__.pyPK!0-B<<Oslack/io/abc.pyPK!ZvWWjslack/io/aiohttp.pyPK!K1++oslack/io/curio.pyPK!0gsslack/io/requests.pyPK!,A))Ԍslack/io/trio.pyPK!;((+slack/methods.pyPK!Eslack/py.typedPK!X;+;+qslack/sansio.pyPK!slack/tests/__init__.pyPK! _??slack/tests/conftest.pyPK!Hslack/tests/data/__init__.pyPK!A)! ! slack/tests/data/actions.pyPK!='.. slack/tests/data/commands.pyPK!;<3<3gslack/tests/data/events.pyPK!ppDslack/tests/data/methods.pyPK!ed&+Sslack/tests/plugin.pyPK!u큼Tslack/tests/test.shPK!]{iNNUslack/tests/test_actions.pyPK!2 Dgslack/tests/test_commands.pyPK!'o55|tslack/tests/test_events.pyPK!@@slack/tests/test_io.pyPK!Lf==slack/tests/test_sansio.pyPK!;++$)slack_sansio-0.7.0.dist-info/LICENSEPK!HnHTU"'.slack_sansio-0.7.0.dist-info/WHEELPK!Hˈn\%.slack_sansio-0.7.0.dist-info/METADATAPK!H @F #l0slack_sansio-0.7.0.dist-info/RECORDPK 5