karaolidis/fbchat
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #510 from carpedm20/improve-exceptions

Browse Source 
Improve exceptions
This commit is contained in:
Mads Marquart
2020-01-15 16:07:37 +01:00
committed by GitHub
parent 656281eacb c0425193d0
commit 9228ac698d
17 changed files with 506 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
fbchat/__init__.py
View File

@@ -12,7 +12,16 @@ _logging.getLogger(__name__).addHandler(_logging.NullHandler())
# The order of these is somewhat significant, e.g. User has to be imported after Thread!
from . import _core, _util
from ._core import Image
from ._exception import FBchatException, FBchatFacebookError
from ._exception import (
FacebookError,
HTTPError,
ParseError,
ExternalError,
GraphQLError,
InvalidParameters,
NotLoggedIn,
PleaseRefresh,
)
from ._session import Session
from ._thread import ThreadLocation, ThreadABC, Thread
from ._user import User, UserData, ActiveStatus

95
fbchat/_client.py
View File

@@ -3,9 +3,19 @@ import time
import requests
from ._core import log
from . import _util, _graphql, _session, _poll, _user, _page, _group, _thread, _message
from . import (
_exception,
_util,
_graphql,
_session,
_poll,
_user,
_page,
_group,
_thread,
_message,
)
from ._exception import FBchatException, FBchatFacebookError
from ._thread import ThreadLocation
from ._user import User, UserData, ActiveStatus
from ._group import Group, GroupData
@@ -228,7 +238,7 @@ class Client:
j = self.session._payload_post("/chat/user_info/", data)
if j.get("profiles") is None:
raise FBchatException("No users/pages returned: {}".format(j))
raise _exception.ParseError("No users/pages returned", data=j)
entries = {}
for _id in j["profiles"]:
@@ -251,9 +261,7 @@ class Client:
"name": k.get("name"),
}
else:
raise FBchatException(
"{} had an unknown thread type: {}".format(_id, k)
)
raise _exception.ParseError("Unknown thread type", data=k)
log.debug(entries)
return entries
@@ -269,9 +277,6 @@ class Client:
Returns:
dict: `Thread` objects, labeled by their ID
Raises:
FBchatException: If request failed
"""
queries = []
for thread_id in thread_ids:
@@ -312,16 +317,16 @@ class Client:
elif entry.get("thread_type") == "ONE_TO_ONE":
_id = entry["thread_key"]["other_user_id"]
if pages_and_users.get(_id) is None:
raise FBchatException("Could not fetch thread {}".format(_id))
raise _exception.ParseError(
"Could not fetch thread {}".format(_id), data=pages_and_users
)
entry.update(pages_and_users[_id])
if "first_name" in entry:
rtn[_id] = UserData._from_graphql(self.session, entry)
else:
rtn[_id] = PageData._from_graphql(self.session, entry)
else:
raise FBchatException(
"{} had an unknown thread type: {}".format(thread_ids[i], entry)
)
raise _exception.ParseError("Unknown thread type", data=entry)
return rtn
@@ -389,9 +394,6 @@ class Client:
Returns:
list: List of unread thread ids
Raises:
FBchatException: If request failed
"""
form = {
"folders[0]": "inbox",
@@ -409,9 +411,6 @@ class Client:
Returns:
list: List of unseen thread ids
Raises:
FBchatException: If request failed
"""
j = self.session._payload_post("/mercury/unseen_thread_ids/", {})
@@ -426,18 +425,15 @@ class Client:
Returns:
str: An URL where you can download the original image
Raises:
FBchatException: If request failed
"""
image_id = str(image_id)
data = {"photo_id": str(image_id)}
j = self.session._post("/mercury/attachments/photo/", data)
_util.handle_payload_error(j)
_exception.handle_payload_error(j)
url = _util.get_jsmods_require(j, 3)
if url is None:
raise FBchatException("Could not fetch image URL from: {}".format(j))
raise _exception.ParseError("Could not fetch image URL", data=j)
return url
def _get_private_data(self):
@@ -488,12 +484,6 @@ class Client:
Args:
thread_id: User/Group ID to which the message belongs. See :ref:`intro_threads`
message_id: Message ID to set as delivered. See :ref:`intro_threads`
Returns:
True
Raises:
FBchatException: If request failed
"""
data = {
"message_ids[0]": message_id,
@@ -526,9 +516,6 @@ class Client:
Args:
thread_ids: User/Group IDs to set as read. See :ref:`intro_threads`
timestamp: Timestamp (as a Datetime) to signal the read cursor at, default is the current time
Raises:
FBchatException: If request failed
"""
self._read_status(True, thread_ids, timestamp)
@@ -540,9 +527,6 @@ class Client:
Args:
thread_ids: User/Group IDs to set as unread. See :ref:`intro_threads`
timestamp: Timestamp (as a Datetime) to signal the read cursor at, default is the current time
Raises:
FBchatException: If request failed
"""
self._read_status(False, thread_ids, timestamp)
@@ -561,12 +545,6 @@ class Client:
Args:
location (ThreadLocation): INBOX, PENDING, ARCHIVED or OTHER
thread_ids: Thread IDs to move. See :ref:`intro_threads`
Returns:
True
Raises:
FBchatException: If request failed
"""
thread_ids = _util.require_list(thread_ids)
@@ -597,12 +575,6 @@ class Client:
Args:
thread_ids: Thread IDs to delete. See :ref:`intro_threads`
Returns:
True
Raises:
FBchatException: If request failed
"""
thread_ids = _util.require_list(thread_ids)
@@ -624,12 +596,6 @@ class Client:
Args:
message_ids: Message IDs to delete
Returns:
True
Raises:
FBchatException: If request failed
"""
message_ids = _util.require_list(message_ids)
data = dict()
@@ -659,7 +625,7 @@ class Client:
"https://{}-edge-chat.facebook.com/active_ping".format(self._pull_channel),
data,
)
_util.handle_payload_error(j)
_exception.handle_payload_error(j)
def _pull_message(self):
"""Call pull api to fetch message data."""
@@ -674,7 +640,7 @@ class Client:
j = self.session._get(
"https://{}-edge-chat.facebook.com/pull".format(self._pull_channel), data
)
_util.handle_payload_error(j)
_exception.handle_payload_error(j)
return j
def _parse_delta(self, delta):
@@ -1237,16 +1203,19 @@ class Client:
self._parse_message(content)
except KeyboardInterrupt:
return False
except requests.Timeout:
pass
except requests.ConnectionError:
# If the client has lost their internet connection, keep trying every 30 seconds
time.sleep(30)
except FBchatFacebookError as e:
except _exception.HTTPError as e:
cause = e.__cause__
# Fix 502 and 503 pull errors
if e.request_status_code in [502, 503]:
if e.status_code in [502, 503]:
# Bump pull channel, while contraining withing 0-4
self._pull_channel = (self._pull_channel + 1) % 5
# TODO: Handle these exceptions better
elif isinstance(cause, requests.ReadTimeout):
pass # Expected
elif isinstance(cause, (requests.ConnectTimeout, requests.ConnectionError)):
# If the client has lost their internet connection, keep trying every 30 seconds
time.sleep(30)
else:
raise e
except Exception as e:

