Files
dvc/tests/unit/remote/test_http.py
T
Batuhan Taskaya 9aec8f1926 dvc: get rid of path_info
Intermediate PR in removing path_info from our code base.

It is completely removed from the main codebase and is still
being used in tests, which will be migrated in the followups.

For fs-specific simple path manipulation, we've introduced
`fs.path` helper (similar to `os.path`).

Fixes #6774
2021-11-10 00:52:48 +02:00

104 lines
2.3 KiB
Python

import ssl
import pytest
from mock import patch
from dvc.fs.http import HTTPFileSystem
def test_download_fails_on_error_code(tmp_dir, http):
fs = HTTPFileSystem(**http.config)
with pytest.raises(FileNotFoundError):
fs.download_file(
(http / "missing.txt").fs_path, (tmp_dir / "missing.txt").fs_path
)
def test_public_auth_method(dvc):
config = {
"url": "http://example.com/",
"path": "file.html",
"user": "",
"password": "",
}
fs = HTTPFileSystem(**config)
assert "auth" not in fs.fs_args["client_kwargs"]
assert "headers" not in fs.fs_args
def test_basic_auth_method(dvc):
user = "username"
password = "password"
config = {
"url": "http://example.com/",
"path": "file.html",
"auth": "basic",
"user": user,
"password": password,
}
fs = HTTPFileSystem(**config)
assert fs.fs_args["client_kwargs"]["auth"].login == user
assert fs.fs_args["client_kwargs"]["auth"].password == password
def test_custom_auth_method(dvc):
header = "Custom-Header"
password = "password"
config = {
"url": "http://example.com/",
"path": "file.html",
"auth": "custom",
"custom_auth_header": header,
"password": password,
}
fs = HTTPFileSystem(**config)
headers = fs.fs_args["headers"]
assert header in headers
assert headers[header] == password
def test_ssl_verify_disable(dvc):
config = {
"url": "http://example.com/",
"path": "file.html",
"ssl_verify": False,
}
fs = HTTPFileSystem(**config)
assert not fs.fs_args["client_kwargs"]["connector"]._ssl
@patch("ssl.SSLContext.load_verify_locations")
def test_ssl_verify_custom_cert(dvc, mocker):
config = {
"url": "http://example.com/",
"path": "file.html",
"ssl_verify": "/path/to/custom/cabundle.pem",
}
fs = HTTPFileSystem(**config)
assert isinstance(
fs.fs_args["client_kwargs"]["connector"]._ssl, ssl.SSLContext
)
def test_http_method(dvc):
config = {
"url": "http://example.com/",
"path": "file.html",
}
fs = HTTPFileSystem(**config, method="PUT")
assert fs.upload_method == "PUT"
fs = HTTPFileSystem(**config, method="POST")
assert fs.upload_method == "POST"