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
|
# Copyright: 2006-2009 Brian Harring <ferringb@gmail.com>
# License: GPL2/BSD
from functools import partial
import os
from snakeoil import data_source
from snakeoil.chksum import get_handlers
import pytest
from pkgcore.fetch import base, fetchable, errors
repeating_str = 'asdf'
data = repeating_str * 4000
handlers = get_handlers()
from snakeoil.mappings import LazyValDict
def _callback(chf):
return handlers[chf](data_source.data_source(data))
chksums = LazyValDict(frozenset(handlers.keys()), _callback)
# get a non size based chksum
known_chksum = [x for x in handlers.keys() if x != "size"][0]
class TestFetcher(object):
@pytest.fixture(autouse=True)
def _setup(self, tmpdir):
self.fp = os.path.join(str(tmpdir), "test")
self.obj = fetchable(self.fp, chksums=chksums)
self.fetcher = base.fetcher()
def write_data(self, data=data):
with open(self.fp, "w") as f:
f.write(data)
def test__call__(self):
l = []
class c(base.fetcher):
def fetch(self, *a, **kw):
l.extend((a, kw))
o = c()
o.fetch(1, foon=True)
assert [(1,), {"foon": True}] == l
self.write_data()
assert self.fetcher._verify(self.fp, self.obj) == None
self.write_data("asdf")
with pytest.raises(errors.FetchError) as excinfo:
self.fetcher._verify(self.fp, self.obj)
assert excinfo.value.resumable
def test_verify_all_chksums(self):
self.write_data()
subhandlers = dict([list(handlers.items())[0]])
with pytest.raises(errors.RequiredChksumDataMissing):
self.fetcher._verify(self.fp, self.obj, handlers=subhandlers)
self.fetcher._verify(self.fp, self.obj)
assert None == self.fetcher._verify(
self.fp, self.obj, handlers=subhandlers, all_chksums=False)
def test_size_verification_first(self):
self.write_data()
chksum_data = dict(chksums.items())
l = []
def f(chf, fp):
l.append(chf)
return chksum_data[chf]
subhandlers = {"size": partial(f, 'size'), known_chksum:partial(f, known_chksum)}
# exact size verification
self.fetcher._verify(self.fp, self.obj, handlers=subhandlers, all_chksums=False)
assert ['size', known_chksum] == l
for x in (-100, 100):
while l:
l.pop(-1)
chksum_data["size"] = chksums["size"] + x
if x > 0:
with pytest.raises(errors.ChksumFailure) as excinfo:
self.fetcher._verify(
self.fp, self.obj, handlers=subhandlers, all_chksums=False)
assert excinfo.value.chksum == 'size'
else:
with pytest.raises(errors.FetchError) as excinfo:
self.fetcher._verify(
self.fp, self.obj, handlers=subhandlers, all_chksums=False)
assert excinfo.value.resumable
assert ['size'] == l
def test_normal(self):
self.write_data()
assert self.fetcher._verify(self.fp, self.obj) == None
self.write_data(data[:-1])
with pytest.raises(errors.FetchError) as excinfo:
self.fetcher._verify(self.fp, self.obj)
assert excinfo.value.resumable
# verify it returns -2 for missing file paths.
os.unlink(self.fp)
with pytest.raises(errors.MissingDistfile) as excinfo:
self.fetcher._verify(self.fp, self.obj)
assert excinfo.value.resumable
self.write_data(data + "foon")
with pytest.raises(errors.ChksumFailure) as excinfo:
self.fetcher._verify(self.fp, self.obj)
assert excinfo.value.chksum == 'size'
# verify they're ran one, and only once
l = []
def f(chf, fp):
l.append(chf)
return chksums[chf]
alt_handlers = {chf: partial(f, chf) for chf in chksums}
assert None == self.fetcher._verify(self.fp, self.obj, handlers=alt_handlers)
assert sorted(l) == sorted(alt_handlers)
|
GitWeb