FM / compute

/opt/imh-python/lib/python3.9/site-packages/libcloud/test/compute/

/opt/imh-python/lib/python3.9/site-packages/libcloud/test/compute

Name	Size	Mode	Actions
fixtures/	-	0755	rm
__pycache__/	-	0755	rm
test_abiquo.py	20919	0644	edit dl rm
test_auroracompute.py	2189	0644	edit dl rm
test_azure.py	22213	0644	edit dl rm
test_azure_arm.py	30598	0644	edit dl rm
test_base.py	4073	0644	edit dl rm
test_bluebox.py	4662	0644	edit dl rm
test_brightbox.py	15764	0644	edit dl rm
test_bsnl.py	1283	0644	edit dl rm
test_cloudscale.py	4540	0644	edit dl rm
test_cloudsigma_v1_0.py	8298	0644	edit dl rm
test_cloudsigma_v2_0.py	28023	0644	edit dl rm
test_cloudstack.py	57399	0644	edit dl rm
test_cloudwatt.py	1349	0644	edit dl rm
test_deployment.py	29057	0644	edit dl rm
test_digitalocean_v2.py	24670	0644	edit dl rm
test_dimensiondata_v2_3.py	158501	0644	edit dl rm
test_dimensiondata_v2_4.py	163733	0644	edit dl rm
test_ec2.py	92953	0644	edit dl rm
test_ecp.py	5111	0644	edit dl rm
test_ecs.py	47770	0644	edit dl rm
test_elasticstack.py	11032	0644	edit dl rm
test_exoscale.py	1147	0644	edit dl rm
test_gandi.py	14657	0644	edit dl rm
test_gce.py	190676	0644	edit dl rm
test_gig_g8.py	6826	0644	edit dl rm
test_gogrid.py	11435	0644	edit dl rm
test_gridscale.py	17208	0644	edit dl rm
test_gridspot.py	7682	0644	edit dl rm
test_hostvirtual.py	8678	0644	edit dl rm
test_ikoula.py	1139	0644	edit dl rm
test_indosat.py	1311	0644	edit dl rm
test_internetsolutions.py	1371	0644	edit dl rm
test_joyent.py	5462	0644	edit dl rm
test_kamatera.py	10284	0644	edit dl rm
test_kili.py	1525	0644	edit dl rm
test_ktucloud.py	4554	0644	edit dl rm
test_kubevirt.py	8365	0644	edit dl rm
test_libvirt_driver.py	3730	0644	edit dl rm
test_linode.py	8778	0644	edit dl rm
test_maxihost.py	4306	0644	edit dl rm
test_medone.py	1295	0644	edit dl rm
test_nephoscale.py	7378	0644	edit dl rm
test_ntta.py	1323	0644	edit dl rm
test_nttcis.py	148380	0644	edit dl rm
test_onapp.py	6175	0644	edit dl rm
test_oneandone.py	42388	0644	edit dl rm
test_opennebula.py	45456	0644	edit dl rm
test_openstack.py	125148	0644	edit dl rm
test_ovh.py	10616	0644	edit dl rm
test_packet.py	21367	0644	edit dl rm
test_profitbricks.py	124039	0644	edit dl rm
test_rackspace.py	8015	0644	edit dl rm
test_rimuhosting.py	4300	0644	edit dl rm
test_scaleway.py	13663	0644	edit dl rm
test_softlayer.py	12521	0644	edit dl rm
test_ssh_client.py	24225	0644	edit dl rm
test_types.py	3202	0644	edit dl rm
test_upcloud.py	12056	0644	edit dl rm
test_vcl.py	4821	0644	edit dl rm
test_vcloud.py	53032	0644	edit dl rm
test_voxel.py	6129	0644	edit dl rm
test_vpsnet.py	8648	0644	edit dl rm
test_vultr.py	6580	0644	edit dl rm
__init__.py	4077	0644	edit dl rm

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())

cmd: