starlingx/update
Code Issues Proposed changes

Add additional unit tests for sw-patch

Browse Source 
This review adds unit tests for several of the classes and
methods that were not previously being unit tested.

Increases line coverage of patch_controller from 10% to 20%
Increases line coverage of patch_client from 19% to 24%

Story: 2008943
Task: 47063
Signed-off-by: Al Bailey <al.bailey@windriver.com>
Change-Id: I364e8bc05f9c150a38317a2e3d44fb1868f7ba69
This commit is contained in:
Al Bailey
2022-12-22 19:48:10 +00:00
parent 8b599699ea
commit d61f37e658
3 changed files with 334 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

164
sw-patch/cgcs-patch/cgcs_patch/tests/test_patch_client.py
View File

@@ -100,7 +100,29 @@ class PatchClientTestCase(testtools.TestCase):
self.addCleanup(patcher.stop)
class PatchClientHelpTestCase(PatchClientTestCase):
class PatchClientNonRootMixin(object):
"""
This Mixin Requires self.MOCK_ENV
Disable printing to stdout
Every client call invokes exit which raises SystemExit
This asserts that happens.
"""
def _test_method(self, shell_args=None):
with mock.patch.dict(os.environ, self.MOCK_ENV):
with mock.patch.object(sys, 'argv', shell_args):
# mock 'print' so running unit tests will
# not print to the tox output
with mock.patch('builtins.print'):
# Every client invocation invokes exit
# which raises SystemExit
self.assertRaises(SystemExit,
patch_client.main)
class PatchClientHelpTestCase(PatchClientTestCase, PatchClientNonRootMixin):
"""Test the sw-patch CLI calls that invoke 'help'
'check_for_os_region_name' is mocked to help determine
@@ -108,41 +130,29 @@ class PatchClientHelpTestCase(PatchClientTestCase):
circuit and invoke 'help' in failure cases.
"""
def _test_print_help(self, shell_args=None):
with mock.patch.dict(os.environ, self.MOCK_ENV):
with mock.patch.object(sys, 'argv',
shell_args):
# mock 'print' so running unit tests will
# not print help usage to the tox output
with mock.patch('builtins.print'):
# Every client invocation invokes exit
# which raises SystemExit
self.assertRaises(SystemExit,
patch_client.main)
@mock.patch('cgcs_patch.patch_client.check_for_os_region_name')
def test_main_no_args_calls_help(self, mock_check):
"""When no arguments are called, this should invoke print_help"""
shell_args = [self.PROG, ]
self._test_print_help(shell_args=shell_args)
self._test_method(shell_args=shell_args)
mock_check.assert_not_called()
@mock.patch('cgcs_patch.patch_client.check_for_os_region_name')
def test_main_help(self, mock_check):
"""When no arguments are called, this should invoke print_help"""
shell_args = [self.PROG, "--help"]
self._test_print_help(shell_args=shell_args)
self._test_method(shell_args=shell_args)
mock_check.assert_called()
@mock.patch('cgcs_patch.patch_client.check_for_os_region_name')
def test_main_invalid_action_calls_help(self, mock_check):
"""invalid args should invoke print_help"""
shell_args = [self.PROG, "invalid_arg"]
self._test_print_help(shell_args=shell_args)
self._test_method(shell_args=shell_args)
mock_check.assert_called()
class PatchClientQueryTestCase(PatchClientTestCase):
class PatchClientQueryTestCase(PatchClientTestCase, PatchClientNonRootMixin):
"""Test the sw-patch CLI calls that invoke 'query'"""
TEST_URL_ALL = "http://127.0.0.1:5487/patch/query?show=all"
@@ -168,28 +178,120 @@ class PatchClientQueryTestCase(PatchClientTestCase):
self.mock_map[self.TEST_URL_APPLIED] = FakeResponse(
self.TEST_PATCH_DATA_SHOW_APPLIED, 200)
def _test_query(self, shell_args=None):
with mock.patch.dict(os.environ, self.MOCK_ENV):
with mock.patch.object(sys, 'argv',
shell_args):
# mock 'print' so running unit tests will
# not print to the tox output
with mock.patch('builtins.print'):
# Every client invocation invokes exit
# which raises SystemExit
self.assertRaises(SystemExit,
patch_client.main)
def test_query(self):
shell_args = [self.PROG, "query"]
self._test_query(shell_args=shell_args)
self._test_method(shell_args=shell_args)
self.mock_requests_get.assert_called_with(
self.TEST_URL_ALL,
headers=mock.ANY)
def test_query_patch(self):
shell_args = [self.PROG, "query", "applied"]
self._test_query(shell_args=shell_args)
self._test_method(shell_args=shell_args)
self.mock_requests_get.assert_called_with(
self.TEST_URL_APPLIED,
headers=mock.ANY)
class PatchClientWhatRequiresTestCase(PatchClientTestCase, PatchClientNonRootMixin):
TEST_URL_VALID = "http://127.0.0.1:5487/patch/what_requires/" + FAKE_PATCH_ID_1
TEST_WHAT_REQUIRES_VALID = {
"error": "",
"info": FAKE_PATCH_ID_1 + " is not required by any patches.\n",
"warning": ""
}
TEST_URL_INVALID = "http://127.0.0.1:5487/patch/what_requires/" + FAKE_PATCH_ID_2
TEST_WHAT_REQUIRES_INVALID = {
"error": "Patch " + FAKE_PATCH_ID_2 + " does not exist\n",
"info": "",
"warning": ""
}
def setUp(self):
super(PatchClientWhatRequiresTestCase, self).setUp()
# update the mock_map with a query result
self.mock_map[self.TEST_URL_VALID] = FakeResponse(
self.TEST_WHAT_REQUIRES_VALID, 200)
self.mock_map[self.TEST_URL_INVALID] = FakeResponse(
self.TEST_WHAT_REQUIRES_INVALID, 200)
def test_what_requires(self):
shell_args = [self.PROG, "what-requires", FAKE_PATCH_ID_1]
self._test_method(shell_args=shell_args)
self.mock_requests_get.assert_called_with(
self.TEST_URL_VALID,
headers=mock.ANY)
def test_what_requires_debug(self):
shell_args = [self.PROG, "--debug", "what-requires", FAKE_PATCH_ID_1]
self._test_method(shell_args=shell_args)
self.mock_requests_get.assert_called_with(
self.TEST_URL_VALID,
headers=mock.ANY)
def test_what_requires_not_found(self):
shell_args = [self.PROG, "what-requires", FAKE_PATCH_ID_2]
self._test_method(shell_args=shell_args)
self.mock_requests_get.assert_called_with(
self.TEST_URL_INVALID,
headers=mock.ANY)
class PatchClientQueryHostsTestCase(PatchClientTestCase, PatchClientNonRootMixin):
TEST_URL = "http://127.0.0.1:5487/patch/query_hosts"
TEST_RESULTS = {'data': [
{
"allow_insvc_patching": 'true',
"hostname": "controller-0",
"interim_state": 'false',
"ip": "192.168.204.3",
"latest_sysroot_commit": "4b26afcf716f1804e70222a5564c2174340c2c6aabae9bcabe3468b2ce309d87",
"nodetype": "controller",
"patch_current": 'true',
"patch_failed": 'false',
"requires_reboot": 'false',
"secs_since_ack": 17,
"stale_details": 'false',
"state": "idle",
"subfunctions": [
"controller",
"worker"
],
"sw_version": "12.34"
},
{
"allow_insvc_patching": 'true',
"hostname": "controller-1",
"interim_state": 'false',
"ip": "192.168.204.4",
"latest_sysroot_commit": "4b26afcf716f1804e70222a5564c2174340c2c6aabae9bcabe3468b2ce309d87",
"nodetype": "controller",
"patch_current": 'true',
"patch_failed": 'false',
"requires_reboot": 'false',
"secs_since_ack": 17,
"stale_details": 'false',
"state": "idle",
"subfunctions": [
"controller",
"worker"
],
"sw_version": "12.34"
}
]}
def setUp(self):
super(PatchClientQueryHostsTestCase, self).setUp()
# update the mock_map with a query result
self.mock_map[self.TEST_URL] = FakeResponse(
self.TEST_RESULTS, 200)
def test_query_hosts(self):
shell_args = [self.PROG, "query-hosts"]
self._test_method(shell_args=shell_args)
# for some reason, this does not pass a HEADER
self.mock_requests_get.assert_called_with(
self.TEST_URL)

