~ubuntu-core-dev/update-manager/main

« back to all changes in this revision

Viewing changes to tests/test_livepatch_socket.py

  • Committer: Benjamin Drung
  • Date: 2023-02-13 12:45:23 UTC
  • Revision ID: benjamin.drung@canonical.com-20230213124523-4f1nv2fyvsdl35jb
Fix pycodestyle complains by formatting Python code with black

```
black -C -l 79 .
```

Show diffs side-by-side

added added

removed removed

Lines of Context:
26
26
from UpdateManager.Core.LivePatchSocket import LivePatchSocket, LivePatchFix
27
27
 
28
28
 
29
 
status0 = {'architecture': 'x86_64',
30
 
           'boot-time': datetime.datetime(2017, 6, 27, 11, 16),
31
 
           'client-version': '7.21',
32
 
           'cpu-model': 'Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz',
33
 
           'last-check': datetime.datetime(2017, 6, 28, 14, 23, 29, 683361),
34
 
           'machine-id': 123456789,
35
 
           'machine-token': 987654321,
36
 
           'uptime': '27h12m12s'}
37
 
 
38
 
status1 = {'architecture': 'x86_64',
39
 
           'boot-time': datetime.datetime(2017, 6, 27, 11, 16),
40
 
           'client-version': '7.21',
41
 
           'cpu-model': 'Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz',
42
 
           'last-check': datetime.datetime(2017, 6, 28, 14, 23, 29, 683361),
43
 
           'machine-id': 123456789,
44
 
           'machine-token': 987654321,
45
 
           'status': [{'kernel': '4.4.0-78.99-generic',
46
 
                       'livepatch': {'checkState': 'needs-check',
47
 
                                     'fixes': '',
48
 
                                     'patchState': 'nothing-to-apply',
49
 
                                     'version': '24.2'},
50
 
                       'running': True}],
51
 
           'uptime': '27h12m12s'}
52
 
 
53
 
status2 = {'architecture': 'x86_64',
54
 
           'boot-time': datetime.datetime(2017, 6, 27, 11, 16),
55
 
           'client-version': '7.21',
56
 
           'cpu-model': 'Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz',
57
 
           'last-check': datetime.datetime(2017, 6, 28, 14, 23, 29, 683361),
58
 
           'machine-id': 123456789,
59
 
           'machine-token': 987654321,
60
 
           'status': [{'kernel': '4.4.0-78.99-generic',
61
 
                       'livepatch': {'checkState': 'checked',
62
 
                                     'fixes': '* CVE-2016-0001\n'
63
 
                                     '* CVE-2016-0002\n'
64
 
                                     '* CVE-2017-0001 (unpatched)\n'
65
 
                                     '* CVE-2017-0001',
66
 
                                     'patchState': 'applied',
67
 
                                     'version': '24.2'},
68
 
                       'running': True}],
69
 
           'uptime': '27h12m12s'}
 
29
status0 = {
 
30
    "architecture": "x86_64",
 
31
    "boot-time": datetime.datetime(2017, 6, 27, 11, 16),
 
32
    "client-version": "7.21",
 
33
    "cpu-model": "Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz",
 
34
    "last-check": datetime.datetime(2017, 6, 28, 14, 23, 29, 683361),
 
35
    "machine-id": 123456789,
 
36
    "machine-token": 987654321,
 
37
    "uptime": "27h12m12s",
 
38
}
 
39
 
 
40
status1 = {
 
41
    "architecture": "x86_64",
 
42
    "boot-time": datetime.datetime(2017, 6, 27, 11, 16),
 
43
    "client-version": "7.21",
 
44
    "cpu-model": "Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz",
 
45
    "last-check": datetime.datetime(2017, 6, 28, 14, 23, 29, 683361),
 
46
    "machine-id": 123456789,
 
47
    "machine-token": 987654321,
 
48
    "status": [
 
49
        {
 
50
            "kernel": "4.4.0-78.99-generic",
 
51
            "livepatch": {
 
52
                "checkState": "needs-check",
 
53
                "fixes": "",
 
54
                "patchState": "nothing-to-apply",
 
55
                "version": "24.2",
 
56
            },
 
57
            "running": True,
 
58
        }
 
59
    ],
 
60
    "uptime": "27h12m12s",
 
61
}
 
