aDriv4 website!

Linux premium73.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64
LiteSpeed
68.65.122.244

File Name : tester.py

# Copyright 2018 www.privaz.io Valletech AB
# Copyright 2002-2024, OpenNebula Project, OpenNebula Systems
#
# 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.

from hashlib import md5
from json import load, dumps as json_dumps
from base64 import b64decode, b64encode
from pickle import dumps, loads
from os import path
from . import OneServer
from tblib import pickling_support
from sys import exc_info
from six import reraise
from collections import OrderedDict
from gzip import open
from os import environ

pickling_support.install()

def read_fixture_file(fixture_file):
    f = open(fixture_file, "rt")
    ret = load(f)
    f.close()
    return ret

def write_fixture_file(fixture_file, obj):
    f = open(fixture_file, "wb")
    f.write(json_dumps(obj).encode())
    f.close()

class OneServerTester(OneServer):
    '''
    This class extends the OneServer to facilitate unit testing
    The idea is to be able to "record" fixtures while testing with a live OpenNebula platform.
    Those recordings can later be use in "replay" mode to simulate an OpenNebula platform.
    XMLAPI method calls are recorded as test_base/unit_test/method_signature_instance
    The signature is generated as the md5 of the parameters
    if several calls with the same signature are doing during the same unit test, instance is incremented.
    The order in which calls happen within the same unit_test must be deterministic.
    '''
    def __init__(self, uri, session, fixture_file=None, fixture_unit=None, timeout=None, fixture_replay=None, **options):

# Environment driven initialization required for capturing fixtures during ansible integration tests

if fixture_file is None:
            fixture_file = environ.get("PYONE_TEST_FIXTURE_FILE", None)

if fixture_replay is None:
            fixture_replay = (environ.get("PYONE_TEST_FIXTURE_REPLAY", "True").lower() in ["1", "yes", "true"])

if fixture_unit is None:
            fixture_unit = environ.get("PYONE_TEST_FIXTURE_UNIT", "init")

if path.isfile(fixture_file):
            self._fixtures = read_fixture_file(fixture_file)
        else:
            self._fixtures = dict()

# all members involved in the fixtures must be predefined or
        # the magic getter method will trigger resulting in stack overflows
        self._fixture_replay = fixture_replay
        self._fixture_file = fixture_file
        self.set_fixture_unit_test(fixture_unit)

OneServer.__init__(self, uri, session, timeout, **options)

def set_fixture_unit_test(self,name):
        """
        Set the name of the unit test. creates an entry in the fixture tree if it does not exist.
        :param name:
        :return:
        """

if not name in self._fixtures:
            self._fixtures[name] = dict()

self._fixture_unit_test = self._fixtures[name]

def _fixture_signature(self, methodname, params):
        signature_md5 = md5()
        sparms=json_dumps(params,sort_keys=True).encode()
        signature_md5.update(sparms)
        return signature_md5.hexdigest()

def _get_fixture(self, methodname, params):
        '''
        returns the next fixture for a given call.
        :param methodname: XMlRPC method
        :param params: the paramters passed
        :return: file name were to store to or read from the fixture data
        '''

ret = self._fixture_unit_test[methodname][self._fixture_signature(methodname,params)].pop(0)

if not ret:
            raise Exception("Could not read fixture, if the tests changed you must re-record fixtures")

return ret

def _set_fixture(self,methodname,params,object):
        '''
        Will create the fixture for a given call
        :param methodname:
        :param params:
        :return:
        '''

signature = self._fixture_signature(methodname,params)

if not methodname in self._fixture_unit_test:
            self._fixture_unit_test[methodname]=dict()

if not signature in self._fixture_unit_test[methodname]:
            self._fixture_unit_test[methodname][signature]=[]

self._fixture_unit_test[methodname][signature].append(object)

def _do_request(self,method,params):
        '''
        Intercepts requests.
        In record mode they are recorded before being returned.
        In replay mode they are read from fixtures instead
        Exceptions are also captured and reraised
        '''

ret = None

if self._fixture_replay:
            ret = self._get_fixture(method,params)
            if 'exception' in ret:
                reraise(*loads(b64decode(ret['exception'])))
        else:
            try:
                ret = OneServer._do_request(self, method, params)
            except Exception as exception:
                ret = {
                    "exception": b64encode(dumps(exc_info(), 2)).decode(),
                }
                raise exception
            finally:
                self._set_fixture(method,params,ret)
        return ret

def _cast_parms(self,params):
        """
        Parameters will be used to generate the signature of the method, an md5.
        So we need signatures to be deterministic. There are two sources of randomness
        - Python version, in particular differences in dealing with encodings
        - Unsorted sets.
        This method will add casting transformations to fix those, only required for testing.

:param params:
        :return: a list of parameters that should generate a deterministic md5 signature.
        """
        lparams = list(params)

for i, param in enumerate(lparams):
            if isinstance(param, dict):
                lparams[i] = self._to_ordered_dict(param)

return  OneServer._cast_parms(self, lparams)

def _to_ordered_dict(self, param):
        """
        deep orders a dictionary
        :param param:
        :return:
        """

if isinstance(param,dict):
            # ensure the dictionary is ordered
            param = OrderedDict(sorted(param.items()))
            # recurse
            for key, value in param.items():
                if isinstance(value, dict):
                    param[key] = self._to_ordered_dict(value)
        return param

def server_retry_interval(self):
        return 0.01

def server_close(self):
        """
        Unpdates the fixture data if we are recording.
        :return:
        """
        if not self._fixture_replay:
            write_fixture_file(self._fixture_file, self._fixtures)

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/root	0755	2025-04-04 08:01:19
..	dir	-	root/root	0755	2025-05-23 06:37:59
__pycache__	dir	-	root/root	0755	2025-04-04 08:01:21
bindings	dir	-	root/root	0755	2025-04-04 08:01:19
__init__.py	text/x-script.python	10.99 KB	root/root	0644	2025-04-04 08:01:19
acl.py	text/x-script.python	5.8 KB	root/root	0644	2025-04-04 08:01:19
helpers.py	text/x-script.python	3.2 KB	root/root	0644	2025-04-04 08:01:19
server.py	text/x-script.python	929 B	root/root	0644	2025-04-04 08:01:19
tester.py	text/x-script.python	6.87 KB	root/root	0644	2025-04-04 08:01:19
util.py	text/x-script.python	5.47 KB	root/root	0644	2025-04-04 08:01:19

For more Private tool come to me!

https://t.me/aDriv4vz