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
131
132
133
134
135
136
137
138
139
|
#!/usr/bin/env python3
'''
t5_cp.py - this file is part of S3QL.
Copyright © 2008 Nikolaus Rath <Nikolaus@rath.org>
This work can be distributed under the terms of the GNU GPLv3.
'''
if __name__ == '__main__':
import sys
import pytest
sys.exit(pytest.main([__file__] + sys.argv[1:]))
import os.path
import shutil
import subprocess
import sys
import tempfile
from subprocess import CalledProcessError, check_output
import pytest
import t4_fuse
from common import populate_dir, retry, skip_without_rsync
class TestCp(t4_fuse.TestFuse):
def test(self):
skip_without_rsync()
self.mkfs()
self.mount()
self.tst_cp()
self.umount()
self.fsck()
def test_bg(self):
skip_without_rsync()
self.mkfs()
self.logfile = tempfile.NamedTemporaryFile()
self.mount(in_foreground=False, extra_args=["--log", self.logfile.name])
self.tst_cp()
self.umount()
self.fsck()
self.logfile.close()
def tst_cp(self):
tempdir = tempfile.mkdtemp(prefix='s3ql-cp-')
try:
populate_dir(tempdir)
# Rsync
subprocess.check_call(
['rsync', '-aHAX', tempdir + '/', os.path.join(self.mnt_dir, 'orig') + '/']
)
# copy
subprocess.check_call(
self.s3ql_cmd_argv('s3qlcp')
+ [
'--quiet',
os.path.join(self.mnt_dir, 'orig'),
os.path.join(self.mnt_dir, 'copy'),
]
)
# compare
try:
out = check_output(
[
'rsync',
'-anciHAX',
'--delete',
tempdir + '/',
os.path.join(self.mnt_dir, 'copy') + '/',
],
universal_newlines=True,
stderr=subprocess.STDOUT,
)
except CalledProcessError as exc:
pytest.fail('rsync failed with ' + exc.output)
if out:
pytest.fail('Copy not equal to original, rsync says:\n' + out)
finally:
shutil.rmtree(tempdir)
def test_cp_inode_invalidate(self):
# check if we can write to drop_caches
try:
open("/proc/sys/vm/drop_caches", "w").close()
except OSError:
pytest.skip('test_cp_inode_invalidate requires drop_caches to be writable, skipping.')
self.passphrase = None
self.mkfs()
# Run monkeypatched mount.s3ql with overridden pyfuse3.invalidate_inode :
# Drop kernel dentries and inodes cache just before calling pyfuse3.invalidate_inode
cmd = [
sys.executable,
os.path.join(os.path.dirname(__file__), 'mount_helper.py'),
"--fg",
'--cachedir',
self.cache_dir,
'--log',
'none',
'--compress',
'zlib',
'--quiet',
self.storage_url,
self.mnt_dir,
'--authfile',
'/dev/null',
]
self.mount_process = subprocess.Popen(cmd, universal_newlines=True)
def poll():
if os.path.ismount(self.mnt_dir):
return True
assert self.mount_process.poll() is None
retry(10, poll)
os.mkdir(os.path.join(self.mnt_dir, 'orig'))
cmd = self.s3ql_cmd_argv('s3qlcp') + [
'--quiet',
os.path.join(self.mnt_dir, 'orig'),
os.path.join(self.mnt_dir, 'copy'),
]
cp_process = subprocess.Popen(cmd)
retry(5, lambda: cp_process.poll() is not None)
assert cp_process.wait() == 0
self.umount()
self.fsck()
|
