diff options
Diffstat (limited to 'tests/unittests/cmd/test_cloud_id.py')
| -rw-r--r-- | tests/unittests/cmd/test_cloud_id.py | 258 |
1 files changed, 159 insertions, 99 deletions
diff --git a/tests/unittests/cmd/test_cloud_id.py b/tests/unittests/cmd/test_cloud_id.py index 12fc80e8..907297a6 100644 --- a/tests/unittests/cmd/test_cloud_id.py +++ b/tests/unittests/cmd/test_cloud_id.py @@ -2,126 +2,186 @@ """Tests for cloud-id command line utility.""" -from cloudinit import util from collections import namedtuple -from io import StringIO -from cloudinit.cmd import cloud_id +import pytest -from tests.unittests.helpers import CiTestCase, mock +from cloudinit import util +from cloudinit.cmd import cloud_id +from tests.unittests.helpers import mock +M_PATH = "cloudinit.cmd.cloud_id." -class TestCloudId(CiTestCase): - args = namedtuple('cloudidargs', ('instance_data json long')) +class TestCloudId: - def setUp(self): - super(TestCloudId, self).setUp() - self.tmp = self.tmp_dir() - self.instance_data = self.tmp_path('instance-data.json', dir=self.tmp) + args = namedtuple("cloudidargs", "instance_data json long") def test_cloud_id_arg_parser_defaults(self): """Validate the argument defaults when not provided by the end-user.""" - cmd = ['cloud-id'] - with mock.patch('sys.argv', cmd): + cmd = ["cloud-id"] + with mock.patch("sys.argv", cmd): args = cloud_id.get_parser().parse_args() - self.assertEqual( - '/run/cloud-init/instance-data.json', - args.instance_data) - self.assertEqual(False, args.long) - self.assertEqual(False, args.json) + assert "/run/cloud-init/instance-data.json" == args.instance_data + assert False is args.long + assert False is args.json - def test_cloud_id_arg_parse_overrides(self): + def test_cloud_id_arg_parse_overrides(self, tmpdir): """Override argument defaults by specifying values for each param.""" - util.write_file(self.instance_data, '{}') - cmd = ['cloud-id', '--instance-data', self.instance_data, '--long', - '--json'] - with mock.patch('sys.argv', cmd): + instance_data = tmpdir.join("instance-data.json") + instance_data.write("{}") + cmd = [ + "cloud-id", + "--instance-data", + instance_data.strpath, + "--long", + "--json", + ] + with mock.patch("sys.argv", cmd): args = cloud_id.get_parser().parse_args() - self.assertEqual(self.instance_data, args.instance_data) - self.assertEqual(True, args.long) - self.assertEqual(True, args.json) - - def test_cloud_id_missing_instance_data_json(self): + assert instance_data.strpath == args.instance_data + assert True is args.long + assert True is args.json + + @mock.patch(M_PATH + "get_status_details") + def test_cloud_id_missing_instance_data_json( + self, get_status_details, tmpdir, capsys + ): """Exit error when the provided instance-data.json does not exist.""" - cmd = ['cloud-id', '--instance-data', self.instance_data] - with mock.patch('sys.argv', cmd): - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: - with self.assertRaises(SystemExit) as context_manager: - cloud_id.main() - self.assertEqual(1, context_manager.exception.code) - self.assertIn( - "ERROR: File not found '%s'" % self.instance_data, - m_stderr.getvalue()) - - def test_cloud_id_non_json_instance_data(self): + get_status_details.return_value = cloud_id.UXAppStatus.DONE, "n/a", "" + instance_data = tmpdir.join("instance-data.json") + cmd = ["cloud-id", "--instance-data", instance_data.strpath] + with mock.patch("sys.argv", cmd): + with pytest.raises(SystemExit) as context_manager: + cloud_id.main() + assert 1 == context_manager.value.code + _out, err = capsys.readouterr() + assert "Error:\nFile not found '%s'" % instance_data.strpath in err + + @mock.patch(M_PATH + "get_status_details") + def test_cloud_id_non_json_instance_data( + self, get_status_details, tmpdir, capsys + ): """Exit error when the provided instance-data.json is not json.""" - cmd = ['cloud-id', '--instance-data', self.instance_data] - util.write_file(self.instance_data, '{') - with mock.patch('sys.argv', cmd): - with mock.patch('sys.stderr', new_callable=StringIO) as m_stderr: - with self.assertRaises(SystemExit) as context_manager: - cloud_id.main() - self.assertEqual(1, context_manager.exception.code) - self.assertIn( - "ERROR: File '%s' is not valid json." % self.instance_data, - m_stderr.getvalue()) - - def test_cloud_id_from_cloud_name_in_instance_data(self): + get_status_details.return_value = cloud_id.UXAppStatus.DONE, "n/a", "" + instance_data = tmpdir.join("instance-data.json") + cmd = ["cloud-id", "--instance-data", instance_data.strpath] + instance_data.write("{") + with mock.patch("sys.argv", cmd): + with pytest.raises(SystemExit) as context_manager: + cloud_id.main() + assert 1 == context_manager.value.code + _out, err = capsys.readouterr() + assert ( + "Error:\nFile '%s' is not valid json." % instance_data.strpath + in err + ) + + @mock.patch(M_PATH + "get_status_details") + def test_cloud_id_from_cloud_name_in_instance_data( + self, get_status_details, tmpdir, capsys + ): """Report canonical cloud-id from cloud_name in instance-data.""" - util.write_file( - self.instance_data, - '{"v1": {"cloud_name": "mycloud", "region": "somereg"}}') - cmd = ['cloud-id', '--instance-data', self.instance_data] - with mock.patch('sys.argv', cmd): - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with self.assertRaises(SystemExit) as context_manager: - cloud_id.main() - self.assertEqual(0, context_manager.exception.code) - self.assertEqual("mycloud\n", m_stdout.getvalue()) - - def test_cloud_id_long_name_from_instance_data(self): + instance_data = tmpdir.join("instance-data.json") + get_status_details.return_value = cloud_id.UXAppStatus.DONE, "n/a", "" + instance_data.write( + '{"v1": {"cloud_name": "mycloud", "region": "somereg"}}', + ) + cmd = ["cloud-id", "--instance-data", instance_data.strpath] + with mock.patch("sys.argv", cmd): + with pytest.raises(SystemExit) as context_manager: + cloud_id.main() + assert 0 == context_manager.value.code + out, _err = capsys.readouterr() + assert "mycloud\n" == out + + @mock.patch(M_PATH + "get_status_details") + def test_cloud_id_long_name_from_instance_data( + self, get_status_details, tmpdir, capsys + ): """Report long cloud-id format from cloud_name and region.""" - util.write_file( - self.instance_data, - '{"v1": {"cloud_name": "mycloud", "region": "somereg"}}') - cmd = ['cloud-id', '--instance-data', self.instance_data, '--long'] - with mock.patch('sys.argv', cmd): - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with self.assertRaises(SystemExit) as context_manager: - cloud_id.main() - self.assertEqual(0, context_manager.exception.code) - self.assertEqual("mycloud\tsomereg\n", m_stdout.getvalue()) - - def test_cloud_id_lookup_from_instance_data_region(self): + get_status_details.return_value = cloud_id.UXAppStatus.DONE, "n/a", "" + instance_data = tmpdir.join("instance-data.json") + instance_data.write( + '{"v1": {"cloud_name": "mycloud", "region": "somereg"}}', + ) + cmd = ["cloud-id", "--instance-data", instance_data.strpath, "--long"] + with mock.patch("sys.argv", cmd): + with pytest.raises(SystemExit) as context_manager: + cloud_id.main() + out, _err = capsys.readouterr() + assert 0 == context_manager.value.code + assert "mycloud\tsomereg\n" == out + + @mock.patch(M_PATH + "get_status_details") + def test_cloud_id_lookup_from_instance_data_region( + self, get_status_details, tmpdir, capsys + ): """Report discovered canonical cloud_id when region lookup matches.""" - util.write_file( - self.instance_data, + get_status_details.return_value = cloud_id.UXAppStatus.DONE, "n/a", "" + instance_data = tmpdir.join("instance-data.json") + instance_data.write( '{"v1": {"cloud_name": "aws", "region": "cn-north-1",' - ' "platform": "ec2"}}') - cmd = ['cloud-id', '--instance-data', self.instance_data, '--long'] - with mock.patch('sys.argv', cmd): - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with self.assertRaises(SystemExit) as context_manager: - cloud_id.main() - self.assertEqual(0, context_manager.exception.code) - self.assertEqual("aws-china\tcn-north-1\n", m_stdout.getvalue()) - - def test_cloud_id_lookup_json_instance_data_adds_cloud_id_to_json(self): + ' "platform": "ec2"}}', + ) + cmd = ["cloud-id", "--instance-data", instance_data.strpath, "--long"] + with mock.patch("sys.argv", cmd): + with pytest.raises(SystemExit) as context_manager: + cloud_id.main() + assert 0 == context_manager.value.code + out, _err = capsys.readouterr() + assert "aws-china\tcn-north-1\n" == out + + @mock.patch(M_PATH + "get_status_details") + def test_cloud_id_lookup_json_instance_data_adds_cloud_id_to_json( + self, get_status_details, tmpdir, capsys + ): """Report v1 instance-data content with cloud_id when --json set.""" - util.write_file( - self.instance_data, + get_status_details.return_value = cloud_id.UXAppStatus.DONE, "n/a", "" + instance_data = tmpdir.join("instance-data.json") + instance_data.write( '{"v1": {"cloud_name": "unknown", "region": "dfw",' - ' "platform": "openstack", "public_ssh_keys": []}}') - expected = util.json_dumps({ - 'cloud_id': 'openstack', 'cloud_name': 'unknown', - 'platform': 'openstack', 'public_ssh_keys': [], 'region': 'dfw'}) - cmd = ['cloud-id', '--instance-data', self.instance_data, '--json'] - with mock.patch('sys.argv', cmd): - with mock.patch('sys.stdout', new_callable=StringIO) as m_stdout: - with self.assertRaises(SystemExit) as context_manager: - cloud_id.main() - self.assertEqual(0, context_manager.exception.code) - self.assertEqual(expected + '\n', m_stdout.getvalue()) + ' "platform": "openstack", "public_ssh_keys": []}}', + ) + expected = util.json_dumps( + { + "cloud_id": "openstack", + "cloud_name": "unknown", + "platform": "openstack", + "public_ssh_keys": [], + "region": "dfw", + } + ) + cmd = ["cloud-id", "--instance-data", instance_data.strpath, "--json"] + with mock.patch("sys.argv", cmd): + with pytest.raises(SystemExit) as context_manager: + cloud_id.main() + out, _err = capsys.readouterr() + assert 0 == context_manager.value.code + assert expected + "\n" == out + + @pytest.mark.parametrize( + "status, exit_code", + ( + (cloud_id.UXAppStatus.DISABLED, 2), + (cloud_id.UXAppStatus.NOT_RUN, 3), + (cloud_id.UXAppStatus.RUNNING, 0), + ), + ) + @mock.patch(M_PATH + "get_status_details") + def test_cloud_id_unique_exit_codes_for_status( + self, get_status_details, status, exit_code, tmpdir, capsys + ): + """cloud-id returns unique exit codes for status.""" + get_status_details.return_value = status, "n/a", "" + instance_data = tmpdir.join("instance-data.json") + if status == cloud_id.UXAppStatus.RUNNING: + instance_data.write("{}") + cmd = ["cloud-id", "--instance-data", instance_data.strpath, "--json"] + with mock.patch("sys.argv", cmd): + with pytest.raises(SystemExit) as context_manager: + cloud_id.main() + assert exit_code == context_manager.value.code + # vi: ts=4 expandtab |