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
|
from XRootD import client
from XRootD.client.flags import OpenFlags
from env import *
def test_copy_smallfile():
f = client.File()
s, r = f.open(smallfile, OpenFlags.DELETE)
assert s.ok
f.write(smallbuffer)
size1 = f.stat(force=True)[1].size
s, r = f.close()
assert s.ok
c = client.CopyProcess()
c.add_job(source=smallfile, target=smallcopy, force=True)
s = c.prepare()
assert s.ok
s, __ = c.run()
assert s.ok
f = client.File()
s, r = f.open(smallcopy, OpenFlags.READ)
size2 = f.stat()[1].size
assert size1 == size2
f.close()
def test_create_bigfile():
f = client.File()
s, r = f.open(bigfile, OpenFlags.DELETE)
assert s.ok
for i in range(1000):
f.write(smallbuffer)
size1 = f.stat(force=True)[1].size
s, r = f.close()
assert s.ok
def test_copy_bigfile():
f = client.File()
s, r = f.open(bigfile)
assert s.ok
size1 = f.stat(force=True)[1].size
f.close()
c = client.CopyProcess()
c.add_job(source=bigfile, target=bigcopy, force=True)
s = c.prepare()
assert s.ok
s, __ = c.run()
assert s.ok
f = client.File()
s, r = f.open(bigcopy, OpenFlags.READ)
size2 = f.stat()[1].size
assert size1 == size2
f.close()
def test_copy_nojobs():
c = client.CopyProcess()
s = c.prepare()
assert s.ok
s, __ = c.run()
assert s.ok
def test_copy_noprep():
c = client.CopyProcess()
c.add_job(source=bigfile, target=bigcopy, force=True)
s, __ = c.run()
assert s.ok
class TestProgressHandler(object):
def begin(self, id, total, source, target):
print('+++ begin(): %d, total: %d' % (id, total))
print('+++ source: %s' % source)
print('+++ target: %s' % target)
def end(self, jobId, status):
print('+++ end(): jobId: %s, status: %s' % (jobId, status))
def update(self, jobId, processed, total):
print('+++ update(): jobid: %s, processed: %d, total: %d' % (jobId, processed, total))
def should_cancel(self, jobId):
print('+++ should_cancel(): jobid: %s' % (jobId))
return False
def test_copy_progress_handler():
c = client.CopyProcess()
c.add_job( source=bigfile, target=bigcopy, force=True )
s = c.prepare()
assert s.ok
h = TestProgressHandler()
s, _ = c.run(handler=h)
assert s.ok
|