43
sw-patch/cgcs-patch/cgcs_patch/tests/test_patch_controller.py
View File

@@ -6,13 +6,50 @@
import mock
import testtools
import time
from cgcs_patch.patch_controller import AgentNeighbour
from cgcs_patch.patch_controller import ControllerNeighbour
from cgcs_patch.patch_controller import PatchController
class CgcsPatchControllerTestCase(testtools.TestCase):
@mock.patch('builtins.open')
def test_cgcs_patch_controller_instantiate(self, _mock_open):
pc = PatchController()
self.assertIsNotNone(pc)
def test_controller(self, _mock_open):
# Disable the 'open'
test_obj = PatchController()
self.assertIsNotNone(test_obj)
def test_controller_neighbour(self):
test_obj = ControllerNeighbour()
self.assertIsNotNone(test_obj)
# reset the age
test_obj.rx_ack()
# get the age. this number should be zero
first_age = test_obj.get_age()
# delay one second. The age should be one
delay = 1
time.sleep(delay)
second_age = test_obj.get_age()
self.assertTrue(second_age > first_age)
# second_age should equal delay
# to accomodate overloaded machines, we use >=
self.assertTrue(second_age >= delay)
# reset the age. the new age should be zero
test_obj.rx_ack()
third_age = test_obj.get_age()
self.assertTrue(third_age < second_age)
# set synched to True
test_obj.rx_synced()
self.assertTrue(test_obj.get_synced())
# set synched to False
test_obj.clear_synced()
self.assertFalse(test_obj.get_synced())
def test_agent_neighbour(self):
test_ip = '127.0.0.1'
test_obj = AgentNeighbour(test_ip)
self.assertIsNotNone(test_obj)

