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

How to use provisioner method in molecule

Best Python code snippet using molecule_python

rest_client.py

Source:rest_client.py

...195    def __init__(self, ips, port, visitor):196        self.visitor = visitor197        self.ips = ips198        self.port = port199    def visit_provisioner(self, token=None):200        r = None201        if self.ips:202            num_of_ips = len(self.ips)203            if num_of_ips > 0:204                for i in range(num_of_ips):205                    ip = self.ips[i]206                    url = 'https://' + ip + ':' + \207                          str(self.port) + '/' + self.visitor.command_str208                    try:209                        if token is None:210                            token = "Unknown"211                        r = self.visitor.visit(url, token)212                        if r:213                            if i != 0:214                                KioxiaProvisioner.switch_path(i)215                            return r216                    except BaseException:217                        continue218                return r219            return r220        return r221class KioxiaProvisioner(object):222    #223    # REST client class that interacts with a specific Provisioner224    # :type ips: str array225    # :param ips: Provisioner management IPs226    # :type cert: str227    # :param cert: KumoScale keystore pem file full path228    #229    mgmt_ips = []230    def __init__(self, ips, cert, token, port=8090):231        self.mgmt_ips = ips232        self.port = port233        self.user = None234        self.token = token235        if cert is None:236            cert = '/etc/kioxia/ssdtoolbox.pem'237        KioxiaProvisioner.mgmt_ips = ips238        self.http = urllib3.PoolManager(239            cert_reqs=ssl.CERT_NONE,240            cert_file=cert,241            assert_hostname=False,242            timeout=urllib3.Timeout(243                connect=5.0,244                read=60.0))245    def set_token(self, user, token):246        self.user = user247        self.token = token248    def result_support(self, result):249        if result is not None:250            if result.data is not None:251                if "Status 401" in str(result.data):252                    ClassBuilder()253                    return entities.ProvisionerResponse(254                        None, None, "Bad credentials")255                if "Status 403" in str(result.data):256                    return entities.ProvisionerResponse(257                        None, None, "Access is denied")258                if str(result.data) == "":259                    return entities.ProvisionerResponse([], None, "Success")260                try:261                    result_data = json.loads(result.data)262                    if ('status' in result_data and263                            result_data['status'] != "Success"):264                        return entities.ProvisionerResponse(265                            result_data, None, result_data['status'],266                            result_data['description'])267                    return entities.ProvisionerResponse(result_data)268                except Exception as e:269                    return entities.ProvisionerResponse(270                        None, None, type(e).__name__, e.message)271        return entities.ProvisionerResponse(272            None,273            None,274            "Provisioner Communication Error",275            "Provisioner Communication Error")276    # Call to switch last successful connected ip277    @staticmethod278    def switch_path(ip_idx):279        temp = KioxiaProvisioner.mgmt_ips[0]280        KioxiaProvisioner.mgmt_ips[0] = KioxiaProvisioner.mgmt_ips[ip_idx]281        KioxiaProvisioner.mgmt_ips[ip_idx] = temp282    # Call Provisioner with get request283    def provisioner_get_request(self, api_name):284        get_visitor = ProvisionerGetVisitor(self.http, api_name)285        provisioner_connector = ProvisionerConnector(286            self.mgmt_ips, self.port, get_visitor)287        r = provisioner_connector.visit_provisioner(self.token)288        return self.result_support(r)289    # Call Provisioner with delete request290    def provisioner_delete_request(self, api_name):291        delete_visitor = ProvisionerDeleteVisitor(self.http, api_name)292        provisioner_connector = ProvisionerConnector(293            self.mgmt_ips, self.port, delete_visitor)294        r = provisioner_connector.visit_provisioner(self.token)295        return self.result_support(r)296    # Call Provisioner with patch request297    def provisioner_patch_request(self, api_name, json_body=None):298        patch_visitor = ProvisionerPatchVisitor(self.http, api_name, json_body)299        provisioner_connector = ProvisionerConnector(300            self.mgmt_ips, self.port, patch_visitor)301        r = provisioner_connector.visit_provisioner(self.token)302        return self.result_support(r)303    # Call Provisioner with update request304    def provisioner_put_request(self, api_name, json_body):305        put_visitor = ProvisionerPutVisitor(self.http, api_name, json_body)306        provisioner_connector = ProvisionerConnector(307            self.mgmt_ips, self.port, put_visitor)308        r = provisioner_connector.visit_provisioner(self.token)309        return self.result_support(r)310    # Call Provisioner with post request311    def provisioner_post_request(self, api_name, json_body, password=None):312        post_visitor = ProvisionerPostVisitor(self.http, api_name, json_body)313        provisioner_connector = ProvisionerConnector(314            KioxiaProvisioner.mgmt_ips, self.port, post_visitor)315        r = provisioner_connector.visit_provisioner(self.token)316        return self.result_support(r)317    def get_info(self):318        # Call to Get Info API319        # @rtype: ProvisionerResponse320        # @returns: Provisioner response data contain Provisioner information321        #322        result_response = self.provisioner_get_request('info')323        if result_response.status == "Success":324            result_entity = JsonToClass(result_response.prov_entities, True)325            return entities.ProvisionerResponse(result_entity)326        return result_response327    def get_provisioner_info(self):328        # Call to Get Info API329        # @rtype: ProvisionerResponse...

