-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
test_cache.py
48 lines (40 loc) 路 1.39 KB
/
test_cache.py
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
import pickle
from typing import Any
import pytest
from dvc.objects.cache import Cache, DiskError
def set_value(cache: Cache, key: str, value: Any) -> Any:
cache[key] = value
return cache[key]
@pytest.mark.parametrize("disk_type", [None, "test"])
def test_pickle_protocol_error(tmp_dir, disk_type):
directory = tmp_dir / "test"
cache = Cache(
directory,
disk_pickle_protocol=pickle.HIGHEST_PROTOCOL + 1,
type=disk_type,
)
with pytest.raises(DiskError) as exc, cache as cache:
set_value(cache, "key", ("value1", "value2"))
assert exc.value.directory == str(directory)
assert exc.value.type == "test"
assert f"Could not open disk 'test' in {directory}" == str(exc.value)
@pytest.mark.parametrize(
"proto_a, proto_b",
[
(pickle.HIGHEST_PROTOCOL - 1, pickle.HIGHEST_PROTOCOL),
(pickle.HIGHEST_PROTOCOL, pickle.HIGHEST_PROTOCOL - 1),
],
)
def test_pickle_backwards_compat(tmp_dir, proto_a, proto_b):
with Cache(
directory=(tmp_dir / "test"),
disk_pickle_protocol=proto_a,
) as cache:
set_value(cache, "key", ("value1", "value2"))
with Cache(
directory=(tmp_dir / "test"),
disk_pickle_protocol=proto_b,
) as cache:
assert ("value1", "value2") == cache["key"]
set_value(cache, "key", ("value3", "value4"))
assert ("value3", "value4") == cache["key"]