~landscape/landscape-charm/ops-framework

« back to all changes in this revision

Viewing changes to tests/test_settings_files.py

  • Committer: Mitch Burton
  • Date: 2023-01-03 22:10:55 UTC
  • Revision ID: mitch.burton@canonical.com-20230103221055-6cyfjveygeozom78
better leadership-related charm statuses
better haproxy settings for leader vs non-leader

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Copyright 2022 Canonical Ltd
 
2
 
 
3
import os
 
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
 
10
 
 
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)
 
15
 
 
16
 
 
17
class CapturingBytesIO(BytesIO):
 
18
    """
 
19
    A BytesIO subclass that maintains its contents after being closed.
 
20
    """
 
21
    def __init__(self, *args, **kwargs):
 
22
        super().__init__(*args, **kwargs)
 
23
 
 
24
        self.captured = b""
 
25
 
 
26
    def close(self, *args, **kwargs):
 
27
        self.captured = self.getvalue()
 
28
 
 
29
        return super().close(*args, **kwargs)
 
30
 
 
31
 
 
32
class CapturingStringIO(StringIO):
 
33
    """
 
34
    A StringIO subclass that maintains its contents after being closed.
 
35
    """
 
36
    def __init__(self, *args, **kwargs):
 
37
        super().__init__(*args, **kwargs)
 
38
 
 
39
        self.captured = ""
 
40
 
 
41
    def close(self, *args, **kwargs):
 
42
        self.captured = self.getvalue()
 
43
 
 
44
        return super().close(*args, **kwargs)
 
45
 
 
46
 
 
47
class PrependDefaultSettingsTestCase(TestCase):
 
48
 
 
49
    def test_prepend(self):
 
50
        infile = StringIO("# Second line")
 
51
        outfile = CapturingStringIO()
 
52
 
 
53
        def return_settings(path, mode):
 
54
            if mode == "r":
 
55
                return infile
 
56
            else:
 
57
                return outfile
 
58
 
 
59
        with patch("builtins.open") as mock_open:
 
60
            mock_open.side_effect = return_settings
 
61
            prepend_default_settings({"TEST": "yes"})
 
62
 
 
63
        self.assertEqual(outfile.captured, 'TEST="yes"\n# Second line')
 
64
 
 
65
 
 
66
class UpdateDefaultSettingsTestCase(TestCase):
 
67
 
 
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()
 
72
 
 
73
        def return_settings(path, mode):
 
74
            if mode == "r":
 
75
                return infile
 
76
            else:
 
77
                return outfile
 
78
 
 
79
        with patch("builtins.open") as mock_open:
 
80
            mock_open.side_effect = return_settings
 
81
            update_default_settings({"TEST": "yes"})
 
82
 
 
83
        self.assertEqual(outfile.captured, 'TEST="yes"\n')
 
84
 
 
85
    def test_setting_does_not_exist(self):
 
86
        """
 
87
        Tests that nothing is changed if the setting does not exist.
 
88
        """
 
89
        infile = StringIO('TEST="no"\n#comment\n')
 
90
        outfile = CapturingStringIO()
 
91
 
 
92
        def return_settings(path, mode):
 
93
            if mode == "r":
 
94
                return infile
 
95
            else:
 
96
                return outfile
 
97
 
 
98
        with patch("builtins.open") as mock_open:
 
99
            mock_open.side_effect = return_settings
 
100
            update_default_settings({"TEST2": "yes"})
 
101
 
 
102
        self.assertEqual(outfile.captured, 'TEST="no"\n#comment\n')
 
103
 
 
104
 
 
105
class UpdateServiceConfTestCase(TestCase):
 
106
 
 
107
    def test_no_section(self):
 
108
        """
 
109
        Tests that a new config section is created if it does not
 
110
        exist.
 
111
        """
 
112
        infile = StringIO("[fixed]\nold = no\n")
 
113
        outfile = CapturingStringIO()
 
114
 
 
115
        i = 0
 
116
 
 
117
        def return_conf(path, *args, **kwargs):
 
118
            nonlocal i
 
119
            retval = (infile, outfile)[i]
 
120
            i += 1
 
121
            return retval
 
