mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-05 23:09:32 +01:00
dependabot[bot]andPeter Rowlands
6c14bf39dc
* build(deps): bump pylint from 2.8.3 to 2.9.6 Bumps [pylint](https://github.com/PyCQA/pylint) from 2.8.3 to 2.9.6. - [Release notes](https://github.com/PyCQA/pylint/releases) - [Changelog](https://github.com/PyCQA/pylint/blob/main/ChangeLog) - [Commits](https://github.com/PyCQA/pylint/compare/v2.8.3...v2.9.6) --- updated-dependencies: - dependency-name: pylint dependency-type: direct:production update-type: version-update:semver-minor ... Signed-off-by: dependabot[bot] <[email protected]> * pylint: fix 2.9.6 issues Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Peter Rowlands <[email protected]>
57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
import io
|
|
|
|
import requests
|
|
|
|
from dvc.utils.http import open_url
|
|
|
|
|
|
def read_nbytes(fd, num, mode="r"):
|
|
data = b"" if mode == "rb" else ""
|
|
while len(data) < num:
|
|
chunk = fd.read(num - len(data))
|
|
if not chunk:
|
|
break
|
|
data += chunk
|
|
return data
|
|
|
|
|
|
def test_open_url(tmp_path, monkeypatch, http):
|
|
# Simulate bad connection
|
|
original_iter_content = requests.Response.iter_content
|
|
|
|
def bad_iter_content(self, *args, **kwargs):
|
|
it = original_iter_content(self, *args, **kwargs)
|
|
for i, chunk in enumerate(it):
|
|
# Drop connection error on second chunk if there is one
|
|
if i > 0:
|
|
raise requests.ConnectionError("Simulated connection drop")
|
|
yield chunk
|
|
|
|
monkeypatch.setattr(requests.Response, "iter_content", bad_iter_content)
|
|
|
|
# Text should be longer than default chunk to test resume,
|
|
# using twice of that plus something tests second resume,
|
|
# this is important because second response is different
|
|
text = "0123456789" * (io.DEFAULT_BUFFER_SIZE // 10 + 1)
|
|
http.gen("sample.txt", text * 2)
|
|
|
|
with open_url((http / "sample.txt").url) as fd:
|
|
# Test various .read() variants
|
|
assert read_nbytes(fd, len(text), mode="r") == text
|
|
assert fd.read() == text
|
|
assert fd.read() == ""
|
|
|
|
|
|
def test_open_url_peek_rb(tmp_path, monkeypatch, http):
|
|
# Goes over seek feature in 'rb' mode
|
|
text = "0123456789" * (io.DEFAULT_BUFFER_SIZE // 10 + 1)
|
|
http.gen("sample.txt", text * 2)
|
|
|
|
with open_url((http / "sample.txt").url, mode="rb") as fd:
|
|
text = text.encode("utf8")
|
|
assert fd.peek(len(text)) == text # pylint: disable=no-member
|
|
assert read_nbytes(fd, len(text), mode="rb") == text
|
|
assert fd.peek(len(text)) == text # pylint: disable=no-member
|
|
assert fd.read() == text
|
|
assert fd.read() == b""
|