diff options
Diffstat (limited to 'tests/unittests/cmd')
| -rw-r--r-- | tests/unittests/cmd/test_cloud_id.py | 203 | ||||
| -rw-r--r-- | tests/unittests/cmd/test_status.py | 12 |
2 files changed, 129 insertions, 86 deletions
diff --git a/tests/unittests/cmd/test_cloud_id.py b/tests/unittests/cmd/test_cloud_id.py index 42941d4f..907297a6 100644 --- a/tests/unittests/cmd/test_cloud_id.py +++ b/tests/unittests/cmd/test_cloud_id.py @@ -3,123 +3,143 @@ """Tests for cloud-id command line utility.""" from collections import namedtuple -from io import StringIO + +import pytest from cloudinit import util from cloudinit.cmd import cloud_id -from tests.unittests.helpers import CiTestCase, mock +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): 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, "{}") + instance_data = tmpdir.join("instance-data.json") + instance_data.write("{}") cmd = [ "cloud-id", "--instance-data", - self.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] + 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 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:\nFile not found '%s'" % self.instance_data, - m_stderr.getvalue(), - ) - - def test_cloud_id_non_json_instance_data(self): + 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, "{") + 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 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:\nFile '%s' is not valid json." % self.instance_data, - m_stderr.getvalue(), + 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 ) - def test_cloud_id_from_cloud_name_in_instance_data(self): + @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, + 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", self.instance_data] + cmd = ["cloud-id", "--instance-data", instance_data.strpath] 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): + 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, + 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", self.instance_data, "--long"] + cmd = ["cloud-id", "--instance-data", instance_data.strpath, "--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): + 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"] + cmd = ["cloud-id", "--instance-data", instance_data.strpath, "--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): + 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": []}}', ) @@ -132,13 +152,36 @@ class TestCloudId(CiTestCase): "region": "dfw", } ) - cmd = ["cloud-id", "--instance-data", self.instance_data, "--json"] + 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 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()) + with pytest.raises(SystemExit) as context_manager: + cloud_id.main() + assert exit_code == context_manager.value.code # vi: ts=4 expandtab diff --git a/tests/unittests/cmd/test_status.py b/tests/unittests/cmd/test_status.py index acd1fea5..17d27597 100644 --- a/tests/unittests/cmd/test_status.py +++ b/tests/unittests/cmd/test_status.py @@ -89,7 +89,7 @@ class TestStatus(CiTestCase): ) def test__is_cloudinit_disabled_true_on_kernel_cmdline(self): - """When using systemd and disable_file is present return disabled.""" + """When kernel command line disables cloud-init return True.""" (is_disabled, reason) = wrap_and_call( "cloudinit.cmd.status", { @@ -107,9 +107,9 @@ class TestStatus(CiTestCase): ) def test__is_cloudinit_disabled_true_when_generator_disables(self): - """When cloud-init-generator doesn't write enabled file return True.""" - enabled_file = os.path.join(self.paths.run_dir, "enabled") - self.assertFalse(os.path.exists(enabled_file)) + """When cloud-init-generator writes disabled file return True.""" + disabled_file = os.path.join(self.paths.run_dir, "disabled") + ensure_file(disabled_file) (is_disabled, reason) = wrap_and_call( "cloudinit.cmd.status", {"uses_systemd": True, "get_cmdline": "something"}, @@ -137,7 +137,7 @@ class TestStatus(CiTestCase): ) def test_status_returns_not_run(self): - """When status.json does not exist yet, return 'not run'.""" + """When status.json does not exist yet, return 'not-run'.""" self.assertFalse( os.path.exists(self.status_file), "Unexpected status.json found" ) @@ -154,7 +154,7 @@ class TestStatus(CiTestCase): cmdargs, ) self.assertEqual(0, retcode) - self.assertEqual("status: not run\n", m_stdout.getvalue()) + self.assertEqual("status: not-run\n", m_stdout.getvalue()) def test_status_returns_disabled_long_on_presence_of_disable_file(self): """When cloudinit is disabled, return disabled reason.""" |