Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 12 additions & 1 deletion massive/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,9 @@
from typing import Optional

# the package star-imports this module; keep Optional out of its namespace
__all__ = ["AuthError", "BadResponse"]


class AuthError(Exception):
"""
Empty or invalid API key
Expand All @@ -9,6 +15,11 @@ class AuthError(Exception):
class BadResponse(Exception):
"""
Non-200 response from API

The message is the response body. ``status`` is the HTTP status code, or
None if not given.
"""

pass
def __init__(self, *args: object, status: Optional[int] = None):
super().__init__(*args)
self.status = status
2 changes: 1 addition & 1 deletion massive/rest/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ def _get(
)

if resp.status != 200:
raise BadResponse(resp.data.decode("utf-8"))
raise BadResponse(resp.data.decode("utf-8"), status=resp.status)

if raw:
return resp
Expand Down
38 changes: 38 additions & 0 deletions test_rest/test_bad_response.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
from massive.exceptions import BadResponse
from base import BaseTest

import pook # type: ignore


class BadResponseTest(BaseTest):
def test_status_is_set(self):
cases = [
(403, '{"status":"NOT_AUTHORIZED","message":"not entitled"}'),
(404, '{"status":"NOT_FOUND","message":"not found"}'),
]
for status, body in cases:
with self.subTest(status=status):
ticker = f"ERR{status}" # not covered by the fixture mocks
pook.get(
self.c.BASE + "/v3/reference/tickers/" + ticker,
reply=status,
response_body=body,
)
with self.assertRaises(BadResponse) as ctx:
self.c.get_ticker_details(ticker)
self.assertEqual(ctx.exception.status, status)
self.assertEqual(ctx.exception.args, (body,))
self.assertEqual(str(ctx.exception), body)

def test_status_defaults_to_none(self):
for args in [(), ("boom",), ("boom", 1)]:
with self.subTest(args=args):
exc = BadResponse(*args)
self.assertEqual(exc.args, args)
self.assertIsNone(exc.status)

def test_status_does_not_change_message(self):
exc = BadResponse("boom", status=500)
self.assertEqual(exc.args, ("boom",))
self.assertEqual(str(exc), "boom")
self.assertEqual(exc.status, 500)
Loading