From 2910fcc6a69e38e0056893b657f088d5cf73e47c Mon Sep 17 00:00:00 2001 From: willforde Date: Wed, 24 Feb 2021 22:36:36 +0000 Subject: [PATCH] update tests --- .coveragerc | 2 +- tests/conftest.py | 96 +++++++++++--- tests/test_misc.py | 10 ++ tests/test_response.py | 46 +++++++ tests/test_session.py | 276 ++++++++++++++++++++++++++++++++++++++++- urlquick.py | 28 ++--- 6 files changed, 414 insertions(+), 44 deletions(-) create mode 100644 tests/test_response.py diff --git a/.coveragerc b/.coveragerc index 965b7eb..b47dfd5 100644 --- a/.coveragerc +++ b/.coveragerc @@ -1,6 +1,6 @@ [run] source=urlquick -branch = True +branch=True [report] exclude_lines = diff --git a/tests/conftest.py b/tests/conftest.py index 13542b6..9e6ccd7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,28 +1,22 @@ import requests import urlquick import pytest +import json as _json import io from requests.adapters import HTTPResponse from requests import adapters -class RequestsMock(object): - def __init__(self, mocker): - self.mocker = mocker - self._store = {} - - def mock_send(self_mock, request, **_): - urlid = urlquick.hash_url(request) - assert urlid in self._store - return self_mock.build_response(request, self._store[urlid]) - - mocker.patch.object(adapters.HTTPAdapter, "send", mock_send) +class MockResponse(object): + """Mock response that keeps track of when or if the response was called.""" + def __init__(self, body="", json=None, **kwargs): + if json is not None: + body = _json.dumps(json) - @staticmethod - def build_urllib3_response(text="", **kwargs): # type: (...) -> HTTPResponse # Alieas to some common params - kwargs["body"] = io.BytesIO(urlquick.to_bytes_string(text)) + self._body = body + kwargs["body"] = io.BytesIO(urlquick.to_bytes_string(body)) kwargs.setdefault("status", 200) kwargs.setdefault("reason", "OK") kwargs["preload_content"] = False @@ -31,9 +25,53 @@ def build_urllib3_response(text="", **kwargs): # type: (...) -> HTTPResponse # Add mock headers headers = kwargs.setdefault("headers", {}) headers.setdefault("Content-Type", "text/html; charset=utf8") - return HTTPResponse(**kwargs) + self._kwargs = kwargs + self._called = 0 + + @property + def response(self): + """Return the response but increment called counter.""" + self._called += 1 + self._kwargs["body"] = io.BytesIO(urlquick.to_bytes_string(self._body)) + return HTTPResponse(**self._kwargs) + + @property + def called(self): + """State that the mock was called at least once.""" + return self._called > 0 + + @property + def called_once(self): + """State that the mock was called exactly once.""" + return self._called == 1 + + def reset_stats(self): + """Reset the stats counters.""" + self._called = 0 + + +class RequestsMock(object): + """ + Mock requests HTTPAdapter. - def request(self, method, url, data=b"", **kwargs): # type: (str, str, bytes, ...) -> None + Example: + + def test(requests_mock): + requests_mock.get('https://www.test.com', text="data") + # Now you can run your tests + """ + def __init__(self, mocker): + self.mocker = mocker + self._store = {} + + def mock_send(self_mock, request, **_): + urlid = urlquick.hash_url(request) + assert urlid in self._store, "There is no mock response for given method & url" + return self_mock.build_response(request, self._store[urlid].response) + + mocker.patch.object(adapters.HTTPAdapter, "send", mock_send) + + def request(self, method, url, data=b"", **kwargs): # type: (str, str, bytes, ...) -> MockResponse req = requests.PreparedRequest() req.prepare_method(method) req.prepare_url(url, None) @@ -41,10 +79,30 @@ def request(self, method, url, data=b"", **kwargs): # type: (str, str, bytes, . req.prepare_body(data, None, None) urlid = urlquick.hash_url(req) - self._store[urlid] = self.build_urllib3_response(**kwargs) + mock_response = MockResponse(**kwargs) + self._store[urlid] = mock_response + return mock_response + + def get(self, url, **kwargs): # type: (str, ...) -> MockResponse + return self.request("GET", url, **kwargs) + + def options(self, url, **kwargs): # type: (str, ...) -> MockResponse + return self.request("OPTIONS", url, **kwargs) + + def head(self, url, **kwargs): # type: (str, ...) -> MockResponse + return self.request("HEAD", url, **kwargs) + + def post(self, url, data=None, **kwargs): # type: (...) -> MockResponse + return self.request("POST", url, data, **kwargs) + + def put(self, url, data=None, **kwargs): # type: (...) -> MockResponse + return self.request("PUT", url, data, **kwargs) + + def patch(self, url, data=None, **kwargs): # type: (...) -> MockResponse + return self.request("PATCH", url, data, **kwargs) - def get(self, url, **kwargs): # type: (str, ...) -> None - self.request("GET", url, **kwargs) + def delete(self, url, **kwargs): # type: (...) -> MockResponse + return self.request("DELETE", url, **kwargs) @pytest.fixture(scope="function") diff --git a/tests/test_misc.py b/tests/test_misc.py index 6980e81..2f41264 100644 --- a/tests/test_misc.py +++ b/tests/test_misc.py @@ -26,3 +26,13 @@ def test_hash_url(method, url, body): urlhash = urlquick.hash_url(req) assert isinstance(urlhash, str) assert len(urlhash) == 40 + + +def test_cache_cleanup(): + with pytest.deprecated_call(): + urlquick.cache_cleanup() + + +def test_auto_cache_cleanup(): + with pytest.deprecated_call(): + urlquick.auto_cache_cleanup() diff --git a/tests/test_response.py b/tests/test_response.py new file mode 100644 index 0000000..9922480 --- /dev/null +++ b/tests/test_response.py @@ -0,0 +1,46 @@ +import urlquick +import pytest +import shutil + + +@pytest.fixture(autouse=True, scope="module") +def clear_cache(): + shutil.rmtree(urlquick.CACHE_LOCATION, ignore_errors=True) + + +def test_xml(requests_mock): + xml = b""" + + Tove + Jani + + """ + requests_mock.get('https://www.test.com/cache/1', body=xml) + ret = urlquick.get('https://www.test.com/cache/1') + assert ret.content == xml + tree = ret.xml() + + assert tree.find("to").text == "Tove" + assert tree.find("from").text == "Jani" + assert tree.find("from").get("secure") == "true" + + +def test_parse(requests_mock): + html = b""" + + + Test title + + + google + + + """ + requests_mock.get('https://www.test.com/cache/2', body=html) + ret = urlquick.get('https://www.test.com/cache/2') + assert ret.content == html + tree = ret.parse() + + assert tree.find(".//title").text == "Test title" + assert tree.find(".//a").text == "google" + assert tree.find(".//a").get("href") == "https://google.ie" diff --git a/tests/test_session.py b/tests/test_session.py index a0a7efa..3e4dece 100644 --- a/tests/test_session.py +++ b/tests/test_session.py @@ -1,9 +1,277 @@ import urlquick +import requests +import shutil +import pytest +import time -def test_get(requests_mock): - requests_mock.get('https://www.test.com', text="data") - ret = urlquick.get('https://www.test.com') - print(type(ret.content), ret.content) +@pytest.mark.parametrize("obj", [urlquick, urlquick.Session()]) +class TestSessionClean(object): + """Clean the database before each and every test.""" + + # noinspection PyMethodMayBeStatic + def setup_method(self): + """Remove cache location before each test.""" + shutil.rmtree(urlquick.CACHE_LOCATION, ignore_errors=True) + + def test_get(self, obj, requests_mock): + mocked = requests_mock.get('https://www.test.com/test/586', body=b"data") + ret = obj.get('https://www.test.com/test/586') + assert mocked.called + assert ret.content == b"data" + assert ret.text == "data" + + def test_options(self, obj, requests_mock): + mocked = requests_mock.options('https://www.test.com', json={"test": True}) + ret = obj.options('https://www.test.com') + assert mocked.called + assert ret.json() == {"test": True} + + def test_head(self, obj, requests_mock): + mocked = requests_mock.head('https://www.test.com', headers={"X-TEST": "12345"}) + ret = obj.head('https://www.test.com') + assert mocked.called + assert ret.content == b"" + assert ret.text == "" + assert "X-TEST" in ret.headers and ret.headers["X-TEST"] == "12345" + + def test_post(self, obj, requests_mock): + mocked = requests_mock.post('https://www.test.com', json={"test": True}, data=b"test") + ret = obj.post('https://www.test.com', data=b"test") + assert mocked.called + assert ret.json() == {"test": True} + + def test_put(self, obj, requests_mock): + mocked = requests_mock.put('https://www.test.com', json={"test": True}) + ret = obj.put('https://www.test.com') + assert mocked.called + assert ret.json() == {"test": True} + + def test_patch(self, obj, requests_mock): + mocked = requests_mock.patch('https://www.test.com', json={"test": True}) + ret = obj.patch('https://www.test.com') + assert mocked.called + assert ret.json() == {"test": True} + + def test_delete(self, obj, requests_mock): + mocked = requests_mock.delete('https://www.test.com', json={"test": True}) + ret = obj.delete('https://www.test.com') + assert mocked.called + assert ret.json() == {"test": True} + + def test_headers_none(self, obj, requests_mock): + mocked = requests_mock.get('https://www.test.com/50', json={"test": True}) + ret = obj.get('https://www.test.com/50', headers=None) + assert mocked.called + assert ret.json() == {"test": True} + + +class TestSessionCaching(object): + """Clean the database before each and every test.""" + + # noinspection PyMethodMayBeStatic + def setup_method(self): + """Remove cache location before each test.""" + shutil.rmtree(urlquick.CACHE_LOCATION, ignore_errors=True) + + def test_cache(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data") + ret = urlquick.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + mocked.reset_stats() + + ret = urlquick.get('https://www.test.com/1') + assert not mocked.called + assert ret.content == b"data" + + def test_delay(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data") + ret = urlquick.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + mocked.reset_stats() + + time.sleep(1.2) # 1.2 seconds should be enough + ret = urlquick.get('https://www.test.com/1', max_age=1) + assert mocked.called + assert ret.content == b"data" + + def test_always_valid(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data") + ret = urlquick.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + mocked.reset_stats() + + ret = urlquick.get('https://www.test.com/1', max_age=-1) + assert not mocked.called + assert ret.content == b"data" + + def test_never_valid(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data") + ret = urlquick.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + mocked.reset_stats() + + ret = urlquick.get('https://www.test.com/1', max_age=0) + assert mocked.called + assert ret.content == b"data" + + def test_etag(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data", headers={"Etag": "12345"}) + ret = urlquick.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + mocked.reset_stats() + + ret = urlquick.get('https://www.test.com/1', max_age=0) + assert mocked.called + assert ret.content == b"data" + + def test_last_modified(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"test 304", headers={"Last-modified": "12345"}) + ret = urlquick.get('https://www.test.com/1') # Gets cached + assert mocked.called + assert ret.content == b"test 304" + + mocked = requests_mock.get('https://www.test.com/1', headers={"Last-modified": "12345"}, status=304) + ret = urlquick.get('https://www.test.com/1', max_age=0) + assert mocked.called + assert ret.content == b"test 304" + + def test_wipe(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data") + session = urlquick.Session() + + ret = session.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + mocked.reset_stats() + + # Wipe the cache clean + session.cache_adapter.wipe() + + ret = session.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + + def test_delete(self, requests_mock): + url = 'https://www.test.com/1' + mocked = requests_mock.get(url, body=b"data") + session = urlquick.Session() + + ret = session.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + mocked.reset_stats() + + # Build Request object + req = requests.PreparedRequest() + req.prepare_method("GET") + req.prepare_url(url, None) + req.prepare_headers(None) + req.prepare_body(b"", None, None) + + # Test del_cache + urlhash = urlquick.hash_url(req) + session.cache_adapter.del_cache(urlhash) + + ret = session.get('https://www.test.com/1') + assert mocked.called + assert ret.content == b"data" + + +class TestRaiseForStatus(object): + """Clean the database before each and every test.""" + + # noinspection PyMethodMayBeStatic + def setup_method(self): + """Remove cache location before each test.""" + shutil.rmtree(urlquick.CACHE_LOCATION, ignore_errors=True) + + def test_false_normal(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data", status=200) + session = urlquick.Session() + + ret = session.get('https://www.test.com/1') + assert mocked.called + assert ret.status_code == 200 + assert ret.content == b"data" + + def test_false_error(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data", status=404) + session = urlquick.Session() + + ret = session.get('https://www.test.com/1') + assert mocked.called + assert ret.status_code == 404 + assert ret.content == b"data" + + def test_true_normal(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data", status=200) + session = urlquick.Session(raise_for_status=True) + + ret = session.get('https://www.test.com/1') + assert mocked.called + assert ret.status_code == 200 + assert ret.content == b"data" + + def test_true_error(self, requests_mock): + mocked = requests_mock.get('https://www.test.com/1', body=b"data", status=404) + session = urlquick.Session(raise_for_status=True) + + with pytest.raises(urlquick.HTTPError): + session.get('https://www.test.com/1') + + assert mocked.called + + +def test_session_send(requests_mock): + shutil.rmtree(urlquick.CACHE_LOCATION, ignore_errors=True) + + url = 'https://www.test.com/1' + mocked = requests_mock.get(url, body=b"data") + session = urlquick.Session() + + # Build Request object + req = requests.PreparedRequest() + req.prepare_method("GET") + req.prepare_url(url, None) + req.prepare_headers(None) + req.prepare_body(b"", None, None) + + ret = session.send(req) + assert mocked.called + assert ret.content == b"data" + + +def test_request_header_none(requests_mock): + shutil.rmtree(urlquick.CACHE_LOCATION, ignore_errors=True) + mocked = requests_mock.get('https://www.test.com/test/542', body=b"data") + session = urlquick.Session() + + ret = session.request("GET", 'https://www.test.com/test/542', None, None, None) + assert mocked.called + assert ret.content == b"data" + + +def test_request_header_data(requests_mock): + shutil.rmtree(urlquick.CACHE_LOCATION, ignore_errors=True) + mocked = requests_mock.get('https://www.test.com/test/542', body=b"data") + session = urlquick.Session() + + ret = session.request("GET", 'https://www.test.com/test/542', None, None, {"X-TEST": "test"}) + assert mocked.called + assert ret.content == b"data" + + +def test_session_method(requests_mock): + shutil.rmtree(urlquick.CACHE_LOCATION, ignore_errors=True) + mocked = requests_mock.get('https://www.test.com', body=b"data") + session = urlquick.session() + ret = session.get('https://www.test.com') + assert mocked.called assert ret.content == b"data" assert ret.text == "data" diff --git a/urlquick.py b/urlquick.py index 3bcfb23..335b840 100644 --- a/urlquick.py +++ b/urlquick.py @@ -59,10 +59,10 @@ import pickle # Works for both python 2 & 3 # Third Party +from htmlement import HTMLement from requests.structures import CaseInsensitiveDict from requests.adapters import HTTPResponse from requests import adapters -from requests import sessions from requests import * import requests @@ -147,10 +147,6 @@ class CacheError(RequestException): pass -class MissingDependency(ImportError): - """Missing optional Dependency""" - - class Response(requests.Response): def xml(self): """ @@ -177,19 +173,11 @@ def parse(self, tag=u"", attrs=None): :return: The root element of the element tree. :rtype: xml.etree.ElementTree.Element - - :raise MissingDependency: If the optional 'HTMLement' dependency is missing. """ - try: - # noinspection PyUnresolvedReferences - from htmlement import HTMLement - except ImportError: - raise MissingDependency("Missing optional dependency: 'HTMLement'") - else: - tag = tag.decode() if isinstance(tag, bytes) else tag - parser = HTMLement(tag, attrs) - parser.feed(self.text) - return parser.close() + tag = tag.decode() if isinstance(tag, bytes) else tag + parser = HTMLement(tag, attrs) + parser.feed(self.text) + return parser.close() @classmethod def extend_response(cls, response): @@ -199,7 +187,7 @@ def extend_response(cls, response): def __conform__(self, protocol): """Convert Response to a sql blob.""" - if protocol is sqlite3.PrepareProtocol: + if protocol is sqlite3.PrepareProtocol: # pragma: no branch data = pickle.dumps(self, protocol=pickle.HIGHEST_PROTOCOL) return sqlite3.Binary(data) @@ -380,7 +368,6 @@ def send(self, request, **kwargs): # type: (PreparedRequest, ...) -> Response def build_response(self, req, resp): # type: (PreparedRequest, HTTPResponse) -> Response """Replace response object with our customized version.""" resp = super(CacheHTTPAdapter, self).build_response(req, resp) - print(type(resp.content)) return Response.extend_response(resp) def process_response(self, response, cache, urlhash): # type: (Response, CacheRecord, str) -> Response @@ -413,7 +400,7 @@ def __init__(self, cache_location=CACHE_LOCATION, **kwargs): # type: (str, ...) #: Defaults to :data:`MAX_AGE ` self.max_age = kwargs.get("max_age", MAX_AGE) - adapter = CacheHTTPAdapter(cache_location) + self.cache_adapter = adapter = CacheHTTPAdapter(cache_location) self.mount("https://", adapter) self.mount("http://", adapter) @@ -431,6 +418,7 @@ def request(self, *args, **kwargs): # type: (...) -> Response # So we need to keep this in mind if len(args) >= 5: headers = args[4] or {} + args = list(args) args[4] = headers else: headers = kwargs.get("headers") or {}