148
fbchat/_exception.py
View File

@@ -1,33 +1,84 @@
import attr
import requests
# Not frozen, since that doesn't work in PyPy
attrs_exception = attr.s(slots=True, auto_exc=True)
@attrs_exception
class FBchatException(Exception):
"""Custom exception thrown by ``fbchat``.
class FacebookError(Exception):
"""Base class for all custom exceptions raised by ``fbchat``.
All exceptions in the module inherits this.
All exceptions in the module inherit this.
"""
message = attr.ib()
#: A message describing the error
message = attr.ib(type=str)
@attrs_exception
class FBchatFacebookError(FBchatException):
"""Raised when Facebook returns an error."""
class HTTPError(FacebookError):
"""Base class for errors with the HTTP(s) connection to Facebook."""
#: The returned HTTP status code, if relevant
status_code = attr.ib(None, type=int)
def __str__(self):
if not self.status_code:
return self.message
return "Got {} response: {}".format(self.status_code, self.message)
@attrs_exception
class ParseError(FacebookError):
"""Raised when we fail parsing a response from Facebook.
This may contain sensitive data, so should not be logged to file.
"""
data = attr.ib()
"""The data that triggered the error.
The format of this cannot be relied on, it's only for debugging purposes.
"""
def __str__(self):
msg = "{}. Please report this, along with the data below!\n{}"
return msg.format(self.message, self.data)
@attrs_exception
class ExternalError(FacebookError):
"""Base class for errors that Facebook return."""
#: The error message that Facebook returned (Possibly in the user's own language)
description = attr.ib(type=str)
#: The error code that Facebook returned
fb_error_code = attr.ib(None)
#: The error message that Facebook returned (In the user's own language)
fb_error_message = attr.ib(None)
#: The status code that was sent in the HTTP response (e.g. 404) (Usually only set if not successful, aka. not 200)
request_status_code = attr.ib(None)
code = attr.ib(None, type=int)
def __str__(self):
if self.code:
return "#{} {}: {}".format(self.code, self.message, self.description)
return "{}: {}".format(self.message, self.description)
@attrs_exception
class FBchatInvalidParameters(FBchatFacebookError):
class GraphQLError(ExternalError):
"""Raised by Facebook if there was an error in the GraphQL query."""
# TODO: Handle multiple errors
#: Query debug information
debug_info = attr.ib(None, type=str)
def __str__(self):
if self.debug_info:
return "{}, {}".format(super().__str__(), self.debug_info)
return super().__str__()
@attrs_exception
class InvalidParameters(ExternalError):
"""Raised by Facebook if:
- Some function supplied invalid parameters.
@@ -37,18 +88,81 @@ class FBchatInvalidParameters(FBchatFacebookError):
@attrs_exception
class FBchatNotLoggedIn(FBchatFacebookError):
class NotLoggedIn(ExternalError):
"""Raised by Facebook if the client has been logged out."""
fb_error_code = attr.ib("1357001")
code = attr.ib(1357001)
@attrs_exception
class FBchatPleaseRefresh(FBchatFacebookError):
class PleaseRefresh(ExternalError):
"""Raised by Facebook if the client has been inactive for too long.
This error usually happens after 1-2 days of inactivity.
"""
fb_error_code = attr.ib("1357004")
fb_error_message = attr.ib("Please try closing and re-opening your browser window.")
code = attr.ib(1357004)
def handle_payload_error(j):
if "error" not in j:
return
code = j["error"]
if code == 1357001:
error_cls = NotLoggedIn
elif code == 1357004:
error_cls = PleaseRefresh
elif code in (1357031, 1545010, 1545003):
error_cls = InvalidParameters
else:
error_cls = ExternalError
raise error_cls(j["errorSummary"], description=j["errorDescription"], code=code)
def handle_graphql_errors(j):
errors = []
if j.get("error"):
errors = [j["error"]]
if "errors" in j:
errors = j["errors"]
if errors:
error = errors[0] # TODO: Handle multiple errors
# TODO: Use `severity` and `description`
raise GraphQLError(
# TODO: What data is always available?
message=error.get("summary", "Unknown error"),
description=error.get("message", ""),
code=error.get("code"),
debug_info=error.get("debug_info"),
)
def handle_http_error(code):
if code == 404:
raise HTTPError(
"This might be because you provided an invalid id"
+ " (Facebook usually require integer ids)",
status_code=code,
)
if code == 500:
raise HTTPError(
"There is probably an error on the endpoint, or it might be rate limited",
status_code=code,
)
if 400 <= code < 600:
raise HTTPError("Failed sending request", status_code=code)
def handle_requests_error(e):
if isinstance(e, requests.ConnectionError):
raise HTTPError("Connection error") from e
if isinstance(e, requests.HTTPError):
pass # Raised when using .raise_for_status, so should never happen
if isinstance(e, requests.URLRequired):
pass # Should never happen, we always prove valid URLs
if isinstance(e, requests.TooManyRedirects):
pass # TODO: Consider using allow_redirects=False to prevent this
if isinstance(e, requests.Timeout):
pass # Should never happen, we don't set timeouts
raise HTTPError("Requests error") from e

