Merge pull request #6 from devanshshukla99/status_code

Added status code in the dump

Author: devanshshukla99

pytest_response/app.py

@@ -11,8 +11,8 @@
 
 
 class BaseMockResponse:
-    def __init__(self, data: bytes, headers: dict = {}) -> None:
-        self.status = self.status_code = self.code = 200
+    def __init__(self, status: int, data: bytes, headers: dict = {}) -> None:
+        self.status = self.status_code = self.code = status
         self.msg = self.reason = "OK"
         self.headers = headers
         self.will_close = True
@@ -159,7 +159,7 @@ def register(self, mock: str) -> None:
 
         # Register for future use.
         self._registered_mocks[mock.stem] = mock_lib
-        log.info(f"{mock.name} registered")
+        log.debug(f"{mock.name} registered")
         return
 
     def registermany(self, mocks: List[str]) -> None:

pytest_response/database.py

@@ -2,10 +2,12 @@
 import zlib
 from base64 import b64decode, b64encode
 from datetime import date
-from urllib.parse import urlparse, urljoin
+from urllib.parse import urljoin, urlparse
 
 from tinydb import TinyDB, where
 
+from pytest_response.exceptions import MalformedUrl
+
 # from collections.abc import MutableMapping
 
 
@@ -37,19 +39,23 @@ def _sanatize_url(self, url: str):
         Expecting:
             from `http://www.python.org:80` -> `http://www.python.org`
         """
-        _urlparsed = urlparse(url)
-        _url = "://".join([_urlparsed.scheme, _urlparsed.hostname])
-        return urljoin(_url, _urlparsed.path)
-
-    def insert(self, url: str, response: bytes, headers: dict, **kwargs):
+        try:
+            _urlparsed = urlparse(url)
+            _url = "://".join([_urlparsed.scheme, _urlparsed.hostname])
+            return urljoin(_url, _urlparsed.path)
+        except Exception:
+            raise MalformedUrl
+
+    def insert(self, url: str, response: bytes, headers: dict, status: int = 200, **kwargs):
         """
         Method for dumping url, headers and responses to the database.
         All additonal kwargs are dumped as well.
         """
         kwargs.update({"url": self._sanatize_url(url)})
         kwargs.update({"cache_date": self.today})
-        kwargs.update({"response": b64encode(zlib.compress(response)).decode("utf-8")})
+        kwargs.update({"status": str(status)})
         kwargs.update({"headers": b64encode(zlib.compress(str(headers).encode("utf-8"))).decode("utf-8")})
+        kwargs.update({"response": b64encode(zlib.compress(response)).decode("utf-8")})
         self._database.upsert(kwargs, where("url") == url)
         return
 
@@ -60,12 +66,15 @@ def get(self, url: str, **kwargs):
         """
         query = where("url") == self._sanatize_url(url)  # and where("request") == "req"
         if element := self._database.search(query):
-            res = element[0].get("response")
+            status = element[0].get("status", 200)
             headers = element[0].get("headers", "[]")
