festival.edmaven.com:80

Linux b747.shapiroconsult.com 3.10.0-1160.119.1.vz7.224.4 #1 SMP Mon Sep 30 15:36:27 MSK 2024 x86_64
Apache
173.231.212.80

File Name : test_cloudstack.py

# 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
import unittest

try:
    import simplejson as json
except ImportError:
    import json

from libcloud.utils.py3 import httplib
from libcloud.utils.py3 import urlparse
from libcloud.utils.py3 import b
from libcloud.utils.py3 import parse_qsl

from libcloud.common.cloudstack import CloudStackConnection
from libcloud.common.types import MalformedResponseError

from libcloud.test import MockHttp

async_delay = 0

class CloudStackMockDriver(object):
    host = 'nonexistent.'
    path = '/path'
    async_poll_frequency = 0

name = 'fake'

async_delay = 0

class CloudStackCommonTest(unittest.TestCase):
    def setUp(self):
        CloudStackConnection.conn_class = CloudStackMockHttp
        self.connection = CloudStackConnection('apikey', 'secret',
                                               host=CloudStackMockDriver.host)
        self.connection.poll_interval = 0.0
        self.driver = self.connection.driver = CloudStackMockDriver()

def test_sync_request_bad_response(self):
        self.driver.path = '/bad/response'
        try:
            self.connection._sync_request('fake')
        except Exception as e:
            self.assertTrue(isinstance(e, MalformedResponseError))
            return
        self.assertTrue(False)

def test_sync_request(self):
        self.driver.path = '/sync'
        self.connection._sync_request('fake')

def test_async_request_successful(self):
        self.driver.path = '/async/success'
        result = self.connection._async_request('fake')
        self.assertEqual(result, {'fake': 'result'})

def test_async_request_unsuccessful(self):
        self.driver.path = '/async/fail'
        try:
            self.connection._async_request('fake')
        except Exception as e:
            self.assertEqual(CloudStackMockHttp.ERROR_TEXT, str(e))
            return
        self.assertFalse(True)

def test_async_request_delayed(self):
        global async_delay
        self.driver.path = '/async/delayed'
        async_delay = 2
        self.connection._async_request('fake')
        self.assertEqual(async_delay, 0)

def test_signature_algorithm(self):
        cases = [
            (
                {
                    'command': 'listVirtualMachines'
                }, 'z/a9Y7J52u48VpqIgiwaGUMCso0='
            ), (
                {
                    'command': 'deployVirtualMachine',
                    'name': 'fred',
                    'displayname': 'George',
                    'serviceofferingid': 5,
                    'templateid': 17,
                    'zoneid': 23,
                    'networkids': 42
                }, 'gHTo7mYmadZ+zluKHzlEKb1i/QU='
            ), (
                {
                    'command': 'deployVirtualMachine',
                    'name': 'fred',
                    'displayname': 'George+Ringo',
                    'serviceofferingid': 5,
                    'templateid': 17,
                    'zoneid': 23,
                    'networkids': 42
                }, 'tAgfrreI1ZvWlWLClD3gu4+aKv4='
            )
        ]

connection = CloudStackConnection('fnord', 'abracadabra')
        for case in cases:
            params = connection.add_default_params(case[0])
            self.assertEqual(connection._make_signature(params), b(case[1]))

class CloudStackMockHttp(MockHttp, unittest.TestCase):

ERROR_TEXT = 'ERROR TEXT'

def _response(self, status, result, response):
        return (status, json.dumps(result), {}, response)

def _check_request(self, url):
        url = urlparse.urlparse(url)
        query = dict(parse_qsl(url.query))

self.assertTrue('apiKey' in query)
        self.assertTrue('command' in query)
        self.assertTrue('response' in query)
        self.assertTrue('signature' in query)

self.assertTrue(query['response'] == 'json')

return query

def _bad_response(self, method, url, body, headers):
        self._check_request(url)
        result = {'success': True}
        return self._response(httplib.OK, result, httplib.responses[httplib.OK])