122
 
 
123
        with patch("builtins.open") as open_mock:
 
124
            open_mock.side_effect = return_conf
 
125
            update_service_conf({"test": {"new": "yes"}})
 
126
 
 
127
        self.assertEqual(outfile.captured,
 
128
                         "[fixed]\nold = no\n\n[test]\nnew = yes\n\n")
 
129
 
 
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()
 
134
 
 
135
        i = 0
 
136
 
 
137
        def return_conf(path, *args, **kwargs):
 
138
            nonlocal i
 
139
            retval = (infile, outfile)[i]
 
140
            i += 1
 
141
            return retval
 
142
 
 
143
        with patch("builtins.open") as open_mock:
 
144
            open_mock.side_effect = return_conf
 
145
            update_service_conf({"fixed": {"old": "yes"}})
 
146
 
 
147
        self.assertEqual(outfile.captured, "[fixed]\nold = yes\n\n")
 
148
 
 
149
 
 
150
class WriteLicenseFileTestCase(TestCase):
 
151
 
 
152
    def test_from_file(self):
 
153
        """
 
154
        Tests that a license can be read from a file:// and written.
 
155
        """
 
156
        outfile = CapturingBytesIO()
 
157
        tempdir = TemporaryDirectory()
 
158
        license_file = os.path.join(tempdir.name, "license.txt")
 
159
 
 
160
        with open(license_file, "wb") as fp:
 
161
            fp.write(b"TEST LICENSE")
 
162
 
 
163
        orig_open = open
 
164
 
 
165
        def return_license(path, *args, **kwargs):
 
166
            if path.startswith("/etc/landscape"):
 
167
                return outfile
 
168
            else:
 
169
                return orig_open(path, *args, **kwargs)
 
170
 
 
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)
 
175
 
 
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")
 
179
 
 
180
        tempdir.cleanup()
 
181
 
 
182
    def test_from_url_URLError(self):
 
183
        """
 
184
        Tests that a LicenseFileReadException is raised if the license
 
185
        file is unreadable.
 
186
        """
 
187
        with patch("settings_files.urlopen") as urlopen_mock:
 
188
            urlopen_mock.side_effect = URLError("unreachable")
 
189
            self.assertRaises(
 
190
                LicenseFileReadException,
 
191
                write_license_file,
 
192
                "http://localhost2:12345/random",
 
193
                1000,
 
194
                1000,
 
195
            )
 
196
 
 
197
    def test_b64_encoded(self):
 
198
        """
 
199
        Tests that a license can be directly written from a b64-encoded
 
200
        bytestring.
 
201
        """
 
202
        outfile = CapturingBytesIO()
 
203
        license_bytes = b64encode(b"LICENSE").decode()
 
204
 
 
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)
 
209
 
 
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")
 
213
 
 
214
    def test_b64_error(self):
 
215
        """
 
216
        Tests that a LicenseFileReadException is raised if the license
 
217
        is invalid b64.
 
218
        """
 
219
        self.assertRaises(
 
220
            LicenseFileReadException,
 
221
            write_license_file,
 
222
            "notvalidb64haha",
 
223
            1000,
 
224
            1000,
 
225
        )
 
226
 
 
227
 
 
228
class WriteSSLCertTestCase(TestCase):
 
229
 
 
230
    def test_write(self):
 
231
        """
 
232
        Tests that we can write a b64-encoded string to the correct
 
233
        path.
 
234
        """
 
235
        outfile = CapturingBytesIO()
 
236
 
 
237
        with patch("builtins.open") as open_mock:
 
238
            open_mock.return_value = outfile
 
239
            write_ssl_cert(b64encode(b"SSL CERT").decode())
 
240
 
 
241
        self.assertEqual(outfile.captured, b"SSL CERT")
 
242
 
 
243
    def test_b64_error(self):
 
244
        """
 
245
        Tests that an SSLCertReadException is raised if the cert is
 
246
        invalid b64.
 
247
        """
 
248
        with patch("builtins.open"):
 
249
            self.assertRaises(
 
250
                SSLCertReadException,
 
251
                write_ssl_cert,
 
252
                "notvalidb64haha",
 
253
            )