1
# Copyright 2022 Canonical Ltd
4
from base64 import b64encode
5
from io import BytesIO, StringIO
6
from tempfile import TemporaryDirectory
7
from unittest import TestCase
8
from unittest.mock import patch
9
from urllib.error import URLError
11
from settings_files import (
12
LICENSE_FILE, LicenseFileReadException, SSLCertReadException,
13
prepend_default_settings, update_default_settings, update_service_conf,
14
write_license_file, write_ssl_cert)
17
class CapturingBytesIO(BytesIO):
19
A BytesIO subclass that maintains its contents after being closed.
21
def __init__(self, *args, **kwargs):
22
super().__init__(*args, **kwargs)
26
def close(self, *args, **kwargs):
27
self.captured = self.getvalue()
29
return super().close(*args, **kwargs)
32
class CapturingStringIO(StringIO):
34
A StringIO subclass that maintains its contents after being closed.
36
def __init__(self, *args, **kwargs):
37
super().__init__(*args, **kwargs)
41
def close(self, *args, **kwargs):
42
self.captured = self.getvalue()
44
return super().close(*args, **kwargs)
47
class PrependDefaultSettingsTestCase(TestCase):
49
def test_prepend(self):
50
infile = StringIO("# Second line")
51
outfile = CapturingStringIO()
53
def return_settings(path, mode):
59
with patch("builtins.open") as mock_open:
60
mock_open.side_effect = return_settings
61
prepend_default_settings({"TEST": "yes"})
63
self.assertEqual(outfile.captured, 'TEST="yes"\n# Second line')
66
class UpdateDefaultSettingsTestCase(TestCase):
68
def test_setting_exists(self):
69
"""Tests that a setting gets updated if it exists."""
70
infile = StringIO('TEST="no"\n')
71
outfile = CapturingStringIO()
73
def return_settings(path, mode):
79
with patch("builtins.open") as mock_open:
80
mock_open.side_effect = return_settings
81
update_default_settings({"TEST": "yes"})
83
self.assertEqual(outfile.captured, 'TEST="yes"\n')
85
def test_setting_does_not_exist(self):
87
Tests that nothing is changed if the setting does not exist.
89
infile = StringIO('TEST="no"\n#comment\n')
90
outfile = CapturingStringIO()
92
def return_settings(path, mode):
98
with patch("builtins.open") as mock_open:
99
mock_open.side_effect = return_settings
100
update_default_settings({"TEST2": "yes"})
102
self.assertEqual(outfile.captured, 'TEST="no"\n#comment\n')
105
class UpdateServiceConfTestCase(TestCase):
107
def test_no_section(self):
109
Tests that a new config section is created if it does not
112
infile = StringIO("[fixed]\nold = no\n")
113
outfile = CapturingStringIO()
117
def return_conf(path, *args, **kwargs):
119
retval = (infile, outfile)[i]
123
with patch("builtins.open") as open_mock:
124
open_mock.side_effect = return_conf
125
update_service_conf({"test": {"new": "yes"}})
127
self.assertEqual(outfile.captured,
128
"[fixed]\nold = no\n\n[test]\nnew = yes\n\n")
130
def test_section_exists(self):
131
"""Tests that a setting is updated if the section exists."""
132
infile = StringIO("[fixed]\nold = no\n")
133
outfile = CapturingStringIO()
137
def return_conf(path, *args, **kwargs):
139
retval = (infile, outfile)[i]
143
with patch("builtins.open") as open_mock:
144
open_mock.side_effect = return_conf
145
update_service_conf({"fixed": {"old": "yes"}})
147
self.assertEqual(outfile.captured, "[fixed]\nold = yes\n\n")
150
class WriteLicenseFileTestCase(TestCase):
152
def test_from_file(self):
154
Tests that a license can be read from a file:// and written.
156
outfile = CapturingBytesIO()
157
tempdir = TemporaryDirectory()
158
license_file = os.path.join(tempdir.name, "license.txt")
160
with open(license_file, "wb") as fp:
161
fp.write(b"TEST LICENSE")
165
def return_license(path, *args, **kwargs):
166
if path.startswith("/etc/landscape"):
169
return orig_open(path, *args, **kwargs)
171
with patch("builtins.open") as open_mock:
172
open_mock.side_effect = return_license
173
with patch("settings_files.os") as os_mock:
174
write_license_file(f"file://{license_file}", 1000, 1000)
176
os_mock.chmod.assert_called_once_with(LICENSE_FILE, 0o640)
177
os_mock.chown.assert_called_once_with(LICENSE_FILE, 1000, 1000)
178
self.assertEqual(outfile.captured, b"TEST LICENSE")
182
def test_from_url_URLError(self):
184
Tests that a LicenseFileReadException is raised if the license
187
with patch("settings_files.urlopen") as urlopen_mock:
188
urlopen_mock.side_effect = URLError("unreachable")
190
LicenseFileReadException,
192
"http://localhost2:12345/random",
197
def test_b64_encoded(self):
199
Tests that a license can be directly written from a b64-encoded
202
outfile = CapturingBytesIO()
203
license_bytes = b64encode(b"LICENSE").decode()
205
with patch("builtins.open") as open_mock:
206
open_mock.return_value = outfile
207
with patch("settings_files.os") as os_mock:
208
write_license_file(license_bytes, 1000, 1000)
210
os_mock.chmod.assert_called_once_with(LICENSE_FILE, 0o640)
211
os_mock.chown.assert_called_once_with(LICENSE_FILE, 1000, 1000)
212
self.assertEqual(outfile.captured, b"LICENSE")
214
def test_b64_error(self):
216
Tests that a LicenseFileReadException is raised if the license
220
LicenseFileReadException,
228
class WriteSSLCertTestCase(TestCase):
230
def test_write(self):
232
Tests that we can write a b64-encoded string to the correct
235
outfile = CapturingBytesIO()
237
with patch("builtins.open") as open_mock:
238
open_mock.return_value = outfile
239
write_ssl_cert(b64encode(b"SSL CERT").decode())
241
self.assertEqual(outfile.captured, b"SSL CERT")
243
def test_b64_error(self):
245
Tests that an SSLCertReadException is raised if the cert is
248
with patch("builtins.open"):
250
SSLCertReadException,