# # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License as # published by the Free Software Foundation; either version 2 of the # License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 # USA import datetime from gi.repository import GLib import http.client from mock import Mock import logging import sys import unittest import yaml from UpdateManager.Core.LivePatchSocket import LivePatchSocket, LivePatchFix status0 = {'architecture': 'x86_64', 'boot-time': datetime.datetime(2017, 6, 27, 11, 16), 'client-version': '7.21', 'cpu-model': 'Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz', 'last-check': datetime.datetime(2017, 6, 28, 14, 23, 29, 683361), 'machine-id': 123456789, 'machine-token': 987654321, 'uptime': '27h12m12s'} status1 = {'architecture': 'x86_64', 'boot-time': datetime.datetime(2017, 6, 27, 11, 16), 'client-version': '7.21', 'cpu-model': 'Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz', 'last-check': datetime.datetime(2017, 6, 28, 14, 23, 29, 683361), 'machine-id': 123456789, 'machine-token': 987654321, 'status': [{'kernel': '4.4.0-78.99-generic', 'livepatch': {'checkState': 'needs-check', 'fixes': '', 'patchState': 'nothing-to-apply', 'version': '24.2'}, 'running': True}], 'uptime': '27h12m12s'} status2 = {'architecture': 'x86_64', 'boot-time': datetime.datetime(2017, 6, 27, 11, 16), 'client-version': '7.21', 'cpu-model': 'Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz', 'last-check': datetime.datetime(2017, 6, 28, 14, 23, 29, 683361), 'machine-id': 123456789, 'machine-token': 987654321, 'status': [{'kernel': '4.4.0-78.99-generic', 'livepatch': {'checkState': 'checked', 'fixes': '* CVE-2016-0001\n' '* CVE-2016-0002\n' '* CVE-2017-0001 (unpatched)\n' '* CVE-2017-0001', 'patchState': 'applied', 'version': '24.2'}, 'running': True}], 'uptime': '27h12m12s'} class TestUtils(object): @staticmethod def __TimeoutCallback(user_data=None): user_data[0] = True return False @staticmethod def __ScheduleTimeout(timeout_reached, timeout_duration=10): return GLib.timeout_add(timeout_duration, TestUtils.__TimeoutCallback, timeout_reached) @staticmethod def WaitUntilMSec(instance, check_function, expected_result=True, max_wait=500, error_msg=''): instance.assertIsNotNone(check_function) timeout_reached = [False] timeout_id = TestUtils.__ScheduleTimeout(timeout_reached, max_wait) result = None while not timeout_reached[0]: result = check_function() if result == expected_result: break GLib.MainContext.default().iteration(True) if result == expected_result: GLib.Source.remove(timeout_id) instance.assertEqual(expected_result, result, error_msg) class MockResponse(): def __init__(self, status, data): self.status = status self.data = data def read(self): return yaml.dump(self.data) class TestLivePatchSocket(unittest.TestCase): def test_get_check_state(self): check_state = LivePatchSocket.get_check_state(status0) self.assertEqual(check_state, 'check-failed') check_state = LivePatchSocket.get_check_state(status1) self.assertEqual(check_state, 'needs-check') check_state = LivePatchSocket.get_check_state(status2) self.assertEqual(check_state, 'checked') def test_get_patch_state(self): patch_state = LivePatchSocket.get_patch_state(status0) self.assertEqual(patch_state, 'unknown') patch_state = LivePatchSocket.get_patch_state(status1) self.assertEqual(patch_state, 'nothing-to-apply') patch_state = LivePatchSocket.get_patch_state(status2) self.assertEqual(patch_state, 'applied') def test_get_fixes(self): fixes = LivePatchSocket.get_fixes(status0) self.assertEqual(fixes, []) fixes = LivePatchSocket.get_fixes(status1) self.assertEqual(fixes, []) fixes = LivePatchSocket.get_fixes(status2) self.assertEqual(fixes, [LivePatchFix('CVE-2016-0001'), LivePatchFix('CVE-2016-0002'), LivePatchFix('CVE-2017-0001 (unpatched)'), LivePatchFix('CVE-2017-0001')]) def test_livepatch_fix(self): fix = LivePatchFix('CVE-2016-0001') self.assertEqual(fix.name, 'CVE-2016-0001') self.assertTrue(fix.patched) fix = LivePatchFix('CVE-2016-0001 (unpatched)') self.assertEqual(fix.name, 'CVE-2016-0001') self.assertFalse(fix.patched) def test_callback_not_active(self): mock_http_conn = Mock(spec=http.client.HTTPConnection) attrs = {'getresponse.return_value': MockResponse(400, None)} mock_http_conn.configure_mock(**attrs) cb_called = [False] def on_done(active, check_state, patch_state, fixes): cb_called[0] = True self.assertFalse(active) lp = LivePatchSocket(mock_http_conn) lp.get_status(on_done) mock_http_conn.request.assert_called_with( 'GET', '/status?verbose=True') TestUtils.WaitUntilMSec(self, lambda: cb_called[0] is True) def test_callback_active(self): mock_http_conn = Mock(spec=http.client.HTTPConnection) attrs = {'getresponse.return_value': MockResponse(200, status2)} mock_http_conn.configure_mock(**attrs) cb_called = [False] def on_done(active, check_state, patch_state, fixes): cb_called[0] = True self.assertTrue(active) self.assertEqual(check_state, 'checked') self.assertEqual(patch_state, 'applied') self.assertEqual(fixes, [LivePatchFix('CVE-2016-0001'), LivePatchFix('CVE-2016-0002'), LivePatchFix('CVE-2017-0001 (unpatched)'), LivePatchFix('CVE-2017-0001')]) lp = LivePatchSocket(mock_http_conn) lp.get_status(on_done) mock_http_conn.request.assert_called_with( 'GET', '/status?verbose=True') TestUtils.WaitUntilMSec(self, lambda: cb_called[0] is True) if __name__ == '__main__': if len(sys.argv) > 1 and sys.argv[1] == "-v": logging.basicConfig(level=logging.DEBUG) unittest.main()