diff options
Diffstat (limited to 'tests/unittests/config/test_cc_rh_subscription.py')
| -rw-r--r-- | tests/unittests/config/test_cc_rh_subscription.py | 366 |
1 files changed, 226 insertions, 140 deletions
diff --git a/tests/unittests/config/test_cc_rh_subscription.py b/tests/unittests/config/test_cc_rh_subscription.py index bd7ebc98..fcc7db34 100644 --- a/tests/unittests/config/test_cc_rh_subscription.py +++ b/tests/unittests/config/test_cc_rh_subscription.py @@ -5,13 +5,12 @@ import copy import logging -from cloudinit.config import cc_rh_subscription from cloudinit import subp - +from cloudinit.config import cc_rh_subscription from tests.unittests.helpers import CiTestCase, mock SUBMGR = cc_rh_subscription.SubscriptionManager -SUB_MAN_CLI = 'cloudinit.config.cc_rh_subscription._sub_man_cli' +SUB_MAN_CLI = "cloudinit.config.cc_rh_subscription._sub_man_cli" @mock.patch(SUB_MAN_CLI) @@ -26,87 +25,115 @@ class GoodTests(CiTestCase): self.args = [] self.handle = cc_rh_subscription.handle - self.config = {'rh_subscription': - {'username': 'scooby@do.com', - 'password': 'scooby-snacks' - }} - self.config_full = {'rh_subscription': - {'username': 'scooby@do.com', - 'password': 'scooby-snacks', - 'auto-attach': True, - 'service-level': 'self-support', - 'add-pool': ['pool1', 'pool2', 'pool3'], - 'enable-repo': ['repo1', 'repo2', 'repo3'], - 'disable-repo': ['repo4', 'repo5'] - }} + self.config = { + "rh_subscription": { + "username": "scooby@do.com", + "password": "scooby-snacks", + } + } + self.config_full = { + "rh_subscription": { + "username": "scooby@do.com", + "password": "scooby-snacks", + "auto-attach": True, + "service-level": "self-support", + "add-pool": ["pool1", "pool2", "pool3"], + "enable-repo": ["repo1", "repo2", "repo3"], + "disable-repo": ["repo4", "repo5"], + } + } def test_already_registered(self, m_sman_cli): - ''' + """ Emulates a system that is already registered. Ensure it gets a non-ProcessExecution error from is_registered() - ''' - self.handle(self.name, self.config, self.cloud_init, - self.log, self.args) + """ + self.handle( + self.name, self.config, self.cloud_init, self.log, self.args + ) self.assertEqual(m_sman_cli.call_count, 1) - self.assertIn('System is already registered', self.logs.getvalue()) + self.assertIn("System is already registered", self.logs.getvalue()) def test_simple_registration(self, m_sman_cli): - ''' + """ Simple registration with username and password - ''' - reg = "The system has been registered with ID:" \ - " 12345678-abde-abcde-1234-1234567890abc" - m_sman_cli.side_effect = [subp.ProcessExecutionError, (reg, 'bar')] - self.handle(self.name, self.config, self.cloud_init, - self.log, self.args) - self.assertIn(mock.call(['identity']), m_sman_cli.call_args_list) - self.assertIn(mock.call(['register', '--username=scooby@do.com', - '--password=scooby-snacks'], - logstring_val=True), - m_sman_cli.call_args_list) - self.assertIn('rh_subscription plugin completed successfully', - self.logs.getvalue()) + """ + reg = ( + "The system has been registered with ID:" + " 12345678-abde-abcde-1234-1234567890abc" + ) + m_sman_cli.side_effect = [subp.ProcessExecutionError, (reg, "bar")] + self.handle( + self.name, self.config, self.cloud_init, self.log, self.args + ) + self.assertIn(mock.call(["identity"]), m_sman_cli.call_args_list) + self.assertIn( + mock.call( + [ + "register", + "--username=scooby@do.com", + "--password=scooby-snacks", + ], + logstring_val=True, + ), + m_sman_cli.call_args_list, + ) + self.assertIn( + "rh_subscription plugin completed successfully", + self.logs.getvalue(), + ) self.assertEqual(m_sman_cli.call_count, 2) @mock.patch.object(cc_rh_subscription.SubscriptionManager, "_getRepos") def test_update_repos_disable_with_none(self, m_get_repos, m_sman_cli): cfg = copy.deepcopy(self.config) - m_get_repos.return_value = ([], ['repo1']) - cfg['rh_subscription'].update( - {'enable-repo': ['repo1'], 'disable-repo': None}) + m_get_repos.return_value = ([], ["repo1"]) + cfg["rh_subscription"].update( + {"enable-repo": ["repo1"], "disable-repo": None} + ) mysm = cc_rh_subscription.SubscriptionManager(cfg) self.assertEqual(True, mysm.update_repos()) m_get_repos.assert_called_with() - self.assertEqual(m_sman_cli.call_args_list, - [mock.call(['repos', '--enable=repo1'])]) + self.assertEqual( + m_sman_cli.call_args_list, [mock.call(["repos", "--enable=repo1"])] + ) def test_full_registration(self, m_sman_cli): - ''' + """ Registration with auto-attach, service-level, adding pools, and enabling and disabling yum repos - ''' + """ call_lists = [] - call_lists.append(['attach', '--pool=pool1', '--pool=pool3']) - call_lists.append(['repos', '--disable=repo5', '--enable=repo2', - '--enable=repo3']) - call_lists.append(['attach', '--auto', '--servicelevel=self-support']) - reg = "The system has been registered with ID:" \ - " 12345678-abde-abcde-1234-1234567890abc" + call_lists.append(["attach", "--pool=pool1", "--pool=pool3"]) + call_lists.append( + ["repos", "--disable=repo5", "--enable=repo2", "--enable=repo3"] + ) + call_lists.append(["attach", "--auto", "--servicelevel=self-support"]) + reg = ( + "The system has been registered with ID:" + " 12345678-abde-abcde-1234-1234567890abc" + ) m_sman_cli.side_effect = [ subp.ProcessExecutionError, - (reg, 'bar'), - ('Service level set to: self-support', ''), - ('pool1\npool3\n', ''), ('pool2\n', ''), ('', ''), - ('Repo ID: repo1\nRepo ID: repo5\n', ''), - ('Repo ID: repo2\nRepo ID: repo3\nRepo ID: repo4', ''), - ('', '')] - self.handle(self.name, self.config_full, self.cloud_init, - self.log, self.args) + (reg, "bar"), + ("Service level set to: self-support", ""), + ("pool1\npool3\n", ""), + ("pool2\n", ""), + ("", ""), + ("Repo ID: repo1\nRepo ID: repo5\n", ""), + ("Repo ID: repo2\nRepo ID: repo3\nRepo ID: repo4", ""), + ("", ""), + ] + self.handle( + self.name, self.config_full, self.cloud_init, self.log, self.args + ) self.assertEqual(m_sman_cli.call_count, 9) for call in call_lists: self.assertIn(mock.call(call), m_sman_cli.call_args_list) - self.assertIn("rh_subscription plugin completed successfully", - self.logs.getvalue()) + self.assertIn( + "rh_subscription plugin completed successfully", + self.logs.getvalue(), + ) @mock.patch(SUB_MAN_CLI) @@ -117,38 +144,48 @@ class TestBadInput(CiTestCase): log = logging.getLogger("bad_tests") args = [] SM = cc_rh_subscription.SubscriptionManager - reg = "The system has been registered with ID:" \ - " 12345678-abde-abcde-1234-1234567890abc" - - config_no_password = {'rh_subscription': - {'username': 'scooby@do.com' - }} - - config_no_key = {'rh_subscription': - {'activation-key': '1234abcde', - }} - - config_service = {'rh_subscription': - {'username': 'scooby@do.com', - 'password': 'scooby-snacks', - 'service-level': 'self-support' - }} - - config_badpool = {'rh_subscription': - {'username': 'scooby@do.com', - 'password': 'scooby-snacks', - 'add-pool': 'not_a_list' - }} - config_badrepo = {'rh_subscription': - {'username': 'scooby@do.com', - 'password': 'scooby-snacks', - 'enable-repo': 'not_a_list' - }} - config_badkey = {'rh_subscription': - {'activation-key': 'abcdef1234', - 'fookey': 'bar', - 'org': '123', - }} + reg = ( + "The system has been registered with ID:" + " 12345678-abde-abcde-1234-1234567890abc" + ) + + config_no_password = {"rh_subscription": {"username": "scooby@do.com"}} + + config_no_key = { + "rh_subscription": { + "activation-key": "1234abcde", + } + } + + config_service = { + "rh_subscription": { + "username": "scooby@do.com", + "password": "scooby-snacks", + "service-level": "self-support", + } + } + + config_badpool = { + "rh_subscription": { + "username": "scooby@do.com", + "password": "scooby-snacks", + "add-pool": "not_a_list", + } + } + config_badrepo = { + "rh_subscription": { + "username": "scooby@do.com", + "password": "scooby-snacks", + "enable-repo": "not_a_list", + } + } + config_badkey = { + "rh_subscription": { + "activation-key": "abcdef1234", + "fookey": "bar", + "org": "123", + } + } def setUp(self): super(TestBadInput, self).setUp() @@ -160,75 +197,124 @@ class TestBadInput(CiTestCase): self.assertEqual([], missing, "Missing expected warnings.") def test_no_password(self, m_sman_cli): - '''Attempt to register without the password key/value.''' - m_sman_cli.side_effect = [subp.ProcessExecutionError, - (self.reg, 'bar')] - self.handle(self.name, self.config_no_password, self.cloud_init, - self.log, self.args) + """Attempt to register without the password key/value.""" + m_sman_cli.side_effect = [ + subp.ProcessExecutionError, + (self.reg, "bar"), + ] + self.handle( + self.name, + self.config_no_password, + self.cloud_init, + self.log, + self.args, + ) self.assertEqual(m_sman_cli.call_count, 0) def test_no_org(self, m_sman_cli): - '''Attempt to register without the org key/value.''' + """Attempt to register without the org key/value.""" m_sman_cli.side_effect = [subp.ProcessExecutionError] - self.handle(self.name, self.config_no_key, self.cloud_init, - self.log, self.args) - m_sman_cli.assert_called_with(['identity']) + self.handle( + self.name, self.config_no_key, self.cloud_init, self.log, self.args + ) + m_sman_cli.assert_called_with(["identity"]) self.assertEqual(m_sman_cli.call_count, 1) - self.assert_logged_warnings(( - 'Unable to register system due to incomplete information.', - 'Use either activationkey and org *or* userid and password', - 'Registration failed or did not run completely', - 'rh_subscription plugin did not complete successfully')) + self.assert_logged_warnings( + ( + "Unable to register system due to incomplete information.", + "Use either activationkey and org *or* userid and password", + "Registration failed or did not run completely", + "rh_subscription plugin did not complete successfully", + ) + ) def test_service_level_without_auto(self, m_sman_cli): - '''Attempt to register using service-level without auto-attach key.''' - m_sman_cli.side_effect = [subp.ProcessExecutionError, - (self.reg, 'bar')] - self.handle(self.name, self.config_service, self.cloud_init, - self.log, self.args) + """Attempt to register using service-level without auto-attach key.""" + m_sman_cli.side_effect = [ + subp.ProcessExecutionError, + (self.reg, "bar"), + ] + self.handle( + self.name, + self.config_service, + self.cloud_init, + self.log, + self.args, + ) self.assertEqual(m_sman_cli.call_count, 1) - self.assert_logged_warnings(( - 'The service-level key must be used in conjunction with ', - 'rh_subscription plugin did not complete successfully')) + self.assert_logged_warnings( + ( + "The service-level key must be used in conjunction with ", + "rh_subscription plugin did not complete successfully", + ) + ) def test_pool_not_a_list(self, m_sman_cli): - ''' + """ Register with pools that are not in the format of a list - ''' - m_sman_cli.side_effect = [subp.ProcessExecutionError, - (self.reg, 'bar')] - self.handle(self.name, self.config_badpool, self.cloud_init, - self.log, self.args) + """ + m_sman_cli.side_effect = [ + subp.ProcessExecutionError, + (self.reg, "bar"), + ] + self.handle( + self.name, + self.config_badpool, + self.cloud_init, + self.log, + self.args, + ) self.assertEqual(m_sman_cli.call_count, 2) - self.assert_logged_warnings(( - 'Pools must in the format of a list', - 'rh_subscription plugin did not complete successfully')) + self.assert_logged_warnings( + ( + "Pools must in the format of a list", + "rh_subscription plugin did not complete successfully", + ) + ) def test_repo_not_a_list(self, m_sman_cli): - ''' + """ Register with repos that are not in the format of a list - ''' - m_sman_cli.side_effect = [subp.ProcessExecutionError, - (self.reg, 'bar')] - self.handle(self.name, self.config_badrepo, self.cloud_init, - self.log, self.args) + """ + m_sman_cli.side_effect = [ + subp.ProcessExecutionError, + (self.reg, "bar"), + ] + self.handle( + self.name, + self.config_badrepo, + self.cloud_init, + self.log, + self.args, + ) self.assertEqual(m_sman_cli.call_count, 2) - self.assert_logged_warnings(( - 'Repo IDs must in the format of a list.', - 'Unable to add or remove repos', - 'rh_subscription plugin did not complete successfully')) + self.assert_logged_warnings( + ( + "Repo IDs must in the format of a list.", + "Unable to add or remove repos", + "rh_subscription plugin did not complete successfully", + ) + ) def test_bad_key_value(self, m_sman_cli): - ''' + """ Attempt to register with a key that we don't know - ''' - m_sman_cli.side_effect = [subp.ProcessExecutionError, - (self.reg, 'bar')] - self.handle(self.name, self.config_badkey, self.cloud_init, - self.log, self.args) + """ + m_sman_cli.side_effect = [ + subp.ProcessExecutionError, + (self.reg, "bar"), + ] + self.handle( + self.name, self.config_badkey, self.cloud_init, self.log, self.args + ) self.assertEqual(m_sman_cli.call_count, 1) - self.assert_logged_warnings(( - 'fookey is not a valid key for rh_subscription. Valid keys are:', - 'rh_subscription plugin did not complete successfully')) + self.assert_logged_warnings( + ( + "fookey is not a valid key for rh_subscription. Valid keys" + " are:", + "rh_subscription plugin did not complete successfully", + ) + ) + # vi: ts=4 expandtab |