10
fbchat/_graphql.py
View File

@@ -38,19 +38,17 @@ def response_to_json(content):
content = _util.strip_json_cruft(content) # Usually only needed in some error cases
try:
j = json.loads(content, cls=ConcatJSONDecoder)
except Exception:
raise _exception.FBchatException(
"Error while parsing JSON: {!r}".format(content)
)
except Exception as e:
raise _exception.ParseError("Error while parsing JSON", data=content) from e
rtn = [None] * (len(j))
for x in j:
if "error_results" in x:
del rtn[-1]
continue
_util.handle_payload_error(x)
_exception.handle_payload_error(x)
[(key, value)] = x.items()
_util.handle_graphql_errors(value)
_exception.handle_graphql_errors(value)
if "response" in value:
rtn[int(key[1:])] = value["response"]
else:

13
fbchat/_message.py
View File

@@ -2,7 +2,16 @@ import attr
import enum
from string import Formatter
from ._core import log, attrs_default
from . import _util, _session, _attachment, _location, _file, _quick_reply, _sticker
from . import (
_exception,
_util,
_session,
_attachment,
_location,
_file,
_quick_reply,
_sticker,
)
from typing import Optional
@@ -112,7 +121,7 @@ class Message:
"variables": _util.json_minimal({"data": data}),
}
j = self.session._payload_post("/webgraphql/mutation", data)
_util.handle_graphql_errors(j)
_exception.handle_graphql_errors(j)
def fetch(self) -> "MessageData":
"""Fetch fresh `MessageData` object."""

5
fbchat/_plan.py
View File

