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
|
# -*- 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 vcr.stubs import VCRHTTPSConnection
from vcr import config
from pyVmomi import SoapAdapter
from pyVmomi import SoapStubAdapter
from pyVmomi import vim
from pyVmomi.VmomiSupport import GetRequestContext
class SerializerTests(tests.VCRTestBase):
def test_serialize_object(self):
val = vim.vm.device.VirtualDeviceSpec.FileOperation()
# This line should not raise an exception, especially on Python 3.
SoapAdapter.Serialize(val)
def test_serialize_integer(self):
lp = vim.LongPolicy()
lp.inherited = False
lp.value = 100
SoapAdapter.Serialize(lp, version='vim.version.version10')
def test_serialize_float(self):
pc = vim.host.VsanInternalSystem.PolicyCost()
pc.diskSpaceToAddressSpaceRatio = 1.0
SoapAdapter.Serialize(pc, version='vim.version.version10')
def test_serialize_unicode(self):
self.assertEqual(SoapAdapter.SerializeToStr('Ḃ'),
u'<obj versionId="" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns="urn:vim25" xsi:type="xsd:string">\u1e02</obj>')
self.assertEqual(SoapAdapter.SerializeToStr(u'Ḃ'),
u'<obj versionId="" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns="urn:vim25" xsi:type="xsd:string">\u1e02</obj>')
self.assertEqual(SoapAdapter.SerializeToStr(u'\u1e02'),
u'<obj versionId="" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns="urn:vim25" xsi:type="xsd:string">\u1e02</obj>')
def _base_serialize_test(self, soap_creator, request_matcher):
my_vcr = config.VCR(
custom_patches=(
(SoapAdapter, 'HTTPSConnection', VCRHTTPSConnection),))
my_vcr.register_matcher('request_matcher', request_matcher)
with my_vcr.use_cassette(
'test_simple_request_serializer.yaml',
cassette_library_dir=tests.fixtures_path,
record_mode='none',
match_on=['request_matcher']) as cass:
stub = soap_creator()
si = vim.ServiceInstance("ServiceInstance", stub)
content = si.RetrieveContent()
self.assertTrue(content is not None)
self.assertTrue(
'<_this type="ServiceInstance">ServiceInstance</_this>'
in cass.requests[0].body.decode("utf-8"))
def _body_request_matcher(self, r1, r2):
soap_msg = ('<soapenv:Body>'
'<RetrieveServiceContent xmlns="urn:vim25">'
'<_this versionId="5.5" type="ServiceInstance">'
'ServiceInstance'
'</_this>'
'</RetrieveServiceContent>'
'</soapenv:Body>')
if soap_msg in r1.body.decode("utf-8"):
return True
raise SystemError('serialization error occurred')
def _request_context_request_matcher(self, r1, r2):
request_context = ('<soapenv:Header><vcSessionCookie versionId="5.5">123456789</vcSessionCookie></soapenv:Header>')
if request_context in r1.body.decode("utf-8"):
return True
raise SystemError('serialization error occurred')
def test_simple_request_serializer(self):
def soap_creator():
return SoapStubAdapter('vcsa', 443)
self._base_serialize_test(soap_creator, self._body_request_matcher)
def test_request_context_serializer_instance(self):
def request_matcher(r1, r2):
return self._request_context_request_matcher(r1, r2) and self._body_request_matcher(r1, r2)
def soap_creator():
return SoapStubAdapter('vcsa', 443, requestContext={'vcSessionCookie': '123456789'})
self._base_serialize_test(soap_creator, request_matcher)
def test_request_context_serializer_global(self):
def request_matcher(r1, r2):
return self._request_context_request_matcher(r1, r2) and self._body_request_matcher(r1, r2)
def soap_creator():
return SoapStubAdapter('vcsa', 443)
GetRequestContext()['vcSessionCookie'] = '123456789'
try:
self._base_serialize_test(soap_creator, request_matcher)
finally:
GetRequestContext().pop("vcSessionCookie")
|
