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
|
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
f.close()
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_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 test_copy_progress_handler():
c = client.CopyProcess()
c.add_job( source=bigfile, target=bigcopy, force=True )
c.prepare()
h = TestProgressHandler()
c.run(handler=h)
|
