From a25bdf7233a6d1397cff8442ea304161e4d42303 Mon Sep 17 00:00:00 2001 From: Alex Schultz Date: Wed, 9 Dec 2020 13:35:07 -0700 Subject: [PATCH] Add validation around stream We don't want to allow for stream repos when we don't have a stream host. Additionally we don't want to disable stream repos when we have a stream host. Both of these are incompatible so let's add better checking when tripleo-repos are run. Change-Id: I0740becea0ba60bfdd7b3256f99136d063f4b7bf --- tripleo_repos/main.py | 36 ++++++++++++++++------ tripleo_repos/tests/test_main.py | 53 +++++++++++++++++++++----------- 2 files changed, 62 insertions(+), 27 deletions(-) diff --git a/tripleo_repos/main.py b/tripleo_repos/main.py index 8237bff..373391d 100755 --- a/tripleo_repos/main.py +++ b/tripleo_repos/main.py @@ -94,9 +94,12 @@ class NoRepoTitle(Exception): def _get_distro(): + """Get distro info from os-release + returns: distro_id, distro_major_version_id, distro_name + """ output = subprocess.Popen( - 'source /etc/os-release && echo -e -n "$ID\n$VERSION_ID"', + 'source /etc/os-release && echo -e -n "$ID\n$VERSION_ID\n$NAME"', shell=True, stdout=subprocess.PIPE, stderr=open(os.devnull, 'w'), @@ -104,7 +107,7 @@ def _get_distro(): universal_newlines=True).communicate() # distro_id and distro_version_id will always be at least an empty string - distro_id, distro_version_id = output[0].split('\n') + distro_id, distro_version_id, distro_name = output[0].split('\n') # if distro_version_id is empty string the major version will be empty # string too @@ -118,12 +121,10 @@ def _get_distro(): distro_id = 'centos' distro_major_version_id = '7' - return distro_id, distro_major_version_id + return distro_id, distro_major_version_id, distro_name -def _parse_args(): - - distro_id, distro_major_version_id = _get_distro() +def _parse_args(distro_id, distro_major_version_id): distro = "{}{}".format(distro_id, distro_major_version_id) @@ -264,10 +265,26 @@ def _validate_tripleo_ci_testing(repos): return True -def _validate_args(args): +def _validate_distro_stream(args, distro_name): + """Validate stream related args vs host + + Fails if stream is to be used but the host isn't a stream OS or vice versa + """ + if args.stream and 'stream' not in distro_name.lower(): + raise InvalidArguments('--stream provided, but OS is not the Stream ' + 'version. Please ensure the host is Stream.') + elif args.no_stream and 'stream' in distro_name.lower(): + raise InvalidArguments('--no-stream provided, but OS is the Stream ' + 'version. Please ensure the host is not the ' + 'Stream version.') + return True + + +def _validate_args(args, distro_name): _validate_current_tripleo(args.repos) _validate_distro_repos(args) _validate_tripleo_ci_testing(args.repos) + _validate_distro_stream(args, distro_name) def _remove_existing(args): @@ -440,8 +457,9 @@ def _run_pkg_clean(distro): def main(): - args = _parse_args() - _validate_args(args) + distro_id, distro_major_version_id, distro_name = _get_distro() + args = _parse_args(distro_id, distro_major_version_id) + _validate_args(args, distro_name) base_path = _get_base_path(args) if args.distro in ['centos7']: _install_priorities() diff --git a/tripleo_repos/tests/test_main.py b/tripleo_repos/tests/test_main.py index 28bfd13..4794c04 100644 --- a/tripleo_repos/tests/test_main.py +++ b/tripleo_repos/tests/test_main.py @@ -24,6 +24,7 @@ from tripleo_repos import main @ddt.ddt class TestTripleORepos(testtools.TestCase): + @mock.patch('tripleo_repos.main._get_distro') @mock.patch('sys.argv', ['tripleo-repos', 'current', '-d', 'centos7']) @mock.patch('tripleo_repos.main._run_pkg_clean') @mock.patch('tripleo_repos.main._validate_args') @@ -32,18 +33,20 @@ class TestTripleORepos(testtools.TestCase): @mock.patch('tripleo_repos.main._remove_existing') @mock.patch('tripleo_repos.main._install_repos') def test_main(self, mock_install, mock_remove, mock_ip, mock_gbp, - mock_validate, mock_clean): - args = main._parse_args() + mock_validate, mock_clean, mock_distro): + mock_distro.return_value = ('centos', '8', 'CentOS 8') + args = main._parse_args('centos', '8') mock_path = mock.Mock() mock_gbp.return_value = mock_path main.main() - mock_validate.assert_called_once_with(args) + mock_validate.assert_called_once_with(args, 'CentOS 8') mock_gbp.assert_called_once_with(args) mock_ip.assert_called_once_with() mock_remove.assert_called_once_with(args) mock_install.assert_called_once_with(args, mock_path) mock_clean.assert_called_once_with('centos7') + @mock.patch('tripleo_repos.main._get_distro') @mock.patch('sys.argv', ['tripleo-repos', 'current', '-d', 'fedora']) @mock.patch('tripleo_repos.main._run_pkg_clean') @mock.patch('tripleo_repos.main._validate_args') @@ -52,12 +55,13 @@ class TestTripleORepos(testtools.TestCase): @mock.patch('tripleo_repos.main._remove_existing') @mock.patch('tripleo_repos.main._install_repos') def test_main_fedora(self, mock_install, mock_remove, mock_ip, mock_gbp, - mock_validate, mock_clean): - args = main._parse_args() + mock_validate, mock_clean, mock_distro): + mock_distro.return_value = ('centos', '8', 'CentOS 8') + args = main._parse_args('centos', '8') mock_path = mock.Mock() mock_gbp.return_value = mock_path main.main() - mock_validate.assert_called_once_with(args) + mock_validate.assert_called_once_with(args, 'CentOS 8') mock_gbp.assert_called_once_with(args) assert not mock_ip.called, '_install_priorities should no tbe called' mock_remove.assert_called_once_with(args) @@ -498,7 +502,7 @@ enabled=1 with mock.patch.object(sys, 'argv', ['', 'current', 'deps', '-d', 'centos7', '-b', 'liberty', '-o', 'test']): - args = main._parse_args() + args = main._parse_args('centos', '8') self.assertEqual(['current', 'deps'], args.repos) self.assertEqual('centos7', args.distro) self.assertEqual('liberty', args.branch) @@ -509,7 +513,7 @@ enabled=1 'centos7', '--branch', 'mitaka', '--output-path', 'test']): - args = main._parse_args() + args = main._parse_args('centos', '8') self.assertEqual(['current'], args.repos) self.assertEqual('centos7', args.distro) self.assertEqual('mitaka', args.branch) @@ -641,51 +645,64 @@ class TestValidate(testtools.TestCase): self.args.repos = ['current'] self.args.branch = 'master' self.args.distro = 'centos7' + self.args.stream = False + self.args.no_stream = False def test_good(self): - main._validate_args(self.args) + main._validate_args(self.args, '') def test_current_and_tripleo_dev(self): self.args.repos = ['current', 'current-tripleo-dev'] self.assertRaises(main.InvalidArguments, main._validate_args, - self.args) + self.args, '') def test_tripleo_ci_testing_and_current_tripleo(self): self.args.repos = ['current-tripleo', 'tripleo-ci-testing'] self.assertRaises(main.InvalidArguments, main._validate_args, - self.args) + self.args, '') def test_tripleo_ci_testing_and_ceph_opstools_allowed(self): self.args.repos = ['ceph', 'opstools', 'tripleo-ci-testing'] - main._validate_args(self.args) + main._validate_args(self.args, '') def test_tripleo_ci_testing_and_deps_allowed(self): self.args.repos = ['deps', 'tripleo-ci-testing'] - main._validate_args(self.args) + main._validate_args(self.args, '') def test_ceph_and_tripleo_dev(self): self.args.repos = ['current-tripleo-dev', 'ceph'] self.args.output_path = main.DEFAULT_OUTPUT_PATH - main._validate_args(self.args) + main._validate_args(self.args, '') def test_deps_and_tripleo_dev(self): self.args.repos = ['deps', 'current-tripleo-dev'] self.assertRaises(main.InvalidArguments, main._validate_args, - self.args) + self.args, '') def test_current_and_tripleo(self): self.args.repos = ['current', 'current-tripleo'] self.assertRaises(main.InvalidArguments, main._validate_args, - self.args) + self.args, '') def test_deps_and_tripleo_allowed(self): self.args.repos = ['deps', 'current-tripleo'] - main._validate_args(self.args) + main._validate_args(self.args, '') def test_invalid_distro(self): self.args.distro = 'Jigawatts 1.21' self.assertRaises(main.InvalidArguments, main._validate_args, - self.args) + self.args, '') + + def test_invalid_stream(self): + self.args.stream = True + self.assertRaises(main.InvalidArguments, main._validate_args, + self.args, 'CentOS 8') + + def test_invalid_no_stream(self): + self.args.stream = False + self.args.no_stream = True + self.assertRaises(main.InvalidArguments, main._validate_args, + self.args, 'CentOS 8 Stream') def test_validate_distro_repos(self): self.assertTrue(main._validate_distro_repos(self.args))