KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use test_provisioner method in molecule

Best Python code snippet using molecule_python

conftest.py

Source:conftest.py

1"""Configuration and fixtures for pytest."""2from kqueen.config import current_config3from kqueen.models import Cluster4from kqueen.models import Organization5from kqueen.models import Provisioner6from kqueen.models import User7from kqueen.server import create_app8import datetime9import etcd10import faker11import json12import pytest13import random14import string15import uuid16import yaml17config = current_config()18fake = faker.Faker()19current_app = None20class ClusterFixture:21    def __init__(self, test_provisioner=None, test_user=None):22        test_user = test_user if test_user is not None else UserFixture()23        if not test_provisioner:24            test_provisioner = ProvisionerFixture(test_user=test_user)25        self.test_provisioner = test_provisioner26        owner = test_user.obj27        provisioner = self.test_provisioner.obj28        provisioner.state = config.get('PROVISIONER_OK_STATE')29        provisioner.save(check_status=False)30        _uuid = uuid.uuid4()31        create_kwargs = {32            'id': _uuid,33            'name': 'Name for cluster {}'.format(_uuid),34            'provisioner': provisioner,35            'state': config.get('CLUSTER_UNKNOWN_STATE'),36            'kubeconfig': yaml.load(open('kubeconfig_localhost', 'r').read()),37            'created_at': datetime.datetime.utcnow().replace(microsecond=0),38            'owner': owner,39            'metadata': {}40        }41        self.obj = Cluster.create(owner.namespace, **create_kwargs)42    def destroy(self):43        try:44            self.obj.delete()45            self.test_provisioner.destroy()46        except Exception:47            # Provisioner may be already deleted48            pass49class ProvisionerFixture:50    def __init__(self, test_user=None):51        if test_user is None:52            test_user = UserFixture()53        self.test_user = test_user54        owner = test_user.obj55        create_kwargs = {56            'name': 'Fixtured provisioner',57            'engine': 'kqueen.engines.ManualEngine',58            'owner': owner59        }60        self.obj = Provisioner.create(owner.namespace, **create_kwargs)61    def destroy(self):62        try:63            self.obj.delete()64            self.test_user.destroy()65        except Exception:66            # Provisioner may be already deleted67            pass68class UserFixture:69    def __init__(self, namespace=None):70        profile = fake.simple_profile()71        self.test_org = OrganizationFixture(namespace)72        user = User.create(73            None,74            username=profile['username'],75            password=profile['username'] + 'password',76            organization=self.test_org.obj,77            role='superadmin',78            active=True79        )80        user.save(validate=False)81        self.obj = user82    def destroy(self):83        try:84            self.obj.delete()85            self.test_org.destroy()86        except Exception:87            # User be may already deleted88            pass89class OrganizationFixture:90    def __init__(self, name=None):91        """Prepare organization object."""92        if not name:93            name = ''.join(random.choice(string.ascii_letters) for _ in range(8))94        organization = Organization(95            None,96            name=name,97            namespace=name98        )99        organization.save(validate=False)100        self.obj = organization101    def destroy(self):102        try:103            self.obj.delete()104        except Exception:105            # Organization be may already deleted106            pass107class AuthHeader:108    def __init__(self, test_user=None):109        self.user = test_user if test_user is not None else UserFixture()110    def get(self, client):111        """112        Get JWT access token and convert it to HTTP header.113        Args:114            client: Flask client115        Returns:116            dict: {'Authorization': 'JWT access_token'}117        """118        _user = self.user.obj119        token = get_auth_token(client, _user)120        return {121            'Authorization': '{token_prefix} {token}'.format(122                token_prefix=config.get('JWT_AUTH_HEADER_PREFIX'),123                token=token,124            ),125            'X-Test-Namespace': _user.namespace,126            'X-User': str(_user.id),127        }128    def destroy(self):129        self.user.destroy()130class UserWithNamespaceFixture:131    def __init__(self):132        self.test_org = OrganizationFixture()133        org = self.test_org.obj134        namespace = fake.user_name()135        org.namespace = namespace136        org.save()137        self.test_user = UserFixture()138        _user = self.test_user.obj139        _user.organization = org140        _user.save()141        self.obj = _user142    def destroy(self):143        self.test_org.destroy()144        self.test_user.destroy()145@pytest.fixture(autouse=True, scope='session')146def app():147    """Prepare app."""148    global current_app149    current_app = create_app()150    current_app.testing = True151    return current_app152@pytest.fixture(autouse=True, scope='class')153def etcd_setup():154    global current_app155    try:156        current_app.db.client.delete(current_app.config['ETCD_PREFIX'], recursive=True)157    except etcd.EtcdKeyNotFound:158        pass159@pytest.fixture160def cluster():161    """Create cluster with manual provisioner."""162    test_cluster = ClusterFixture()163    yield test_cluster.obj164    test_cluster.destroy()165@pytest.fixture166def provisioner():167    """Create dummy manual provisioner."""168    test_provisioner = ProvisionerFixture()169    test_provisioner.obj.save()170    yield test_provisioner.obj171    test_provisioner.destroy()172def get_auth_token(_client, _user):173    """174    Acquire token for given user175    Args:176        client: Client connection177        user: User object178    Returns:179        str: Auth token.180    """181    data = {182        'username': _user.username,183        'password': _user.username + 'password'184    }185    response = _client.post(186        '/api/v1/auth',187        data=json.dumps(data),188        content_type='application/json')189    try:190        token = response.json['access_token']191    except KeyError as e:192        raise KeyError('Unable to read access token from response: {}'.format(response.data))193    return token194@pytest.fixture(scope='class', autouse=True)195def user():196    """Prepare user object."""197    with current_app.app_context():198        test_org = OrganizationFixture()199        profile = fake.simple_profile()200        user = User.create(201            None,202            username=profile['username'],203            password=profile['username'] + 'password',204            organization=test_org.obj,205            role='superadmin',206            active=True207        )208        user.save()209        yield user210        try:211            user.delete()212            test_org.destroy()213        except Exception:214            # objects may be already deleted...