test_provisioning.py

Source:test_provisioning.py

...127    kernel_dir = pjoin(paths.jupyter_data_dir(), 'kernels', name)128    os.makedirs(kernel_dir)129    with open(pjoin(kernel_dir, 'kernel.json'), 'w') as f:130        f.write(json.dumps(spec))131def new_provisioner():132    build_kernelspec('new_provisioner', 'new-test-provisioner')133def custom_provisioner():134    build_kernelspec('custom_provisioner', 'custom-test-provisioner')135@pytest.fixture136def all_provisioners():137    build_kernelspec('no_provisioner')138    build_kernelspec('missing_provisioner', 'missing-provisioner')139    build_kernelspec('default_provisioner', 'local-provisioner')140    build_kernelspec('subclassed_provisioner', 'subclassed-test-provisioner')141    custom_provisioner()142@pytest.fixture(143    params=[144        'no_provisioner',145        'default_provisioner',146        'missing_provisioner',147        'custom_provisioner',148        'subclassed_provisioner',149    ]150)151def akm(request, all_provisioners):152    return AsyncKernelManager(kernel_name=request.param)153initial_provisioner_map = {154    'local-provisioner': ('jupyter_client.provisioning', 'LocalProvisioner'),155    'subclassed-test-provisioner': (156        'jupyter_client.tests.test_provisioning',157        'SubclassedTestProvisioner',158    ),159    'custom-test-provisioner': ('jupyter_client.tests.test_provisioning', 'CustomTestProvisioner'),160}161def mock_get_all_provisioners() -> List[EntryPoint]:162    result = []163    for name, epstr in initial_provisioner_map.items():164        result.append(EntryPoint(name, epstr[0], epstr[1]))165    return result166def mock_get_provisioner(factory, name) -> EntryPoint:167    if name == 'new-test-provisioner':168        return EntryPoint(169            'new-test-provisioner', 'jupyter_client.tests.test_provisioning', 'NewTestProvisioner'170        )171    if name in initial_provisioner_map:172        return EntryPoint(name, initial_provisioner_map[name][0], initial_provisioner_map[name][1])173    raise NoSuchEntryPoint(KernelProvisionerFactory.GROUP_NAME, name)174@pytest.fixture175def kpf(monkeypatch):176    """Setup the Kernel Provisioner Factory, mocking the entrypoint fetch calls."""177    monkeypatch.setattr(178        KernelProvisionerFactory, '_get_all_provisioners', mock_get_all_provisioners179    )180    monkeypatch.setattr(KernelProvisionerFactory, '_get_provisioner', mock_get_provisioner)181    factory = KernelProvisionerFactory.instance()182    return factory183class TestDiscovery:184    def test_find_all_specs(self, kpf, all_provisioners):185        ksm = KernelSpecManager()186        kernels = ksm.get_all_specs()187        # Ensure specs for initial provisioners exist,188        # and missing_provisioner & new_provisioner don't189        assert 'no_provisioner' in kernels190        assert 'default_provisioner' in kernels191        assert 'subclassed_provisioner' in kernels192        assert 'custom_provisioner' in kernels193        assert 'missing_provisioner' not in kernels194        assert 'new_provisioner' not in kernels195    def test_get_missing(self, all_provisioners):196        ksm = KernelSpecManager()197        with pytest.raises(NoSuchKernel):198            ksm.get_kernel_spec('missing_provisioner')199    def test_get_new(self, kpf):200        new_provisioner()  # Introduce provisioner after initialization of KPF201        ksm = KernelSpecManager()202        kernel = ksm.get_kernel_spec('new_provisioner')203        assert 'new-test-provisioner' == kernel.metadata['kernel_provisioner']['provisioner_name']204class TestRuntime:205    async def akm_test(self, kernel_mgr):206        """Starts a kernel, validates the associated provisioner's config, shuts down kernel """207        assert kernel_mgr.provisioner is None208        if kernel_mgr.kernel_name == 'missing_provisioner':209            with pytest.raises(NoSuchKernel):210                await kernel_mgr.start_kernel()211        else:212            await kernel_mgr.start_kernel()213            TestRuntime.validate_provisioner(kernel_mgr)214            await kernel_mgr.shutdown_kernel()215            assert kernel_mgr.provisioner.has_process is False216    @pytest.mark.asyncio217    async def test_existing(self, kpf, akm):218        await self.akm_test(akm)219    @pytest.mark.asyncio220    async def test_new(self, kpf):221        new_provisioner()  # Introduce provisioner after initialization of KPF222        new_km = AsyncKernelManager(kernel_name='new_provisioner')223        await self.akm_test(new_km)224    @pytest.mark.asyncio225    async def test_custom_lifecycle(self, kpf):226        custom_provisioner()227        async_km = AsyncKernelManager(kernel_name='custom_provisioner')228        await async_km.start_kernel(stdout=PIPE, stderr=PIPE)229        is_alive = await async_km.is_alive()230        assert is_alive231        await async_km.restart_kernel(now=True)232        is_alive = await async_km.is_alive()233        assert is_alive234        await async_km.interrupt_kernel()235        assert isinstance(async_km, AsyncKernelManager)236        await async_km.shutdown_kernel(now=True)237        is_alive = await async_km.is_alive()238        assert is_alive is False239        assert async_km.context.closed240    @pytest.mark.asyncio241    async def test_default_provisioner_config(self, kpf, all_provisioners):242        kpf.default_provisioner_name = 'custom-test-provisioner'243        async_km = AsyncKernelManager(kernel_name='no_provisioner')244        await async_km.start_kernel(stdout=PIPE, stderr=PIPE)245        is_alive = await async_km.is_alive()246        assert is_alive247        assert isinstance(async_km.provisioner, CustomTestProvisioner)248        assert async_km.provisioner.config_var_1 == 0  # Not in kernelspec, so default of 0 exists249        await async_km.shutdown_kernel(now=True)250        is_alive = await async_km.is_alive()251        assert is_alive is False252        assert async_km.context.closed253    @staticmethod254    def validate_provisioner(akm: AsyncKernelManager):255        # Ensure the provisioner is managing a process at this point256        assert akm.provisioner is not None and akm.provisioner.has_process257        # Validate provisioner config258        if akm.kernel_name in ['no_provisioner', 'default_provisioner']:259            assert not hasattr(akm.provisioner, 'config_var_1')260            assert not hasattr(akm.provisioner, 'config_var_2')261        else:262            assert akm.provisioner.config_var_1 == 42263            assert akm.provisioner.config_var_2 == akm.kernel_name264        # Validate provisioner class265        if akm.kernel_name in ['no_provisioner', 'default_provisioner', 'subclassed_provisioner']:266            assert isinstance(akm.provisioner, LocalProvisioner)267            if akm.kernel_name == 'subclassed_provisioner':268                assert isinstance(akm.provisioner, SubclassedTestProvisioner)...

