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
|
# This file is part of cloud-init. See LICENSE file for license information.
"""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
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: File 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: File '%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
|