@@ -53,10 +53,7 @@ class Plan:
}
j = thread.session._payload_post("/ajax/eventreminder/create", data)
if "error" in j:
raise _exception.FBchatFacebookError(
"Failed creating plan: {}".format(j["error"]),
fb_error_message=j["error"],
)
raise _exception.ExternalError("Failed creating plan", j["error"])
def edit(
self,

4
fbchat/_poll.py
View File

@@ -101,7 +101,7 @@ class Poll:
"/messaging/group_polling/update_vote/?dpr=1", data
)
if j.get("status") != "success":
raise _exception.FBchatFacebookError(
raise _exception.ExternalError(
"Failed updating poll vote: {}".format(j.get("errorTitle")),
fb_error_message=j.get("errorMessage"),
j.get("errorMessage"),
)

101
fbchat/_session.py
View File

@@ -13,9 +13,10 @@ FB_DTSG_REGEX = re.compile(r'name="fb_dtsg" value="(.*?)"')
def get_user_id(session):
# TODO: Optimize this `.get_dict()` call!
rtn = session.cookies.get_dict().get("c_user")
cookies = session.cookies.get_dict()
rtn = cookies.get("c_user")
if rtn is None:
raise _exception.FBchatException("Could not find user id")
raise _exception.ParseError("Could not find user id", data=cookies)
return str(rtn)
@@ -153,13 +154,15 @@ class Session:
password: Facebook account password
on_2fa_callback: Function that will be called, in case a 2FA code is needed.
This should return the requested 2FA code.
Raises:
FBchatException: On failed login
"""
session = session_factory()
soup = find_input_fields(session.get("https://m.facebook.com/").text)
try:
r = session.get("https://m.facebook.com/")
except requests.RequestException as e:
_exception.handle_requests_error(e)
soup = find_input_fields(r.text)
data = dict(
(elem["name"], elem["value"])
for elem in soup
@@ -169,29 +172,37 @@ class Session:
data["pass"] = password
data["login"] = "Log In"
r = session.post("https://m.facebook.com/login.php?login_attempt=1", data=data)
try:
url = "https://m.facebook.com/login.php?login_attempt=1"
r = session.post(url, data=data)
except requests.RequestException as e:
_exception.handle_requests_error(e)
# Usually, 'Checkpoint' will refer to 2FA
if "checkpoint" in r.url and ('id="approvals_code"' in r.text.lower()):
if not on_2fa_callback:
raise _exception.FBchatException(
raise ValueError(
"2FA code required, please add `on_2fa_callback` to .login"
)
code = on_2fa_callback()
r = _2fa_helper(session, code, r)
try:
r = _2fa_helper(session, code, r)
except requests.RequestException as e:
_exception.handle_requests_error(e)
# Sometimes Facebook tries to show the user a "Save Device" dialog
if "save-device" in r.url:
r = session.get("https://m.facebook.com/login/save-device/cancel/")
try:
r = session.get("https://m.facebook.com/login/save-device/cancel/")
except requests.RequestException as e:
_exception.handle_requests_error(e)
if is_home(r.url):
return cls._from_session(session=session)
else:
code, msg = get_error_data(r.text, r.url)
raise _exception.FBchatFacebookError(
"Login failed (Failed on url: {})".format(r.url),
fb_error_code=code,
fb_error_message=msg,
raise _exception.ExternalError(
"Login failed at url {!r}".format(r.url), msg, code=code
)
def is_logged_in(self):
@@ -202,36 +213,42 @@ class Session:
"""
# Send a request to the login url, to see if we're directed to the home page
url = "https://m.facebook.com/login.php?login_attempt=1"
r = self._session.get(url, allow_redirects=False)
try:
r = self._session.get(url, allow_redirects=False)
except requests.RequestException as e:
_exception.handle_requests_error(e)
return "Location" in r.headers and is_home(r.headers["Location"])
def logout(self):
"""Safely log out the user.
The session object must not be used after this action has been performed!
Raises:
FBchatException: On failed logout
"""
logout_h = self._logout_h
if not logout_h:
url = _util.prefix_url("/bluebar/modern_settings_menu/")
h_r = self._session.post(url, data={"pmid": "4"})
try:
h_r = self._session.post(url, data={"pmid": "4"})
except requests.RequestException as e:
_exception.handle_requests_error(e)
logout_h = re.search(r'name=\\"h\\" value=\\"(.*?)\\"', h_r.text).group(1)
url = _util.prefix_url("/logout.php")
r = self._session.get(url, params={"ref": "mb", "h": logout_h})
if not r.ok:
raise exception.FBchatException(
"Failed logging out: {}".format(r.status_code)
)
try:
r = self._session.get(url, params={"ref": "mb", "h": logout_h})
except requests.RequestException as e:
_exception.handle_requests_error(e)
_exception.handle_http_error(r.status_code)
@classmethod
def _from_session(cls, session):
# TODO: Automatically set user_id when the cookie changes in the session
user_id = get_user_id(session)
r = session.get(_util.prefix_url("/"))
try:
r = session.get(_util.prefix_url("/"))
except requests.RequestException as e:
_exception.handle_requests_error(e)
soup = find_input_fields(r.text)
@@ -242,9 +259,7 @@ class Session:
# Fall back to searching with a regex
res = FB_DTSG_REGEX.search(r.text)
if not res:
raise _exception.FBchatException(
"Failed loading session: Could not find fb_dtsg"
)
raise ValueError("Failed loading session, could not find fb_dtsg")
fb_dtsg = res.group(1)
revision = int(r.text.split('"client_revision":', 1)[1].split(",", 1)[0])
@@ -274,9 +289,6 @@ class Session:
Args:
cookies (dict): A dictionary containing session cookies
Raises:
FBchatException: If given invalid cookies
"""
session = session_factory()
session.cookies = requests.cookies.merge_cookies(session.cookies, cookies)
@@ -284,13 +296,19 @@ class Session:
def _get(self, url, params, error_retries=3):
params.update(self._get_params())
r = self._session.get(_util.prefix_url(url), params=params)
try:
r = self._session.get(_util.prefix_url(url), params=params)
except requests.RequestException as e:
_exception.handle_requests_error(e)
content = _util.check_request(r)
return _util.to_json(content)
def _post(self, url, data, files=None, as_graphql=False):
data.update(self._get_params())
r = self._session.post(_util.prefix_url(url), data=data, files=files)
try:
r = self._session.post(_util.prefix_url(url), data=data, files=files)
except requests.RequestException as e:
_exception.handle_requests_error(e)
content = _util.check_request(r)
if as_graphql:
return _graphql.response_to_json(content)
@@ -299,11 +317,11 @@ class Session:
def _payload_post(self, url, data, files=None):
j = self._post(url, data, files=files)
_util.handle_payload_error(j)
_exception.handle_payload_error(j)
try:
return j["payload"]
except (KeyError, TypeError):
raise _exception.FBchatException("Missing payload: {}".format(j))
except (KeyError, TypeError) as e:
raise _exception.ParseError("Missing payload", data=j) from e
def _graphql_requests(self, *queries):
data = {
@@ -330,9 +348,7 @@ class Session:
)
if len(j["metadata"]) != len(files):
raise _exception.FBchatException(
"Some files could not be uploaded: {}, {}".format(j, files)
)
raise _exception.ParseError("Some files could not be uploaded", data=j)
return [
(data[_util.mimetype_to_key(data["filetype"])], data["filetype"])
@@ -351,6 +367,8 @@ class Session:
data["ephemeral_ttl_mode:"] = "0"
j = self._post("/messaging/send/", data)
_exception.handle_payload_error(j)
# update JS token if received in response
fb_dtsg = _util.get_jsmods_require(j, 2)
if fb_dtsg is not None:
@@ -366,7 +384,4 @@ class Session:
log.warning("Got multiple message ids' back: {}".format(message_ids))
return message_ids[0]
except (KeyError, IndexError, TypeError) as e:
raise _exception.FBchatException(
"Error when sending message: "
"No message IDs could be found: {}".format(j)
)
raise _exception.ParseError("No message IDs could be found", data=j) from e

16
fbchat/_thread.py
View File

@@ -313,9 +313,7 @@ class ThreadABC(metaclass=abc.ABCMeta):
)
if j.get("message_thread") is None:
raise _exception.FBchatException(
"Could not fetch thread {}: {}".format(self.id, j)
)
raise _exception.ParseError("Could not fetch messages", data=j)
# TODO: Should we parse the returned thread data, too?
@@ -361,6 +359,9 @@ class ThreadABC(metaclass=abc.ABCMeta):
_graphql.from_query_id("515216185516880", data)
)
if not j[self.id]:
raise _exception.ParseError("Could not find images", data=j)
result = j[self.id]["message_shared_media"]
print(len(result["edges"]))
@@ -481,10 +482,7 @@ class ThreadABC(metaclass=abc.ABCMeta):
}
j = self.session._payload_post("/mercury/attachments/forward/", data)
if not j.get("success"):
raise _exception.FBchatFacebookError(
"Failed forwarding attachment: {}".format(j["error"]),
fb_error_message=j["error"],
)
raise _exception.ExternalError("Failed forwarding attachment", j["error"])
def _set_typing(self, typing):
data = {
@@ -547,9 +545,9 @@ class ThreadABC(metaclass=abc.ABCMeta):
"/messaging/group_polling/create_poll/?dpr=1", data
)
if j.get("status") != "success":
raise _exception.FBchatFacebookError(
raise _exception.ExternalError(
"Failed creating poll: {}".format(j.get("errorTitle")),
fb_error_message=j.get("errorMessage"),
j.get("errorMessage"),
)
def mute(self, duration: datetime.timedelta = None):

71
fbchat/_util.py
View File

@@ -5,13 +5,7 @@ import random
import urllib.parse
from ._core import log
from ._exception import (
FBchatException,
FBchatFacebookError,
FBchatInvalidParameters,
FBchatNotLoggedIn,
FBchatPleaseRefresh,
)
from . import _exception
from typing import Iterable, Optional
@@ -57,8 +51,8 @@ def strip_json_cruft(text):
"""Removes `for(;;);` (and other cruft) that preceeds JSON responses."""
try:
return text[text.index("{") :]
except ValueError:
raise FBchatException("No JSON object found: {!r}".format(text))
except ValueError as e:
raise _exception.ParseError("No JSON object found", data=text) from e
def get_decoded_r(r):
@@ -72,8 +66,8 @@ def get_decoded(content):
def parse_json(content):
try:
return json.loads(content)
except ValueError:
raise FBchatFacebookError("Error while parsing JSON: {!r}".format(content))
except ValueError as e:
raise _exception.ParseError("Error while parsing JSON", data=content) from e
def digit_to_char(digit):
@@ -109,67 +103,16 @@ def generate_offline_threading_id():
return str(int(msgs, 2))
def handle_payload_error(j):
if "error" not in j:
return
error = j["error"]
if j["error"] == 1357001:
error_cls = FBchatNotLoggedIn
elif j["error"] == 1357004:
error_cls = FBchatPleaseRefresh
elif j["error"] in (1357031, 1545010, 1545003):
error_cls = FBchatInvalidParameters
else:
error_cls = FBchatFacebookError
# TODO: Use j["errorSummary"]
# "errorDescription" is in the users own language!
raise error_cls(
"Error #{} when sending request: {}".format(error, j["errorDescription"]),
fb_error_code=error,
fb_error_message=j["errorDescription"],
)
def handle_graphql_errors(j):
errors = []
if j.get("error"):
errors = [j["error"]]
if "errors" in j:
errors = j["errors"]
if errors:
error = errors[0] # TODO: Handle multiple errors
# TODO: Use `summary`, `severity` and `description`
raise FBchatFacebookError(
"GraphQL error #{}: {} / {!r}".format(
error.get("code"), error.get("message"), error.get("debug_info")
),
fb_error_code=error.get("code"),
fb_error_message=error.get("message"),
)
def check_request(r):
check_http_code(r.status_code)
_exception.handle_http_error(r.status_code)
content = get_decoded_r(r)
check_content(content)
return content
def check_http_code(code):
msg = "Error when sending request: Got {} response.".format(code)
if code == 404:
raise FBchatFacebookError(
msg + " This is either because you specified an invalid URL, or because"
" you provided an invalid id (Facebook usually requires integer ids).",
request_status_code=code,
)
if 400 <= code < 600:
raise FBchatFacebookError(msg, request_status_code=code)
def check_content(content, as_json=True):
if content is None or len(content) == 0:
raise FBchatFacebookError("Error when sending request: Got empty response")
raise _exception.HTTPError("Error when sending request: Got empty response")
def to_json(content):

4
tests/test_base.py
View File

@@ -3,7 +3,7 @@ import py_compile
from glob import glob
from os import path, environ
from fbchat import FBchatException, Message, Client
from fbchat import FacebookError, Message, Client
def test_examples():
@@ -23,7 +23,7 @@ def test_login(client1):
assert not client1.is_logged_in()
with pytest.raises(FBchatException):
with pytest.raises(FacebookError):
client1.login("<invalid email>", "<invalid password>", max_tries=1)
client1.login(email, password)

228
tests/test_exception.py Normal file
View File

@@ -0,0 +1,228 @@
import pytest
import requests
from fbchat import (
FacebookError,
HTTPError,
ParseError,
ExternalError,
GraphQLError,
InvalidParameters,
NotLoggedIn,
PleaseRefresh,
)
from fbchat._exception import (
handle_payload_error,
handle_graphql_errors,
handle_http_error,
handle_requests_error,
)
ERROR_DATA = [
(NotLoggedIn, 1357001, "Not logged in", "Please log in to continue."),
(
PleaseRefresh,
1357004,
"Sorry, something went wrong",
"Please try closing and re-opening your browser window.",
),
(
InvalidParameters,
1357031,
"This content is no longer available",
(
"The content you requested cannot be displayed at the moment. It may be"
" temporarily unavailable, the link you clicked on may have expired or you"
" may not have permission to view this page."
),
),
(
InvalidParameters,
1545010,
"Messages Unavailable",
(
"Sorry, messages are temporarily unavailable."
" Please try again in a few minutes."
),
),
(
ExternalError,
1545026,
"Unable to Attach File",
(
"The type of file you're trying to attach isn't allowed."
" Please try again with a different format."
),
),
(InvalidParameters, 1545003, "Invalid action", "You cannot perform that action."),
(
ExternalError,
1545012,
"Temporary Failure",
"There was a temporary error, please try again.",
),
]
@pytest.mark.parametrize("exception,code,summary,description", ERROR_DATA)
def test_handle_payload_error(exception, code, summary, description):
data = {"error": code, "errorSummary": summary, "errorDescription": description}
with pytest.raises(exception, match=r"#\d+ .+:"):
handle_payload_error(data)
def test_handle_payload_error_no_error():
assert handle_payload_error({}) is None
assert handle_payload_error({"payload": {"abc": ["Something", "else"]}}) is None
def test_handle_graphql_crash():
error = {
"allow_user_retry": False,
"api_error_code": -1,
"code": 1675030,
"debug_info": None,
"description": "Error performing query.",
"fbtrace_id": "ABCDEFG",
"is_silent": False,
"is_transient": False,
"message": (
'Errors while executing operation "MessengerThreadSharedLinks":'
" At Query.message_thread: Field implementation threw an exception."
" Check your server logs for more information."
),
"path": ["message_thread"],
"query_path": None,
"requires_reauth": False,
"severity": "CRITICAL",
"summary": "Query error",
}
with pytest.raises(
GraphQLError, match="#1675030 Query error: Errors while executing"
):
handle_graphql_errors({"data": {"message_thread": None}, "errors": [error]})
def test_handle_graphql_invalid_values():
error = {
"message": (
'Invalid values provided for variables of operation "MessengerThreadlist":'
' Value ""as"" cannot be used for variable "$limit": Expected an integer'
' value, got "as".'
),
"severity": "CRITICAL",
"code": 1675012,
"api_error_code": None,
"summary": "Your request couldn't be processed",
"description": (
"There was a problem with this request."
" We're working on getting it fixed as soon as we can."
),
"is_silent": False,
"is_transient": False,
"requires_reauth": False,
"allow_user_retry": False,
"debug_info": None,
"query_path": None,
"fbtrace_id": "ABCDEFG",
"www_request_id": "AABBCCDDEEFFGG",
}
msg = "#1675012 Your request couldn't be processed: Invalid values"
with pytest.raises(GraphQLError, match=msg):
handle_graphql_errors({"errors": [error]})
def test_handle_graphql_no_message():
error = {
"code": 1675012,
"api_error_code": None,
"summary": "Your request couldn't be processed",
"description": (
"There was a problem with this request."
" We're working on getting it fixed as soon as we can."
),
"is_silent": False,
"is_transient": False,
"requires_reauth": False,
"allow_user_retry": False,
"debug_info": None,
"query_path": None,
"fbtrace_id": "ABCDEFG",
"www_request_id": "AABBCCDDEEFFGG",
"sentry_block_user_info": None,
"help_center_id": None,
}
msg = "#1675012 Your request couldn't be processed: "
with pytest.raises(GraphQLError, match=msg):
handle_graphql_errors({"errors": [error]})
def test_handle_graphql_no_summary():
error = {
"message": (
'Errors while executing operation "MessengerViewerContactMethods":'
" At Query.viewer:Viewer.all_emails: Field implementation threw an"
" exception. Check your server logs for more information."
),
"severity": "ERROR",
"path": ["viewer", "all_emails"],
}
with pytest.raises(GraphQLError, match="Unknown error: Errors while executing"):
handle_graphql_errors(
{"data": {"viewer": {"user": None, "all_emails": []}}, "errors": [error]}
)
def test_handle_graphql_syntax_error():
error = {
"code": 1675001,
"api_error_code": None,
"summary": "Query Syntax Error",
"description": "Syntax error.",
"is_silent": True,
"is_transient": False,
"requires_reauth": False,
"allow_user_retry": False,
"debug_info": 'Unexpected ">" at character 328: Expected ")".',
"query_path": None,
"fbtrace_id": "ABCDEFG",
"www_request_id": "AABBCCDDEEFFGG",
"sentry_block_user_info": None,
"help_center_id": None,
}
msg = "#1675001 Query Syntax Error: "
with pytest.raises(GraphQLError, match=msg):
handle_graphql_errors({"response": None, "error": error})
def test_handle_graphql_errors_singular_error_key():
with pytest.raises(GraphQLError, match="#123"):
handle_graphql_errors({"error": {"code": 123}})
def test_handle_graphql_errors_no_error():
assert handle_graphql_errors({"data": {"message_thread": None}}) is None
def test_handle_http_error():
with pytest.raises(HTTPError):
handle_http_error(400)
with pytest.raises(HTTPError):
handle_http_error(500)
def test_handle_http_error_404_handling():
with pytest.raises(HTTPError, match="invalid id"):
handle_http_error(404)
def test_handle_http_error_no_error():
assert handle_http_error(200) is None
assert handle_http_error(302) is None
def test_handle_requests_error():
with pytest.raises(HTTPError, match="Connection error"):
handle_requests_error(requests.ConnectionError())
with pytest.raises(HTTPError, match="Requests error"):
handle_requests_error(requests.RequestException())

2
tests/test_plans.py
View File

@@ -1,6 +1,6 @@
import pytest
from fbchat import PlanData, FBchatFacebookError
from fbchat import PlanData
from utils import random_hex, subset
from time import time

2
tests/test_send.py
View File

@@ -1,7 +1,7 @@
import pytest
from os import path
from fbchat import FBchatFacebookError, Message, Mention
from fbchat import Message, Mention
from utils import subset, STICKER_LIST, EMOJI_LIST, TEXT_LIST
pytestmark = pytest.mark.online

10
tests/test_thread_interraction.py
View File

@@ -1,6 +1,6 @@
import pytest
from fbchat import Message, FBchatFacebookError
from fbchat import Message, FacebookError
from utils import random_hex, subset
from os import path
@@ -60,10 +60,8 @@ def test_change_nickname(client, client_all, catch_event, compare):
"😂",
"😕",
"😍",
pytest.param("🙃", marks=[pytest.mark.xfail(raises=FBchatFacebookError)]),
pytest.param(
"not an emoji", marks=[pytest.mark.xfail(raises=FBchatFacebookError)]
),
pytest.param("🙃", marks=[pytest.mark.xfail(raises=FacebookError)]),
pytest.param("not an emoji", marks=[pytest.mark.xfail(raises=FacebookError)]),
],
)
def test_change_emoji(client, catch_event, compare, emoji):
@@ -97,7 +95,7 @@ def test_change_color(client, catch_event, compare):
assert compare(x, new_color="#44bec7")
@pytest.mark.xfail(raises=FBchatFacebookError, reason="Should fail, but doesn't")
@pytest.mark.xfail(raises=FacebookError, reason="Should fail, but doesn't")
def test_change_color_invalid(client):
class InvalidColor:
value = "#0077ff"

126
tests/test_util.py
View File

@@ -7,9 +7,6 @@ from fbchat._util import (
str_base,
generate_message_id,
get_signature_id,
handle_payload_error,
handle_graphql_errors,
check_http_code,
get_jsmods_require,
require_list,
mimetype_to_key,
@@ -33,7 +30,7 @@ def test_strip_json_cruft():
def test_strip_json_cruft_invalid():
with pytest.raises(AttributeError):
strip_json_cruft(None)
with pytest.raises(fbchat.FBchatException, match="No JSON object found"):
with pytest.raises(fbchat.ParseError, match="No JSON object found"):
strip_json_cruft("No JSON object here!")
@@ -42,7 +39,7 @@ def test_parse_json():
def test_parse_json_invalid():
with pytest.raises(fbchat.FBchatFacebookError, match="Error while parsing JSON"):
with pytest.raises(fbchat.ParseError, match="Error while parsing JSON"):
parse_json("No JSON object here!")
@@ -71,125 +68,6 @@ def test_get_signature_id():
get_signature_id()
ERROR_DATA = [
(
fbchat._exception.FBchatNotLoggedIn,
1357001,
"Not logged in",
"Please log in to continue.",
),
(
fbchat._exception.FBchatPleaseRefresh,
1357004,
"Sorry, something went wrong",
"Please try closing and re-opening your browser window.",
),
(
fbchat._exception.FBchatInvalidParameters,
1357031,
"This content is no longer available",
(
"The content you requested cannot be displayed at the moment. It may be"
" temporarily unavailable, the link you clicked on may have expired or you"
" may not have permission to view this page."
),
),
(
fbchat._exception.FBchatInvalidParameters,
1545010,
"Messages Unavailable",
(
"Sorry, messages are temporarily unavailable."
" Please try again in a few minutes."
),
),
(
fbchat.FBchatFacebookError,
1545026,
"Unable to Attach File",
(
"The type of file you're trying to attach isn't allowed."
" Please try again with a different format."
),
),
(
fbchat._exception.FBchatInvalidParameters,
1545003,
"Invalid action",
"You cannot perform that action.",
),
(
fbchat.FBchatFacebookError,
1545012,
"Temporary Failure",
"There was a temporary error, please try again.",
),
]
@pytest.mark.parametrize("exception,code,description,summary", ERROR_DATA)
def test_handle_payload_error(exception, code, summary, description):
data = {"error": code, "errorSummary": summary, "errorDescription": description}
with pytest.raises(exception, match=r"Error #\d+ when sending request"):
handle_payload_error(data)
def test_handle_payload_error_no_error():
assert handle_payload_error({}) is None
assert handle_payload_error({"payload": {"abc": ["Something", "else"]}}) is None
def test_handle_graphql_errors():
error = {
"allow_user_retry": False,
"api_error_code": -1,
"code": 1675030,
"debug_info": None,
"description": "Error performing query.",
"fbtrace_id": "CLkuLR752sB",
"is_silent": False,
"is_transient": False,
"message": (
'Errors while executing operation "MessengerThreadSharedLinks":'
" At Query.message_thread: Field implementation threw an exception."
" Check your server logs for more information."
),
"path": ["message_thread"],
"query_path": None,
"requires_reauth": False,
"severity": "CRITICAL",
"summary": "Query error",
}
with pytest.raises(fbchat.FBchatFacebookError, match="GraphQL error"):
handle_graphql_errors({"data": {"message_thread": None}, "errors": [error]})
def test_handle_graphql_errors_singular_error_key():
with pytest.raises(fbchat.FBchatFacebookError, match="GraphQL error #123"):
handle_graphql_errors({"error": {"code": 123}})
def test_handle_graphql_errors_no_error():
assert handle_graphql_errors({"data": {"message_thread": None}}) is None
def test_check_http_code():
with pytest.raises(fbchat.FBchatFacebookError):
check_http_code(400)
with pytest.raises(fbchat.FBchatFacebookError):
check_http_code(500)
def test_check_http_code_404_handling():
with pytest.raises(fbchat.FBchatFacebookError, match="invalid id"):
check_http_code(404)
def test_check_http_code_no_error():
assert check_http_code(200) is None
assert check_http_code(302) is None
def test_get_jsmods_require_get_image_url():
data = {
"__ar": 1,

14
tests/utils.py
View File

@@ -5,7 +5,7 @@ import pytest
from os import environ
from random import randrange
from contextlib import contextmanager
from fbchat import EmojiSize, FBchatFacebookError, Sticker, Client
from fbchat import EmojiSize, FacebookError, Sticker, Client
log = logging.getLogger("fbchat.tests").addHandler(logging.NullHandler())
@@ -23,12 +23,8 @@ EMOJI_LIST = [
STICKER_LIST = [
Sticker(id="767334476626295"),
pytest.param(
Sticker(id="0"), marks=[pytest.mark.xfail(raises=FBchatFacebookError)]
),
pytest.param(
Sticker(id=None), marks=[pytest.mark.xfail(raises=FBchatFacebookError)]
),
pytest.param(Sticker(id="0"), marks=[pytest.mark.xfail(raises=FacebookError)]),
pytest.param(Sticker(id=None), marks=[pytest.mark.xfail(raises=FacebookError)]),
]
TEXT_LIST = [
@@ -37,8 +33,8 @@ TEXT_LIST = [
"\\\n\t%?&'\"",
"ˁҭʚ¹Ʋջوװ՞ޱɣࠚԹБɑȑңКએ֭ʗыԈٌʼőԈ×௴nચϚࠖణٔє܅Ԇޑط",
"a" * 20000, # Maximum amount of characters you can send
pytest.param("a" * 20001, marks=[pytest.mark.xfail(raises=FBchatFacebookError)]),
pytest.param(None, marks=[pytest.mark.xfail(raises=FBchatFacebookError)]),
pytest.param("a" * 20001, marks=[pytest.mark.xfail(raises=FacebookError)]),
pytest.param(None, marks=[pytest.mark.xfail(raises=FacebookError)]),
]