/opt/imh-python/lib/python3.9/site-packages/libcloud/test/compute
Edit: /opt/imh-python/lib/python3.9/site-packages/libcloud/test/compute/test_kubevirt.py (8365B)
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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 sys
from libcloud.compute.drivers.kubevirt import KubeVirtNodeDriver
from libcloud.compute.types import NodeState
from libcloud.utils.py3 import httplib
from libcloud.test import unittest
from libcloud.test import MockHttp
from libcloud.test.common.test_kubernetes import KubernetesAuthTestCaseMixin
from libcloud.test.file_fixtures import ComputeFileFixtures
class KubeVirtTestCase(unittest.TestCase, KubernetesAuthTestCaseMixin):
driver_cls = KubeVirtNodeDriver
fixtures = ComputeFileFixtures('kubevirt')
def setUp(self):
KubeVirtNodeDriver.connectionCls.conn_class = KubeVirtMockHttp
self.driver = KubeVirtNodeDriver(key='user',
secret='pass',
secure=True,
host='foo',
port=6443)
def test_list_locations(self):
locations = self.driver.list_locations()
self.assertEqual(len(locations), 5)
self.assertEqual(locations[0].name, 'default')
self.assertEqual(locations[1].name, 'kube-node-lease')
self.assertEqual(locations[2].name, 'kube-public')
self.assertEqual(locations[3].name, 'kube-system')
namespace4 = locations[0].driver.list_locations()[4].name
self.assertEqual(namespace4, 'kubevirt')
id4 = locations[2].driver.list_locations()[4].id
self.assertEqual(id4, 'e6d3d7e8-0ee5-428b-8e17-5187779e5627')
def test_list_nodes(self):
nodes = self.driver.list_nodes()
id0 = "74fd7665-fbd6-4565-977c-96bd21fb785a"
self.assertEqual(len(nodes), 1)
self.assertEqual(nodes[0].extra['namespace'], 'default')
valid_node_states = {NodeState.RUNNING, NodeState.PENDING, NodeState.STOPPED}
self.assertTrue(nodes[0].state in valid_node_states)
self.assertEqual(nodes[0].name, 'testvm')
self.assertEqual(nodes[0].id, id0)
def test_destroy_node(self):
nodes = self.driver.list_nodes()
to_destroy = nodes[-1]
resp = self.driver.destroy_node(to_destroy)
self.assertTrue(resp)
def test_start_node(self):
nodes = self.driver.list_nodes()
r1 = self.driver.start_node(nodes[0])
self.assertTrue(r1)
def test_stop_node(self):
nodes = self.driver.list_nodes()
r1 = self.driver.stop_node(nodes[0])
self.assertTrue(r1)
def test_reboot_node(self):
nodes = self.driver.list_nodes()
for node in nodes:
if node.name == "testvm":
resp = self.driver.reboot_node(node)
self.assertTrue(resp)
class KubeVirtMockHttp(MockHttp):
fixtures = ComputeFileFixtures('kubevirt')
def _api_v1_namespaces(self, method, url, body, headers):
if method == "GET":
body = self.fixtures.load('_api_v1_namespaces.json')
else:
raise AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _apis_kubevirt_io_v1alpha3_namespaces_default_virtualmachines(self,
method, url, body, headers):
if method == "GET":
body = self.fixtures.load('get_default_vms.json')
resp = httplib.OK
elif method == "POST":
body = self.fixtures.load('create_vm.json')
resp = httplib.CREATED
else:
AssertionError('Unsupported method')
return (resp, body, {}, httplib.responses[httplib.OK])
def _apis_kubevirt_io_v1alpha3_namespaces_kube_node_lease_virtualmachines(self,
method, url, body, headers):
if method == "GET":
body = self.fixtures.load('get_kube_node_lease_vms.json')
elif method == "POST":
pass
else:
AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _apis_kubevirt_io_v1alpha3_namespaces_kube_public_virtualmachines(self,
method, url, body, headers):
if method == "GET":
body = self.fixtures.load('get_kube_public_vms.json')
elif method == "POST":
pass
else:
AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _apis_kubevirt_io_v1alpha3_namespaces_kube_system_virtualmachines(self,
method, url, body, headers):
if method == "GET":
body = self.fixtures.load('get_kube_system_vms.json')
elif method == "POST":
pass
else:
AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _apis_kubevirt_io_v1alpha3_namespaces_kubevirt_virtualmachines(self,
method, url, body, headers):
if method == "GET":
body = self.fixtures.load('get_kube_public_vms.json')
elif method == "POST":
pass
else:
AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _apis_kubevirt_io_v1alpha3_namespaces_default_virtualmachines_testvm(self,
method, url, body, headers):
header = "application/merge-patch+json"
data_stop = {"spec": {"running": False}}
data_start = {"spec": {"running": True}}
if method == "PATCH" and headers['Content-Type'] == header and body == data_start:
body = self.fixtures.load('start_testvm.json')
elif method == "PATCH" and headers['Content-Type'] == header and body == data_stop:
body = self.fixtures.load('stop_testvm.json')
else:
AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _apis_kubevirt_io_v1alpha3_namespaces_default_virtualmachines_vm_cirros(self,
method, url, body, headers):
header = "application/merge-patch+json"
data_stop = {"spec": {"running": False}}
data_start = {"spec": {"running": True}}
if method == "PATCH" and headers['Content-Type'] == header and body == data_start:
body = self.fixtures.load('start_vm_cirros.json')
elif method == "PATCH" and headers['Content-Type'] == header and body == data_stop:
body = self.fixtures.load('stop_vm_cirros.json')
else:
AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _apis_kubevirt_io_v1alpha3_namespaces_default_virtualmachineinstances_testvm(self,
method, url, body, headers):
if method == "DELETE":
body = self.fixtures.load('delete_vmi_testvm.json')
else:
AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
def _api_v1_namespaces_default_pods(self, method, url, body, headers):
if method == "GET":
body = self.fixtures.load('get_pods.json')
else:
AssertionError('Unsupported method')
return (httplib.OK, body, {}, httplib.responses[httplib.OK])
if __name__ == '__main__':
sys.exit(unittest.main())