factory.py

Source:factory.py

...98        """99        is_available = True100        if provisioner_name not in self.provisioners:101            try:102                ep = self._get_provisioner(provisioner_name)103                self.provisioners[provisioner_name] = ep  # Update cache104            except NoSuchEntryPoint:105                is_available = False106        return is_available107    def _get_provisioner_config(self, kernel_spec: Any) -> Dict[str, Any]:108        """109        Return the kernel_provisioner stanza from the kernel_spec.110        Checks the kernel_spec's metadata dictionary for a kernel_provisioner entry.111        If found, it is returned, else one is created relative to the DEFAULT_PROVISIONER112        and returned.113        Parameters114        ----------115        kernel_spec : Any - this is a KernelSpec type but listed as Any to avoid circular import116            The kernel specification object from which the provisioner dictionary is derived.117        Returns118        -------119        dict120            The provisioner portion of the kernel_spec.  If one does not exist, it will contain121            the default information.  If no `config` sub-dictionary exists, an empty `config`122            dictionary will be added.123        """124        env_provisioner = kernel_spec.metadata.get('kernel_provisioner', {})125        if 'provisioner_name' in env_provisioner:  # If no provisioner_name, return default126            if (127                'config' not in env_provisioner128            ):  # if provisioner_name, but no config stanza, add one129                env_provisioner.update({"config": {}})130            return env_provisioner  # Return what we found (plus config stanza if necessary)131        return {"provisioner_name": self.default_provisioner_name, "config": {}}132    def get_provisioner_entries(self) -> Dict[str, str]:133        """134        Returns a dictionary of provisioner entries.135        The key is the provisioner name for its entry point.  The value is the colon-separated136        string of the entry point's module name and object name.137        """138        entries = {}139        for name, ep in self.provisioners.items():140            entries[name] = f"{ep.module_name}:{ep.object_name}"141        return entries142    @staticmethod143    def _get_all_provisioners() -> List[EntryPoint]:144        """Wrapper around entrypoints.get_group_all() - primarily to facilitate testing."""145        return get_group_all(KernelProvisionerFactory.GROUP_NAME)146    def _get_provisioner(self, name: str) -> EntryPoint:147        """Wrapper around entrypoints.get_single() - primarily to facilitate testing."""148        try:149            ep = get_single(KernelProvisionerFactory.GROUP_NAME, name)150        except NoSuchEntryPoint:151            # Check if the entrypoint name is 'local-provisioner'.  Although this should never152            # happen, we have seen cases where the previous distribution of jupyter_client has153            # remained which doesn't include kernel-provisioner entrypoints (so 'local-provisioner'154            # is deemed not found even though its definition is in THIS package).  In such cass,155            # the entrypoints package uses what it first finds - which is the older distribution156            # resulting in a violation of a supposed invariant condition.  To address this scenario,157            # we will log a warning message indicating this situation, then build the entrypoint158            # instance ourselves - since we have that information.159            if name == 'local-provisioner':160                distros = glob.glob(f"{path.dirname(path.dirname(__file__))}-*")...

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.