Spaces:
Runtime error
Runtime error
File size: 3,476 Bytes
63deadc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 |
import os
from warnings import catch_warnings
import pytest
import fsspec
from fsspec.config import conf, set_conf_env, set_conf_files
@pytest.fixture
def clean_conf():
"""Tests should start and end with clean config dict"""
conf.clear()
yield
conf.clear()
def test_from_env_ignored(clean_conf):
env = {
"FSSPEC": "missing_protocol",
"FSSPEC_": "missing_protocol",
"FSSPEC__INVALID_KEY": "invalid_protocol",
"FSSPEC_INVALID1": "not_json_dict",
"FSSPEC_INVALID2": '["not_json_dict"]',
}
cd = {}
with catch_warnings(record=True) as w:
set_conf_env(conf_dict=cd, envdict=env)
assert len(w) == 5
assert "unexpected name" in str(w[0].message)
assert "unexpected name" in str(w[1].message)
assert "unexpected name" in str(w[2].message)
assert "parse failure" in str(w[3].message)
assert "not being a dict" in str(w[4].message)
assert cd == {}
def test_from_env_kwargs(clean_conf):
env = {
"FSSPEC_PROTO_KEY": "value",
"FSSPEC_PROTO_LONG_KEY": "othervalue",
"FSSPEC_MALFORMED": "novalue",
}
cd = {}
with catch_warnings(record=True) as w:
set_conf_env(conf_dict=cd, envdict=env)
assert len(w) == 1
assert "parse failure" in str(w[0].message)
assert cd == {"proto": {"key": "value", "long_key": "othervalue"}}
def test_from_env_protocol_dict(clean_conf):
env = {
"FSSPEC_PROTO": '{"int": 1, "float": 2.3, "bool": true, "dict": {"key": "val"}}'
}
cd = {}
set_conf_env(conf_dict=cd, envdict=env)
assert cd == {
"proto": {"int": 1, "float": 2.3, "bool": True, "dict": {"key": "val"}}
}
def test_from_env_kwargs_override_protocol_dict(clean_conf):
env = {
"FSSPEC_PROTO_LONG_KEY": "override1",
"FSSPEC_PROTO": '{"key": "value1", "long_key": "value2", "otherkey": "value3"}',
"FSSPEC_PROTO_KEY": "override2",
}
cd = {}
set_conf_env(conf_dict=cd, envdict=env)
assert cd == {
"proto": {"key": "override2", "long_key": "override1", "otherkey": "value3"}
}
def test_from_file_ini(clean_conf, tmpdir):
file1 = os.path.join(tmpdir, "1.ini")
file2 = os.path.join(tmpdir, "2.ini")
with open(file1, "w") as f:
f.write(
"""[proto]
key=value
other_key:othervalue
overwritten=dont_see
"""
)
with open(file2, "w") as f:
f.write(
"""[proto]
overwritten=see
"""
)
cd = {}
set_conf_files(tmpdir, cd)
assert cd == {
"proto": {"key": "value", "other_key": "othervalue", "overwritten": "see"}
}
def test_from_file_json(clean_conf, tmpdir):
file1 = os.path.join(tmpdir, "1.json")
file2 = os.path.join(tmpdir, "2.json")
with open(file1, "w") as f:
f.write(
"""{"proto":
{"key": "value",
"other_key": "othervalue",
"overwritten": false}}
"""
)
with open(file2, "w") as f:
f.write(
"""{"proto":
{"overwritten": true}}
"""
)
cd = {}
set_conf_files(tmpdir, cd)
assert cd == {
"proto": {"key": "value", "other_key": "othervalue", "overwritten": True}
}
def test_apply(clean_conf):
conf["file"] = {"auto_mkdir": "test"}
fs = fsspec.filesystem("file")
assert fs.auto_mkdir == "test"
fs = fsspec.filesystem("file", auto_mkdir=True)
assert fs.auto_mkdir is True
|