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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
|
# -*- coding: utf-8 -*-
# VMware vSphere Python SDK tests
#
# Copyright (c) 2008-2024 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import tests
from pyVim import connect
from pyVmomi import vim
from pyVmomi.VmomiJSONEncoder import VmomiJSONEncoder, templateOf
import atexit
import inspect
import json
import os
class JSONTests(tests.VCRTestBase):
@property
def datacenter(self):
return getattr(vim, 'Datacenter')('datacenter-3', self.si._stub)
@property
def datastore(self):
return getattr(vim, 'Datastore')('datastore-17', self.si._stub)
@property
def host(self):
return getattr(vim, 'HostSystem')('host-14', self.si._stub)
@property
def network(self):
return getattr(vim, 'Network')('network-13', self.si._stub)
@property
def si(self):
if not hasattr(self, '_si'):
self._si = connect.SmartConnect(
host='vcenter', user='my_user', pwd='my_pass', disableSslCertValidation=True)
atexit.register(connect.Disconnect, self._si)
return self._si
@property
def vm(self):
return getattr(vim, 'VirtualMachine')('vm-19', self.si._stub)
@property
def vm2(self):
return getattr(vim, 'VirtualMachine')('vm-20', self.si._stub)
def expect(self, data):
"""
Handle results expectation. If the expect data file does not
exist we write one out. Otherwise we read and compare. This
means it is the responsibility of the individual creating the
.expect files to ensure they are correct.
"""
testname = inspect.stack()[1][3]
expectfile = 'tests/files/{0}.expect'.format(testname)
if os.path.exists(expectfile):
with open(expectfile, 'r') as file:
expectdata = file.read()
self.assertEqual(data, expectdata)
else:
with open(expectfile, 'w') as file:
file.write(data)
# NOTE: sort_keys is needed for expect comparison to work;
# not required for consumption
# Explodes the VM
# By definition if the input is a ManagedObject then it explodes.
@tests.VCRTestBase.my_vcr.use_cassette(
'test_json_vm_explode_default.yaml',
cassette_library_dir=tests.fixtures_path, record_mode='once', decode_compressed_response=True)
def test_json_vm_explode_default(self):
raw = json.dumps(self.vm, cls=VmomiJSONEncoder, sort_keys=True)
self.expect(raw)
# Basic sanity check
data = json.loads(raw)
self.assertEqual(data['_vimid'], 'vm-19')
self.assertEqual(data['_vimref'], 'vim.VirtualMachine:vm-19')
self.assertEqual(data['_vimtype'], 'vim.VirtualMachine')
self.assertEqual(data['overallStatus'], 'green')
self.assertEqual(len(data['network']), 0)
self.assertEqual(len(data['capability']['dynamicProperty']), 0)
self.assertIsNone(data['capability']['dynamicType'])
# Explodes the VM disabling the dynamic field stripping
# By definition if the input is a ManagedObject then it explodes.
@tests.VCRTestBase.my_vcr.use_cassette(
'test_json_vm_explode_strip_dynamic.yaml',
cassette_library_dir=tests.fixtures_path, record_mode='once', decode_compressed_response=True)
def test_json_vm_explode_strip_dynamic(self):
raw = json.dumps(self.vm, cls=VmomiJSONEncoder, sort_keys=True,
strip_dynamic=True)
self.expect(raw)
# Basic sanity check
data = json.loads(raw)
self.assertEqual(data['_vimid'], 'vm-19')
self.assertEqual(data['_vimref'], 'vim.VirtualMachine:vm-19')
self.assertEqual(data['_vimtype'], 'vim.VirtualMachine')
self.assertEqual(data['overallStatus'], 'green')
self.assertEqual(len(data['network']), 0)
self.assertTrue('dynamicProperty' not in data['capability'])
self.assertTrue('dynamicType' not in data['capability'])
# Explodes the VM and the VM's networks
# Here self.vm is redundant (see above) but not harmful.
@tests.VCRTestBase.my_vcr.use_cassette(
'test_json_vm_explode_objs.yaml',
cassette_library_dir=tests.fixtures_path, record_mode='once', decode_compressed_response=True)
def test_json_vm_explode_objs_match(self):
to_explode = [self.vm]
for item in self.vm.network:
to_explode.append(item)
self.expect(json.dumps(self.vm, cls=VmomiJSONEncoder, sort_keys=True,
explode=to_explode))
# Explodes by type: all VirtualMachine and all of its snapshots
@tests.VCRTestBase.my_vcr.use_cassette(
'test_json_vm_explode_type.yaml',
cassette_library_dir=tests.fixtures_path, record_mode='once', decode_compressed_response=True)
def test_json_vm_explode_type_match(self):
self.expect(json.dumps([self.vm, self.vm2], cls=VmomiJSONEncoder,
sort_keys=True,
explode=[templateOf('VirtualMachine'),
templateOf('VirtualMachineSnapshot')]))
# Test Datacenter
@tests.VCRTestBase.my_vcr.use_cassette(
'test_json_datacenter_explode.yaml',
cassette_library_dir=tests.fixtures_path, record_mode='once', decode_compressed_response=True)
def test_json_datacenter_explode(self):
self.expect(json.dumps(self.datacenter, cls=VmomiJSONEncoder,
sort_keys=True))
# Test Datastore
@tests.VCRTestBase.my_vcr.use_cassette(
'test_json_datastore_explode.yaml',
cassette_library_dir=tests.fixtures_path, record_mode='once', decode_compressed_response=True)
def test_json_datastore_explode(self):
self.expect(json.dumps(self.datastore, cls=VmomiJSONEncoder,
sort_keys=True))
# Test HostSystem
@tests.VCRTestBase.my_vcr.use_cassette(
'test_json_host_explode.yaml',
cassette_library_dir=tests.fixtures_path, record_mode='once', decode_compressed_response=True)
def test_json_host_explode(self):
self.expect(json.dumps(self.host, cls=VmomiJSONEncoder,
sort_keys=True))
# Test Network
@tests.VCRTestBase.my_vcr.use_cassette(
'test_json_network_explode.yaml',
cassette_library_dir=tests.fixtures_path, record_mode='once', decode_compressed_response=True)
def test_json_network_explode(self):
self.expect(json.dumps(self.network, cls=VmomiJSONEncoder,
sort_keys=True))
|