-            return zlib.decompress(b64decode(res.encode("utf-8"))), ast.literal_eval(
-                zlib.decompress(b64decode(headers)).decode("utf-8")
+            res = element[0].get("response")
+            return (
+                int(status),
+                zlib.decompress(b64decode(res.encode("utf-8"))),
+                ast.literal_eval(zlib.decompress(b64decode(headers)).decode("utf-8")),
             )
-        return b"", {}
+        return 404, b"", {}
 
     def all(self):
         """

pytest_response/interceptors/requests_quick.py

@@ -16,26 +16,26 @@ def inner_func(url, params=None, **kwargs):
             log.error(f"RemoteBlockedError remote:{response.remote}")
             raise RemoteBlockedError
         if response.response:
-            data, headers = response.get(url=url)
+            status, data, headers = response.get(url=url)
             if not data:
                 log.error(f"Response not found url:{url}")
                 raise ResponseNotFound
-            return MockResponse(data, headers)
+            return MockResponse(status, data, headers)
         _ = func(url, params, **kwargs)
         if not response.capture:
             return _
         data = _.content
-        response.insert(url=url, response=data, headers=dict(_.headers))
+        response.insert(url=url, response=data, headers=dict(_.headers), status=_.status_code)
         return _
 
     return inner_func
 
 
 class MockResponse(BaseMockResponse):
-    def __init__(self, data, headers={}):
+    def __init__(self, status, data, headers={}):
         headers = urllib3.response.HTTPHeaderDict(headers)
         self.content = data
-        super().__init__(data, headers)
+        super().__init__(status, data, headers)
 
     pass
 

pytest_response/interceptors/urllib.py

@@ -196,7 +196,7 @@ def begin(self, *args, **kwargs):
         if response.response:
             global CONFIG
             self.fp = io.BytesIO()
-            data, headers = response.get(url=CONFIG.get("url", ""))
+            status, data, headers = response.get(url=CONFIG.get("url", ""))
             if not data:
                 self.code = self.status = 404
                 self.reason = "Response Not Found (pytest-response)"

pytest_response/interceptors/urllib3_quick.py

@@ -22,26 +22,26 @@ def inner_func(self, method, url, *args, **kwargs):
         if not response.remote:
             raise RemoteBlockedError
         if response.response:
-            data, headers = response.get(url=_url)
+            status, data, headers = response.get(url=_url)
             if not data:
                 log.error(f"Response not found url:{_url}")
                 raise ResponseNotFound
-            return MockResponse(data, headers)
+            return MockResponse(status, data, headers)
         _ = func(self, method, url, *args, **kwargs)
         if not response.capture:
             return _
         data = _._fp.read()
         _._fp = io.BytesIO(data)
-        response.insert(url=_url, response=data, headers=dict(_.headers))
+        response.insert(url=_url, response=data, headers=dict(_.headers), status=_.status)
         return _
 
     return inner_func
 
 
 class MockResponse(BaseMockResponse):
-    def __init__(self, data, headers={}):
+    def __init__(self, status, data, headers={}):
         headers = urllib3.response.HTTPHeaderDict(headers)
-        super().__init__(data, headers)
+        super().__init__(status, data, headers)
 
 
 def install_opener():

pytest_response/interceptors/urllib_quick.py

@@ -15,27 +15,27 @@ def inner_func(url, *args, **kwargs):
             log.error(f"RemoteBlockedError remote:{response.remote}")
             raise RemoteBlockedError
         if response.response:
-            data, headers = response.get(url=url)
+            status, data, headers = response.get(url=url)
             if not data:
                 log.error(f"Response not found url:{url}")
                 raise ResponseNotFound
-            return MockResponse(data, headers)
+            return MockResponse(status, data, headers)
 
         _ = func(url, *args, **kwargs)
         if not response.capture:
             return _
         data = _.fp.read()
         _.fp = io.BytesIO(data)
         headers = _.headers
-        response.insert(url=url, response=data, headers=dict(headers))
+        response.insert(url=url, response=data, headers=dict(headers), status=_.status)
         return _
 
     return inner_func
 
 
 class MockResponse(BaseMockResponse):
-    def __init__(self, data, headers={}):
-        super().__init__(data, headers)
+    def __init__(self, status, data, headers={}):
+        super().__init__(status, data, headers)
 
     pass
 

tests/test_app.py

@@ -45,7 +45,7 @@ def test_response_obj():
 
 
 def test_basemockresponse():
-    res = BaseMockResponse(b"Hello", headers={"Mock": True})
+    res = BaseMockResponse(200, b"Hello", headers={"Mock": True})
     assert res.code == res.status_code == res.status == res.getcode() == 200
     assert res.headers
     assert res.info()

tests/test_database.py

@@ -1,16 +1,21 @@
+import pytest
+
 from pytest_response.database import ResponseDB
+from pytest_response.exceptions import MalformedUrl
 
 
 def test_response_db(tmp_path):
     path = tmp_path / "temp_db.json"
     db = ResponseDB(path)
 
-    url = "testurl"
+    url = "https://testurl"
     data = b"Hello, this is a test."
     headers = {"Test": True}
+    status = 200
 
-    assert db.insert(url=url, response=data, headers=headers) is None
-    get_data, get_headers = db.get(url)
+    assert db.insert(url=url, response=data, headers=headers, status=status) is None
+    get_status, get_data, get_headers = db.get(url)
+    assert get_status == 200
     assert get_data == data
     assert get_headers == headers
 
@@ -20,8 +25,12 @@ def test_response_db(tmp_path):
     assert db.truncate() is None
     assert db.all() == []
 
-    noget_data, noget_headers = db.get("invalid_url")
+    noget_status, noget_data, noget_headers = db.get("https://invalid_url")
+    assert noget_status == 404
     assert noget_data == b""
     assert noget_headers == {}
 
+    with pytest.raises(MalformedUrl):
+        db.get("invalid_url")
+
     db.close()