def _sync(self, method, url, body, headers):
        query = self._check_request(url)
        result = {query['command'].lower() + 'response': {}}
        return self._response(httplib.OK, result, httplib.responses[httplib.OK])

def _async_success(self, method, url, body, headers):
        query = self._check_request(url)
        if query['command'].lower() == 'queryasyncjobresult':
            self.assertEqual(query['jobid'], '42')
            result = {
                query['command'].lower() + 'response': {
                    'jobstatus': 1,
                    'jobresult': {'fake': 'result'}
                }
            }
        else:
            result = {query['command'].lower() + 'response': {'jobid': '42'}}
        return self._response(httplib.OK, result, httplib.responses[httplib.OK])

def _async_fail(self, method, url, body, headers):
        query = self._check_request(url)
        if query['command'].lower() == 'queryasyncjobresult':
            self.assertEqual(query['jobid'], '42')
            result = {
                query['command'].lower() + 'response': {
                    'jobstatus': 2,
                    'jobresult': {'errortext': self.ERROR_TEXT}
                }
            }
        else:
            result = {query['command'].lower() + 'response': {'jobid': '42'}}
        return self._response(httplib.OK, result, httplib.responses[httplib.OK])

def _async_delayed(self, method, url, body, headers):
        global async_delay

query = self._check_request(url)
        if query['command'].lower() == 'queryasyncjobresult':
            self.assertEqual(query['jobid'], '42')
            if async_delay == 0:
                result = {
                    query['command'].lower() + 'response': {
                        'jobstatus': 1,
                        'jobresult': {'fake': 'result'}
                    }
                }
            else:
                result = {
                    query['command'].lower() + 'response': {
                        'jobstatus': 0,
                    }
                }
                async_delay -= 1
        else:
            result = {query['command'].lower() + 'response': {'jobid': '42'}}
        return self._response(httplib.OK, result, httplib.responses[httplib.OK])

if __name__ == '__main__':
    sys.exit(unittest.main())

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2025-01-07 04:51:40
..	dir	-	root/root	0755	2025-01-07 04:51:41
__pycache__	dir	-	root/root	0755	2025-01-07 04:51:40
fixtures	dir	-	root/root	0755	2025-01-06 04:11:22
__init__.py	text/plain	781 B	root/root	0644	2025-01-06 04:11:22
test_aliyun.py	text/plain	2.11 KB	root/root	0644	2025-01-06 04:11:22
test_aws.py	text/x-python	13.15 KB	root/root	0644	2025-01-06 04:11:22
test_azure.py	text/plain	2.1 KB	root/root	0644	2025-01-06 04:11:22
test_base.py	text/x-python	4.14 KB	root/root	0644	2025-01-06 04:11:22
test_base_driver.py	text/plain	2.5 KB	root/root	0644	2025-01-06 04:11:22
test_cloudstack.py	text/x-python	7.06 KB	root/root	0644	2025-01-06 04:11:22
test_digitalocean_v2.py	text/plain	4.03 KB	root/root	0644	2025-01-06 04:11:22
test_gandi.py	text/x-python	1.26 KB	root/root	0644	2025-01-06 04:11:22
test_gandi_live.py	text/x-python	1.16 KB	root/root	0644	2025-01-06 04:11:22
test_google.py	text/x-python	16.11 KB	root/root	0644	2025-01-06 04:11:22
test_kubernetes.py	text/x-python	4.85 KB	root/root	0644	2025-01-06 04:11:22
test_nfsn.py	text/plain	2.68 KB	root/root	0644	2025-01-06 04:11:22
test_openstack.py	text/plain	1.79 KB	root/root	0644	2025-01-06 04:11:22
test_openstack_identity.py	text/x-python	42.08 KB	root/root	0644	2025-01-06 04:11:22
test_ovh.py	text/x-python	1.15 KB	root/root	0644	2025-01-06 04:11:22
test_retry_limit.py	text/plain	2.23 KB	root/root	0644	2025-01-06 04:11:22
test_upcloud.py	text/x-python	11.81 KB	root/root	0644	2025-01-06 04:11:22