test_epumgmt_main_em_core_findworkers.py

Source:test_epumgmt_main_em_core_findworkers.py

1import epumgmt.main.em_core_findworkers2import epumgmt.api3from epumgmt.api.exceptions import IncompatibleEnvironment4from epumgmt.defaults.runlogs import DefaultRunlogs5import epumgmt.main.em_args as em_args6from epumgmt.defaults.parameters import DefaultParameters7from mocks.common import FakeCommon8from mocks.modules import FakeModules9from mocks.event import Event10from mocks.runlogs import FakeRunlogs11def test_get_provisioner():12    from epumgmt.main.em_core_findworkers import _get_provisioner13    m = FakeModules()14    run_name = "TESTRUN"15    m.persistence.store_run_vms(run_name, [])16    try:17        _get_provisioner(m, run_name)18        raised_incompatible_env = False19    except IncompatibleEnvironment:20        raised_incompatible_env = True21    assert raised_incompatible_env22    non_provisioner_vm = epumgmt.api.RunVM()23    non_provisioner_vm.service_type = "something"24    m.persistence.store_run_vms(run_name, [non_provisioner_vm])25    try:26        _get_provisioner(m, run_name)27        raised_incompatible_env = False28    except IncompatibleEnvironment:29        raised_incompatible_env = True30    assert raised_incompatible_env31    test_service_name = "provisioner"32    test_provisioner = epumgmt.api.RunVM()33    test_provisioner.service_type = test_service_name34    test_provisioner_instanceid = "i-TEST"35    test_provisioner.instanceid = test_provisioner_instanceid36    m.persistence.store_run_vms(run_name, [non_provisioner_vm, test_provisioner])37    got_provisioner = _get_provisioner(m, run_name)38    assert got_provisioner == test_provisioner39def test_vms_launched():40    from epumgmt.main.em_core_findworkers import vms_launched41    run_name = "TESTRUN"42    c = FakeCommon()43    m = FakeModules()44    optdict = {}45    optdict[em_args.NAME.name] = run_name46    p = DefaultParameters(None, None)47    p.optdict = optdict48    m.runlogs = FakeRunlogs()49    test_service_name = "provisioner"50    test_provisioner = epumgmt.api.RunVM()51    test_provisioner.service_type = test_service_name52    test_provisioner_instanceid = "i-TEST"53    test_provisioner.instanceid = test_provisioner_instanceid54    m.persistence.store_run_vms(run_name, [test_provisioner])55    got_vms = vms_launched(m, run_name, "new_node")56    assert got_vms == []57    vm_0_id = "i-apple"58    vm_0_nodeid = "applenodeid"59    vm_0_publicip = "8.8.8.8"60    vm_0_new_node_event = Event(name="new_node", iaas_id=vm_0_id,61                       node_id=vm_0_nodeid, public_ip=vm_0_publicip)62    test_provisioner.events.append(vm_0_new_node_event)63    got_vms = vms_launched(m, run_name, "new_node")64    assert got_vms[0].instanceid == vm_0_id65    test_provisioner.events = []66    got_vms = vms_launched(m, run_name, "new_node")67    assert got_vms == []68    vm_0_node_started_event = Event(name="node_started", iaas_id=vm_0_id,69                                   node_id=vm_0_nodeid, public_ip=vm_0_publicip)70    test_provisioner.events.append(vm_0_node_started_event)71    72    got_vms = vms_launched(m, run_name, "node_started")73    assert got_vms[0].instanceid == vm_0_id74    test_provisioner.events = []75    got_vms = vms_launched(m, run_name, "new_node")76    assert got_vms == []77    vm_0_bad_event = Event(name="bad", iaas_id=vm_0_id,78                                   node_id=vm_0_nodeid, public_ip=vm_0_publicip)79    test_provisioner.events.append(vm_0_bad_event)80    81    try:82        got_vms = vms_launched(m, run_name, "bad")83        raised_incompatible_env = False84    except IncompatibleEnvironment:85        raised_incompatible_env = True86    assert raised_incompatible_env...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.