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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
import os
import re
import shutil
import tarfile
import tempfile
import time
import uuid
from apt_listchanges.ALCConfig import ALCConfig
from apt_listchanges.snapshot import Snapshot
def make_config(capture=None, sdir=None) -> ALCConfig:
config = ALCConfig()
if capture is not None:
config.capture_snapshots = capture
if sdir is not None:
config.snapshot_dir = sdir
return config
def test_not_capturing():
snapshot = Snapshot(make_config())
assert not snapshot.capturing
# Exercise all the `if not self.capturing` blocks
snapshot.add_file('/dev/null')
snapshot.add_data('foo', 'bar')
snapshot.commit()
def test_no_snapshot_dir():
snapshot = Snapshot(make_config(capture=True))
assert not snapshot.capturing
def test_nested_missing_dir():
with tempfile.TemporaryDirectory() as tempdir:
snapshot_dir = os.path.join(tempdir, 'foo', 'bar')
snapshot = Snapshot(make_config(capture=True, sdir=snapshot_dir))
assert not snapshot.capturing
def test_snapshot_dir_created():
with tempfile.TemporaryDirectory() as tempdir:
snapshot_dir = os.path.join(tempdir, 'foo')
snapshot = Snapshot(make_config(capture=True, sdir=snapshot_dir))
assert snapshot.capturing
assert snapshot_dir == os.path.dirname(snapshot.snapshot_path)
assert re.match(r'\d\d\d\d-\d\d-\d\dT',
os.path.basename(snapshot.snapshot_path))
def test_snapshot_dir_creation_error():
with tempfile.TemporaryDirectory() as tempdir:
dir1 = os.path.join(tempdir, 'foo')
os.mkdir(dir1, mode=0o555)
snapshot_dir = os.path.join(dir1, 'bar')
snapshot = Snapshot(make_config(capture=True, sdir=snapshot_dir))
assert not snapshot.capturing
def test_snapshot_subdir_creation_error():
with tempfile.TemporaryDirectory() as tempdir:
snapshot_dir = os.path.join(tempdir, 'foo')
# Create a file where the directory should be
# pylint: disable=consider-using-with
open(snapshot_dir, 'wb').close()
snapshot = Snapshot(make_config(capture=True, sdir=snapshot_dir))
assert not snapshot.capturing
def test_context_manager():
# Also tests add_data
with tempfile.TemporaryDirectory() as tempdir:
snapshot_dir = os.path.join(tempdir, 'foo')
with Snapshot(make_config(capture=True,
sdir=snapshot_dir)) as snapshot:
snapshot.add_data('foo', 'bar')
assert not snapshot.capturing
assert snapshot.archive is not None
with tarfile.open(snapshot.archive) as archive:
path = os.path.join(snapshot.snapshot_subdir, 'bar')
assert archive.extractfile(path).read() == b'foo'
def test_add_file():
with tempfile.TemporaryDirectory() as tempdir:
snapshot_dir = os.path.join(tempdir, 'foo')
data_file = os.path.join(tempdir, 'data')
data_contents = uuid.uuid1().bytes
with open(data_file, 'wb') as f:
f.write(data_contents)
snapshot = Snapshot(make_config(capture=True, sdir=snapshot_dir))
assert snapshot.capturing
snapshot.add_file(data_file)
snapshot.add_file(data_file, 'data2')
snapshot.commit()
assert not snapshot.capturing
assert snapshot.archive is not None
with tarfile.open(snapshot.archive) as archive:
path = os.path.join(snapshot.snapshot_subdir, 'data')
assert archive.extractfile(path).read() == data_contents
path = os.path.join(snapshot.snapshot_subdir, 'data2')
assert archive.extractfile(path).read() == data_contents
def test_add_unreadable_file():
with tempfile.TemporaryDirectory() as tempdir:
snapshot_dir = os.path.join(tempdir, 'foo')
data_file = os.path.join(tempdir, 'data')
# pylint: disable=consider-using-with
open(data_file, 'wb').close()
os.chmod(data_file, 0o000)
snapshot = Snapshot(make_config(capture=True, sdir=snapshot_dir))
assert snapshot.capturing
snapshot.add_file(data_file)
assert not snapshot.capturing
def test_add_missing_file():
with tempfile.TemporaryDirectory() as tempdir:
snapshot_dir = os.path.join(tempdir, 'foo')
data_file = os.path.join(tempdir, 'data')
snapshot = Snapshot(make_config(capture=True, sdir=snapshot_dir))
assert snapshot.capturing
snapshot.add_file(data_file)
assert snapshot.capturing
snapshot.commit()
with tarfile.open(snapshot.archive) as archive:
path = os.path.join(snapshot.snapshot_subdir, 'data.missing')
assert archive.extractfile(path).read() == b''
def test_add_binary_data():
with tempfile.TemporaryDirectory() as tempdir:
snapshot_dir = os.path.join(tempdir, 'foo')
data_contents = uuid.uuid1().bytes
snapshot = Snapshot(make_config(capture=True, sdir=snapshot_dir))
assert snapshot.capturing
snapshot.add_data(data_contents, 'data_in_archive')
snapshot.commit()
assert not snapshot.capturing
assert snapshot.archive is not None
with tarfile.open(snapshot.archive) as archive:
path = os.path.join(snapshot.snapshot_subdir, 'data_in_archive')
assert archive.extractfile(path).read() == data_contents
def test_prune():
with tempfile.TemporaryDirectory() as tempdir:
for i in range(2):
snapshot = Snapshot(make_config(capture=True, sdir=tempdir))
snapshot.SAVED_SNAPSHOTS = 1
snapshot.commit()
# Snapshot file names have one-second granularity
time.sleep(1)
leftovers = list(os.scandir(tempdir))
assert len(leftovers) == 1
def test_abort():
with tempfile.TemporaryDirectory() as tempdir:
assert len(list(os.scandir(tempdir))) == 0
snapshot = Snapshot(make_config(capture=True, sdir=tempdir))
assert snapshot.capturing
assert len(list(os.scandir(tempdir))) == 1
snapshot.abort()
assert not snapshot.capturing
assert len(list(os.scandir(tempdir))) == 0
def test_make_archive_failure():
with tempfile.TemporaryDirectory() as tempdir:
assert len(list(os.scandir(tempdir))) == 0
snapshot = Snapshot(make_config(capture=True, sdir=tempdir))
shutil.rmtree(snapshot.snapshot_path)
snapshot.commit()
|