161
sw-patch/cgcs-patch/cgcs_patch/tests/test_patch_controller_messages.py Normal file
View File

@@ -0,0 +1,161 @@
#
# SPDX-License-Identifier: Apache-2.0
#
# Copyright (c) 2019 Wind River Systems, Inc.
#
import testtools
from unittest import mock
from cgcs_patch.messages import PatchMessage
from cgcs_patch.patch_controller import PatchMessageHello
from cgcs_patch.patch_controller import PatchMessageHelloAck
from cgcs_patch.patch_controller import PatchMessageSyncReq
from cgcs_patch.patch_controller import PatchMessageSyncComplete
from cgcs_patch.patch_controller import PatchMessageHelloAgent
from cgcs_patch.patch_controller import PatchMessageSendLatestFeedCommit
from cgcs_patch.patch_controller import PatchMessageHelloAgentAck
from cgcs_patch.patch_controller import PatchMessageQueryDetailed
from cgcs_patch.patch_controller import PatchMessageQueryDetailedResp
from cgcs_patch.patch_controller import PatchMessageAgentInstallReq
from cgcs_patch.patch_controller import PatchMessageAgentInstallResp
from cgcs_patch.patch_controller import PatchMessageDropHostReq
FAKE_AGENT_ADDRESS = "127.0.0.1"
FAKE_AGENT_MCAST_GROUP = "239.1.1.4"
FAKE_CONTROLLER_ADDRESS = "127.0.0.1"
FAKE_HOST_IP = "10.10.10.2"
FAKE_OSTREE_FEED_COMMIT = "12345"
class FakePatchController(object):
def __init__(self):
self.agent_address = FAKE_AGENT_ADDRESS
self.allow_insvc_patching = True
self.controller_address = FAKE_CONTROLLER_ADDRESS
self.controller_neighbours = {}
self.hosts = {}
self.interim_state = {}
self.latest_feed_commit = FAKE_OSTREE_FEED_COMMIT
self.patch_op_counter = 0
self.sock_in = None
self.sock_out = None
# mock all the lock objects
self.controller_neighbours_lock = mock.Mock()
self.hosts_lock = mock.Mock()
self.patch_data_lock = mock.Mock()
self.socket_lock = mock.Mock()
# mock the patch data
self.base_pkgdata = mock.Mock()
self.patch_data = mock.Mock()
def check_patch_states(self):
pass
def drop_host(self, host_ip, sync_nbr=True):
pass
def sync_from_nbr(self, host):
pass
class PatchControllerMessagesTestCase(testtools.TestCase):
message_classes = [
PatchMessageHello,
PatchMessageHelloAck,
PatchMessageSyncReq,
PatchMessageSyncComplete,
PatchMessageHelloAgent,
PatchMessageSendLatestFeedCommit,
PatchMessageHelloAgentAck,
PatchMessageQueryDetailed,
PatchMessageQueryDetailedResp,
PatchMessageAgentInstallReq,
PatchMessageAgentInstallResp,
PatchMessageDropHostReq,
]
def test_message_class_creation(self):
for message_class in PatchControllerMessagesTestCase.message_classes:
test_obj = message_class()
self.assertIsNotNone(test_obj)
self.assertIsInstance(test_obj, PatchMessage)
@mock.patch('cgcs_patch.patch_controller.pc', FakePatchController())
def test_message_class_encode(self):
"""'encode' method populates self.message"""
# mock the global patch_controller 'pc' variable used by encode
# PatchMessageQueryDetailedResp does not support 'encode'
# so it can be executed, but it will not change the message
excluded = [
PatchMessageQueryDetailedResp
]
for message_class in PatchControllerMessagesTestCase.message_classes:
test_obj = message_class()
# message variable should be empty dict (ie: False)
self.assertFalse(test_obj.message)
test_obj.encode()
# message variable no longer empty (ie: True)
if message_class not in excluded:
self.assertTrue(test_obj.message)
# decode one message into another
test_obj2 = message_class()
test_obj2.decode(test_obj.message)
# decode does not populate 'message' so nothing to compare
@mock.patch('cgcs_patch.patch_controller.pc', FakePatchController())
@mock.patch('cgcs_patch.config.agent_mcast_group', FAKE_AGENT_MCAST_GROUP)
def test_message_class_send(self):
"""'send' writes to a socket"""
mock_sock = mock.Mock()
# socket sendto and sendall are not called by:
# PatchMessageHelloAgentAck
# PatchMessageQueryDetailedResp
# PatchMessageAgentInstallResp,
send_to = [
PatchMessageHello,
PatchMessageHelloAck,
PatchMessageSyncReq,
PatchMessageSyncComplete,
PatchMessageHelloAgent,
PatchMessageSendLatestFeedCommit,
PatchMessageAgentInstallReq,
PatchMessageDropHostReq,
]
send_all = [
PatchMessageQueryDetailed,
]
for message_class in PatchControllerMessagesTestCase.message_classes:
mock_sock.reset_mock()
test_obj = message_class()
test_obj.send(mock_sock)
if message_class in send_to:
mock_sock.sendto.assert_called()
if message_class in send_all:
mock_sock.sendall.assert_called()
@mock.patch('cgcs_patch.patch_controller.pc', FakePatchController())
def test_message_class_handle(self):
"""'handle' method tests"""
addr = [FAKE_CONTROLLER_ADDRESS, ] # addr is a list
mock_sock = mock.Mock()
special_setup = {
PatchMessageDropHostReq: ('ip', FAKE_HOST_IP),
}
for message_class in PatchControllerMessagesTestCase.message_classes:
mock_sock.reset_mock()
test_obj = message_class()
# some classes require special setup
special = special_setup.get(message_class)
if special:
setattr(test_obj, special[0], special[1])
test_obj.handle(mock_sock, addr)