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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
|
# This file is part of cloud-init. See LICENSE file for license information.
import os
from collections import namedtuple
from io import StringIO
from cloudinit.cmd import clean
from cloudinit.util import ensure_dir, sym_link, write_file
from tests.unittests.helpers import CiTestCase, mock, wrap_and_call
mypaths = namedtuple("MyPaths", "cloud_dir")
class TestClean(CiTestCase):
def setUp(self):
super(TestClean, self).setUp()
self.new_root = self.tmp_dir()
self.artifact_dir = self.tmp_path("artifacts", self.new_root)
self.log1 = self.tmp_path("cloud-init.log", self.new_root)
self.log2 = self.tmp_path("cloud-init-output.log", self.new_root)
class FakeInit(object):
cfg = {
"def_log_file": self.log1,
"output": {"all": "|tee -a {0}".format(self.log2)},
}
# Ensure cloud_dir has a trailing slash, to match real behaviour
paths = mypaths(cloud_dir="{}/".format(self.artifact_dir))
def __init__(self, ds_deps):
pass
def read_cfg(self):
pass
self.init_class = FakeInit
def test_remove_artifacts_removes_logs(self):
"""remove_artifacts removes logs when remove_logs is True."""
write_file(self.log1, "cloud-init-log")
write_file(self.log2, "cloud-init-output-log")
self.assertFalse(
os.path.exists(self.artifact_dir), "Unexpected artifacts dir"
)
retcode = wrap_and_call(
"cloudinit.cmd.clean",
{"Init": {"side_effect": self.init_class}},
clean.remove_artifacts,
remove_logs=True,
)
self.assertFalse(os.path.exists(self.log1), "Unexpected file")
self.assertFalse(os.path.exists(self.log2), "Unexpected file")
self.assertEqual(0, retcode)
def test_remove_artifacts_preserves_logs(self):
"""remove_artifacts leaves logs when remove_logs is False."""
write_file(self.log1, "cloud-init-log")
write_file(self.log2, "cloud-init-output-log")
retcode = wrap_and_call(
"cloudinit.cmd.clean",
{"Init": {"side_effect": self.init_class}},
clean.remove_artifacts,
remove_logs=False,
)
self.assertTrue(os.path.exists(self.log1), "Missing expected file")
self.assertTrue(os.path.exists(self.log2), "Missing expected file")
self.assertEqual(0, retcode)
def test_remove_artifacts_removes_unlinks_symlinks(self):
"""remove_artifacts cleans artifacts dir unlinking any symlinks."""
dir1 = os.path.join(self.artifact_dir, "dir1")
ensure_dir(dir1)
symlink = os.path.join(self.artifact_dir, "mylink")
sym_link(dir1, symlink)
retcode = wrap_and_call(
"cloudinit.cmd.clean",
{"Init": {"side_effect": self.init_class}},
clean.remove_artifacts,
remove_logs=False,
)
self.assertEqual(0, retcode)
for path in (dir1, symlink):
self.assertFalse(
os.path.exists(path), "Unexpected {0} dir".format(path)
)
def test_remove_artifacts_removes_artifacts_skipping_seed(self):
"""remove_artifacts cleans artifacts dir with exception of seed dir."""
dirs = [
self.artifact_dir,
os.path.join(self.artifact_dir, "seed"),
os.path.join(self.artifact_dir, "dir1"),
os.path.join(self.artifact_dir, "dir2"),
]
for _dir in dirs:
ensure_dir(_dir)
retcode = wrap_and_call(
"cloudinit.cmd.clean",
{"Init": {"side_effect": self.init_class}},
clean.remove_artifacts,
remove_logs=False,
)
self.assertEqual(0, retcode)
for expected_dir in dirs[:2]:
self.assertTrue(
os.path.exists(expected_dir),
"Missing {0} dir".format(expected_dir),
)
for deleted_dir in dirs[2:]:
self.assertFalse(
os.path.exists(deleted_dir),
"Unexpected {0} dir".format(deleted_dir),
)
def test_remove_artifacts_removes_artifacts_removes_seed(self):
"""remove_artifacts removes seed dir when remove_seed is True."""
dirs = [
self.artifact_dir,
os.path.join(self.artifact_dir, "seed"),
os.path.join(self.artifact_dir, "dir1"),
os.path.join(self.artifact_dir, "dir2"),
]
for _dir in dirs:
ensure_dir(_dir)
retcode = wrap_and_call(
"cloudinit.cmd.clean",
{"Init": {"side_effect": self.init_class}},
clean.remove_artifacts,
remove_logs=False,
remove_seed=True,
)
self.assertEqual(0, retcode)
self.assertTrue(
os.path.exists(self.artifact_dir), "Missing artifact dir"
)
for deleted_dir in dirs[1:]:
self.assertFalse(
os.path.exists(deleted_dir),
"Unexpected {0} dir".format(deleted_dir),
)
def test_remove_artifacts_returns_one_on_errors(self):
"""remove_artifacts returns non-zero on failure and prints an error."""
ensure_dir(self.artifact_dir)
ensure_dir(os.path.join(self.artifact_dir, "dir1"))
with mock.patch("sys.stderr", new_callable=StringIO) as m_stderr:
retcode = wrap_and_call(
"cloudinit.cmd.clean",
{
"del_dir": {"side_effect": OSError("oops")},
"Init": {"side_effect": self.init_class},
},
clean.remove_artifacts,
remove_logs=False,
)
self.assertEqual(1, retcode)
self.assertEqual(
"Error:\nCould not remove %s/dir1: oops\n" % self.artifact_dir,
m_stderr.getvalue(),
)
def test_handle_clean_args_reboots(self):
"""handle_clean_args_reboots when reboot arg is provided."""
called_cmds = []
def fake_subp(cmd, capture):
called_cmds.append((cmd, capture))
return "", ""
myargs = namedtuple("MyArgs", "remove_logs remove_seed reboot")
cmdargs = myargs(remove_logs=False, remove_seed=False, reboot=True)
retcode = wrap_and_call(
"cloudinit.cmd.clean",
{
"subp": {"side_effect": fake_subp},
"Init": {"side_effect": self.init_class},
},
clean.handle_clean_args,
name="does not matter",
args=cmdargs,
)
self.assertEqual(0, retcode)
self.assertEqual([(["shutdown", "-r", "now"], False)], called_cmds)
def test_status_main(self):
"""clean.main can be run as a standalone script."""
write_file(self.log1, "cloud-init-log")
with self.assertRaises(SystemExit) as context_manager:
wrap_and_call(
"cloudinit.cmd.clean",
{
"Init": {"side_effect": self.init_class},
"sys.argv": {"new": ["clean", "--logs"]},
},
clean.main,
)
self.assertEqual(0, context_manager.exception.code)
self.assertFalse(
os.path.exists(self.log1), "Unexpected log {0}".format(self.log1)
)
# vi: ts=4 expandtab syntax=python
|
