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
|
# VMware vSphere Python SDK tests
#
# Copyright (c) 2017-2025 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 ssl
import unittest
import tests
from pyVmomi import pbm, VmomiSupport, SoapStubAdapter
from pyVim.connect import SmartConnect
class PBMTests(tests.VCRTestBase):
def get_pbm_connection(self, vpxd_stub):
VmomiSupport.GetRequestContext()["vcSessionCookie"] = \
vpxd_stub.cookie.split('"')[1]
hostname = vpxd_stub.host.split(":")[0]
pbm_stub = SoapStubAdapter(
host=hostname,
version="pbm.version.version11",
path="/pbm/sdk",
poolSize=0,
sslContext=ssl._create_unverified_context())
pbm_si = pbm.ServiceInstance("ServiceInstance", pbm_stub)
pbm_content = pbm_si.RetrieveContent()
return pbm_content
def get_profile(self, profile_name, pbm_content):
pm = pbm_content.profileManager
profile_ids = pm.PbmQueryProfile(
resourceType=pbm.profile.ResourceType(resourceType="STORAGE"),
profileCategory="REQUIREMENT")
if len(profile_ids) > 0:
profiles = pm.PbmRetrieveContent(profileIds=profile_ids)
for profile in profiles:
if profile_name in profile.name:
return profile
raise Exception('Profile not found')
@tests.VCRTestBase.my_vcr.use_cassette('pbm_check_compatibility.yaml',
cassette_library_dir=tests.fixtures_path,
record_mode='once')
def test_pbm_check_compatibility(self):
si = SmartConnect(host='vcsa',
user='my_user',
pwd='my_password',
disableSslCertValidation=True)
# Connect to SPBM Endpoint
pbm_content = self.get_pbm_connection(si._stub)
sp = self.get_profile("test-profile", pbm_content)
pbm_content.placementSolver.PbmCheckCompatibility(profile=sp.profileId)
if __name__ == '__main__':
unittest.main()
|