62
 
 
63
status2 = {
 
64
    "architecture": "x86_64",
 
65
    "boot-time": datetime.datetime(2017, 6, 27, 11, 16),
 
66
    "client-version": "7.21",
 
67
    "cpu-model": "Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz",
 
68
    "last-check": datetime.datetime(2017, 6, 28, 14, 23, 29, 683361),
 
69
    "machine-id": 123456789,
 
70
    "machine-token": 987654321,
 
71
    "status": [
 
72
        {
 
73
            "kernel": "4.4.0-78.99-generic",
 
74
            "livepatch": {
 
75
                "checkState": "checked",
 
76
                "fixes": "* CVE-2016-0001\n"
 
77
                "* CVE-2016-0002\n"
 
78
                "* CVE-2017-0001 (unpatched)\n"
 
79
                "* CVE-2017-0001",
 
80
                "patchState": "applied",
 
81
                "version": "24.2",
 
82
            },
 
83
            "running": True,
 
84
        }
 
85
    ],
 
86
    "uptime": "27h12m12s",
 
87
}
70
88
 
71
89
 
72
90
class TestUtils(object):
73
 
 
74
91
    @staticmethod
75
92
    def __TimeoutCallback(user_data=None):
76
93
        user_data[0] = True
78
95
 
79
96
    @staticmethod
80
97
    def __ScheduleTimeout(timeout_reached, timeout_duration=10):
81
 
        return GLib.timeout_add(timeout_duration,
82
 
                                TestUtils.__TimeoutCallback,
83
 
                                timeout_reached)
 
98
        return GLib.timeout_add(
 
99
            timeout_duration, TestUtils.__TimeoutCallback, timeout_reached
 
100
        )
84
101
 
85
102
    @staticmethod
86
 
    def WaitUntilMSec(instance, check_function, expected_result=True,
87
 
                      max_wait=500, error_msg=''):
 
103
    def WaitUntilMSec(
 
104
        instance,
 
105
        check_function,
 
106
        expected_result=True,
 
107
        max_wait=500,
 
108
        error_msg="",
 
109
    ):
88
110
        instance.assertIsNotNone(check_function)
89
111
 
90
112
        timeout_reached = [False]
103
125
        instance.assertEqual(expected_result, result, error_msg)
104
126
 
105
127
 
106
 
class MockResponse():
107
 
 
 
128
class MockResponse:
108
129
    def __init__(self, status, data):
109
130
        self.status = status
110
131
        self.data = data
114
135
 
115
136
 
116
137
class TestLivePatchSocket(unittest.TestCase):
117
 
 
118
138
    def test_get_check_state(self):
119
139
        check_state = LivePatchSocket.get_check_state(status0)
120
 
        self.assertEqual(check_state, 'check-failed')
 
140
        self.assertEqual(check_state, "check-failed")
121
141
        check_state = LivePatchSocket.get_check_state(status1)
122
 
        self.assertEqual(check_state, 'needs-check')
 
142
        self.assertEqual(check_state, "needs-check")
123
143
        check_state = LivePatchSocket.get_check_state(status2)
124
 
        self.assertEqual(check_state, 'checked')
 
144
        self.assertEqual(check_state, "checked")
125
145
 
126
146
    def test_get_patch_state(self):
127
147
        patch_state = LivePatchSocket.get_patch_state(status0)
128
 
        self.assertEqual(patch_state, 'unknown')
 
148
        self.assertEqual(patch_state, "unknown")
129
149
        patch_state = LivePatchSocket.get_patch_state(status1)
130
 
        self.assertEqual(patch_state, 'nothing-to-apply')
 
150
        self.assertEqual(patch_state, "nothing-to-apply")
131
151
        patch_state = LivePatchSocket.get_patch_state(status2)
132
 
        self.assertEqual(patch_state, 'applied')
 
152
        self.assertEqual(patch_state, "applied")
133
153
 
134
154
    def test_get_fixes(self):
135
155
        fixes = LivePatchSocket.get_fixes(status0)
137
157
        fixes = LivePatchSocket.get_fixes(status1)
138
158
        self.assertEqual(fixes, [])
139
159
        fixes = LivePatchSocket.get_fixes(status2)
