1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
# This file is part of cloud-init. See LICENSE file for license information.
"""Tests for cloud-id command line utility."""
from collections import namedtuple
from io import StringIO
from cloudinit import util
from cloudinit.cmd import cloud_id
from tests.unittests.helpers import CiTestCase, mock
class TestCloudId(CiTestCase):
args = namedtuple("cloudidargs", "instance_data json long")
def setUp(self):
super(TestCloudId, self).setUp()
self.tmp = self.tmp_dir()
self.instance_data = self.tmp_path("instance-data.json", dir=self.tmp)
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)
def test_cloud_id_arg_parse_overrides(self):
"""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):
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):
"""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:\nFile not found '%s'" % self.instance_data,
m_stderr.getvalue(),
)
def test_cloud_id_non_json_instance_data(self):
"""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:\nFile '%s' is not valid json." % self.instance_data,
m_stderr.getvalue(),
)
def test_cloud_id_from_cloud_name_in_instance_data(self):
"""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):
"""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):
"""Report discovered canonical cloud_id when region lookup matches."""
util.write_file(
self.instance_data,
'{"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):
"""Report v1 instance-data content with cloud_id when --json set."""
util.write_file(
self.instance_data,
'{"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())
# vi: ts=4 expandtab
|
