Mirrors/aqtinstall
1
0
Fork 0
mirror of https://github.com/miurahr/aqtinstall.git synced 2025-12-18 13:14:37 +03:00
Code Issues Actions Packages Projects Releases Wiki Activity

Test: helper: update mock for requests.sessions.Session

Browse Source 
Signed-off-by: Hiroshi Miura <miurahr@linux.com>
This commit is contained in:
Hiroshi Miura
2022-01-11 20:36:34 +09:00
parent 87f95ddc59
commit 948fb18650
2 changed files with 15 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
aqt/helper.py
View File

@@ -49,7 +49,7 @@ def _check_content_type(ct: str) -> bool:
def getUrl(url: str, timeout) -> str: def getUrl(url: str, timeout) -> str:
logger = getLogger("aqt.helper") logger = getLogger("aqt.helper")
with requests.Session() as session: with requests.sessions.Session() as session:
retries = requests.adapters.Retry( retries = requests.adapters.Retry(
total=Settings.max_retries_on_connection_error, backoff_factor=Settings.backoff_factor total=Settings.max_retries_on_connection_error, backoff_factor=Settings.backoff_factor
) )
@@ -82,7 +82,7 @@ def getUrl(url: str, timeout) -> str:
def downloadBinaryFile(url: str, out: Path, hash_algo: str, exp: bytes, timeout): def downloadBinaryFile(url: str, out: Path, hash_algo: str, exp: bytes, timeout):
logger = getLogger("aqt.helper") logger = getLogger("aqt.helper")
filename = Path(url).name filename = Path(url).name
with requests.Session() as session: with requests.sessions.Session() as session:
retries = requests.adapters.Retry( retries = requests.adapters.Retry(
total=Settings.max_retries_on_connection_error, backoff_factor=Settings.backoff_factor total=Settings.max_retries_on_connection_error, backoff_factor=Settings.backoff_factor
) )

25
tests/test_helper.py
View File

@@ -234,28 +234,25 @@ def mocked_request_response_class(num_redirects: int = 0, forbidden_baseurls=Non
return MockResponse return MockResponse
def test_helper_getUrl_ok(monkeypatch):
response_class = mocked_request_response_class()
def _mock_get(url, **kwargs):
return response_class(url, {}, "some_html_content")
monkeypatch.setattr(requests, "get", _mock_get)
assert getUrl("some_url", timeout=(5, 5)) == "some_html_content"
def mock_get_redirect(num_redirects: int): def mock_get_redirect(num_redirects: int):
response_class = mocked_request_response_class(num_redirects) response_class = mocked_request_response_class(num_redirects)
def _mock(url: str, timeout, allow_redirects): def _mock(url: str, timeout=None, allow_redirects=None):
return response_class(url, {}, text="some_html_content") return response_class(url, {}, text="some_html_content")
def _mock_session(self, url: str, timeout, stream): def _mock_session(self, url: str, allow_redirects=None, timeout=None, stream=None):
return response_class(url, {}, text="some_html_content") return response_class(url, {}, text="some_html_content")
return _mock, _mock_session return _mock, _mock_session
def test_helper_getUrl_ok(monkeypatch):
mocked_get, mocked_session_get = mock_get_redirect(0)
monkeypatch.setattr(requests, "get", mocked_get)
monkeypatch.setattr(requests.Session, "get", mocked_session_get)
assert getUrl("some_url", timeout=(5, 5)) == "some_html_content"
def test_helper_getUrl_redirect_5(monkeypatch): def test_helper_getUrl_redirect_5(monkeypatch):
mocked_get, mocked_session_get = mock_get_redirect(num_redirects=5) mocked_get, mocked_session_get = mock_get_redirect(num_redirects=5)
monkeypatch.setattr(requests, "get", mocked_get) monkeypatch.setattr(requests, "get", mocked_get)
@@ -282,7 +279,11 @@ def test_helper_getUrl_conn_error(monkeypatch):
def _mock(url: str, *args, **kwargs): def _mock(url: str, *args, **kwargs):
return response_class(url, {}, text="some_html_content") return response_class(url, {}, text="some_html_content")
def _mock_session(self, url: str, *args, **kargs):
return response_class(url, {}, text="some_html_content")
monkeypatch.setattr(requests, "get", _mock) monkeypatch.setattr(requests, "get", _mock)
monkeypatch.setattr(requests.sessions.Session, "get", _mock_session)
with pytest.raises(ArchiveConnectionError) as e: with pytest.raises(ArchiveConnectionError) as e:
getUrl(url, timeout) getUrl(url, timeout)
assert e.type == ArchiveConnectionError assert e.type == ArchiveConnectionError