140
 
        self.assertEqual(fixes, [LivePatchFix('CVE-2016-0001'),
141
 
                                 LivePatchFix('CVE-2016-0002'),
142
 
                                 LivePatchFix('CVE-2017-0001 (unpatched)'),
143
 
                                 LivePatchFix('CVE-2017-0001')])
 
160
        self.assertEqual(
 
161
            fixes,
 
162
            [
 
163
                LivePatchFix("CVE-2016-0001"),
 
164
                LivePatchFix("CVE-2016-0002"),
 
165
                LivePatchFix("CVE-2017-0001 (unpatched)"),
 
166
                LivePatchFix("CVE-2017-0001"),
 
167
            ],
 
168
        )
144
169
 
145
170
    def test_livepatch_fix(self):
146
 
        fix = LivePatchFix('CVE-2016-0001')
147
 
        self.assertEqual(fix.name, 'CVE-2016-0001')
 
171
        fix = LivePatchFix("CVE-2016-0001")
 
172
        self.assertEqual(fix.name, "CVE-2016-0001")
148
173
        self.assertTrue(fix.patched)
149
174
 
150
 
        fix = LivePatchFix('CVE-2016-0001 (unpatched)')
151
 
        self.assertEqual(fix.name, 'CVE-2016-0001')
 
175
        fix = LivePatchFix("CVE-2016-0001 (unpatched)")
 
176
        self.assertEqual(fix.name, "CVE-2016-0001")
152
177
        self.assertFalse(fix.patched)
153
178
 
154
179
    def test_callback_not_active(self):
155
180
        mock_http_conn = Mock(spec=http.client.HTTPConnection)
156
 
        attrs = {'getresponse.return_value': MockResponse(400, None)}
 
181
        attrs = {"getresponse.return_value": MockResponse(400, None)}
157
182
        mock_http_conn.configure_mock(**attrs)
158
183
 
159
184
        cb_called = [False]
166
191
        lp.get_status(on_done)
167
192
 
168
193
        mock_http_conn.request.assert_called_with(
169
 
            'GET', '/status?verbose=True')
 
194
            "GET", "/status?verbose=True"
 
195
        )
170
196
        TestUtils.WaitUntilMSec(self, lambda: cb_called[0] is True)
171
197
 
172
198
    def test_callback_active(self):
173
199
        mock_http_conn = Mock(spec=http.client.HTTPConnection)
174
 
        attrs = {'getresponse.return_value': MockResponse(200, status2)}
 
200
        attrs = {"getresponse.return_value": MockResponse(200, status2)}
175
201
        mock_http_conn.configure_mock(**attrs)
176
202
 
177
203
        cb_called = [False]
179
205
        def on_done(active, check_state, patch_state, fixes):
180
206
            cb_called[0] = True
181
207
            self.assertTrue(active)
182
 
            self.assertEqual(check_state, 'checked')
183
 
            self.assertEqual(patch_state, 'applied')
184
 
            self.assertEqual(fixes, [LivePatchFix('CVE-2016-0001'),
185
 
                                     LivePatchFix('CVE-2016-0002'),
186
 
                                     LivePatchFix('CVE-2017-0001 (unpatched)'),
187
 
                                     LivePatchFix('CVE-2017-0001')])
 
208
            self.assertEqual(check_state, "checked")
 
209
            self.assertEqual(patch_state, "applied")
 
210
            self.assertEqual(
 
211
                fixes,
 
212
                [
 
213
                    LivePatchFix("CVE-2016-0001"),
 
214
                    LivePatchFix("CVE-2016-0002"),
 
215
                    LivePatchFix("CVE-2017-0001 (unpatched)"),
 
216
                    LivePatchFix("CVE-2017-0001"),
 
217
                ],
 
218
            )
188
219
 
189
220
        lp = LivePatchSocket(mock_http_conn)
190
221
        lp.get_status(on_done)
191
222
 
192
223
        mock_http_conn.request.assert_called_with(
193
 
            'GET', '/status?verbose=True')
 
224
            "GET", "/status?verbose=True"
 
225
        )
194
226
        TestUtils.WaitUntilMSec(self, lambda: cb_called[0] is True)
195
227
 
196
228
 
197
 
if __name__ == '__main__':
 
229
if __name__ == "__main__":
198
230
    if len(sys.argv) > 1 and sys.argv[1] == "-v":
199
231
        logging.basicConfig(level=logging.DEBUG)
200
232
    unittest.main()