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

How to use get_upgrade_status method in localstack

Best Python code snippet using localstack_python

test_boto3_elasticsearch.py

Source:test_boto3_elasticsearch.py

...879            "ResponseMetadata": None,880        }881        with patch.object(self.conn, "get_upgrade_status", return_value=ret_val):882            self.assertEqual(883                boto3_elasticsearch.get_upgrade_status(884                    domain_name="testdomain", **CONN_PARAMETERS885                ),886                {"result": True, "response": ret_val},887            )888    def test_get_upgrade_status_error(self):889        """890        Test that when calling get_upgrade_status and boto3891        returns an error, it returns {'result': False, 'error': 'the error'}.892        """893        with patch.object(894            self.conn,895            "get_upgrade_status",896            side_effect=ClientError(ERROR_CONTENT, "get_upgrade_status"),897        ):898            result = boto3_elasticsearch.get_upgrade_status(899                domain_name="testdomain", **CONN_PARAMETERS900            )901            self.assertFalse(result["result"])902            self.assertEqual(903                result.get("error", ""), ERROR_MESSAGE.format(101, "get_upgrade_status")904            )905    def test_list_domain_names_positive(self):906        """907        Test that when calling list_domain_names and it908        succeeds, it returns {'result': True, 'response': some_value}.909        """910        ret_val = {"DomainNames": [{"DomainName": "string"}]}911        with patch.object(self.conn, "list_domain_names", return_value=ret_val):912            self.assertEqual(...

pss_upgrade.py

Source:pss_upgrade.py

...133            return134        return True135    def __str__(self):136        return self.desired_status137def get_upgrade_status(module):138    command = 'config soft upgrade status'139    response = run_command(module, command)140    responses = [response, 'executed: %s' % command]141    return (responses, parser.parse_upgrade_status(responses[0]))142def execute_command(module, command, **kwargs):143    conditional = kwargs.get('conditional')144    timeout = kwargs.get('timeout', 0)145    interval = kwargs.get('interval', 1)146    if command == 'commit':147        cmd_line = 'config soft upgrade commit'148    elif command == 'abort':149        cmd_line = 'config soft upgrade abort'150    elif command == 'backout':151        cmd_line = 'config soft upgrade backout yes'152    elif command == 'activate':153        cmd_line = 'config soft upgrade manual activate yes'154    elif command == 'audit':155        cmd_line = 'config soft upgrade manual audit {release} {audit_option}'.format(**kwargs)156    elif command == 'load':157        cmd_line = 'config soft upgrade manual load'158    else:159        cmd_line = ''160    responses = []161    if cmd_line:162        responses.append('executed command: %s' % cmd_line)163        responses.append(run_command(module, cmd_line))164        success = False165        if conditional:166            for _ in range(timeout):167                _, status = get_upgrade_status(module)168                if conditional(status):169                    success = True170                    break171                time.sleep(interval)172            if not success:173                msg = 'The condition (%s) has not been satisfied' % conditional174                module.fail_json(msg=msg, failed_conditions=[])175    return responses176def main():177    """main entry point for module execution178    """179    argument_spec = dict(180        command=dict(choices=['status', 'abort', 'commit', 'manual', 'backout', 'auto'], required=True),181        manual=dict(choices=['activate', 'audit', 'load']),182        release=dict(type='str'),183        audit_option=dict(choices=['force', 'nobackup', 'nobackupforce']),184        wait_for=dict(185            type='dict', aliases=['waitfor'],186            options=dict(187                operation=dict(choices=['Abort', 'Backout', 'Commit', 'Load', 'Activate']),188                operation_status=dict(choices=['Completed', 'In Progress', 'Failure']),189                committed_release=dict(type='str'),190                stdout=dict(type='str'),191                )192        ),193        wait_timeout=dict(default=18000, type='int'),194        check_interval=dict(default=10, type='int'),195    )196    module = AnsibleModule(argument_spec=argument_spec, supports_check_mode=True)197    warnings = list()198    result = {'changed': False, 'warnings': warnings}199    conditional = None200    wait_for = module.params['wait_for']201    if wait_for:202        try:203            conditional = StatusConditional(wait_for)204        except AttributeError as exc:205            module.fail_json(msg=to_text(exc))206    wait_timeout = module.params['wait_timeout']207    check_interval = module.params['check_interval']208    command = module.params['command'].lower()209    kwargs = dict(conditional=conditional,210                  timeout=wait_timeout,211                  interval=check_interval)212    commands = []213    responses = []214    if command == 'status':215        responses, status = get_upgrade_status(module)216        result['upgrade_status'] = status217    else:218        _, status = get_upgrade_status(module)219        if status['operation_status'] == 'Failure' and command != 'backout':220            module.fail_json(msg='upgrade operation %s is %s' % (status['operation'], status['operation_status']))221        if command == 'commit':222            if status['operation'] == 'Activate' or status['operation'] == 'Commit':223                if status['operation'] == 'Commit':224                    warnings.append('software (%s) has already been committed' % status['committed_release'])225            else:226                module.fail_json(msg='cannot execute commit when in state (%s, %s)' % (status['operation'], status['operation_status']))227        elif command == 'manual':228            option = module.params['manual']229            release = module.params['release']230            if option == 'audit':231                if not release:232                    module.fail_json(msg='audit operation requires release option')...

urls.py

Source:urls.py

1"""assets_discovery URL Configuration2The `urlpatterns` list routes URLs to views. For more information please see:3    https://docs.djangoproject.com/en/1.10/topics/http/urls/4Examples:5Function views6    1. Add an import:  from my_app import views7    2. Add a URL to urlpatterns:  url(r'^$', views.home, name='home')8Class-based views9    1. Add an import:  from other_app.views import Home10    2. Add a URL to urlpatterns:  url(r'^$', Home.as_view(), name='home')11Including another URLconf12    1. Import the include() function: from django.conf.urls import url, include13    2. Add a URL to urlpatterns:  url(r'^blog/', include('blog.urls'))14"""15from django.conf.urls import url16from django.contrib import admin17from search import search,result18from asset import asset19from threats import threats,pocs20from scan import scan_management,scan_settings,system_info,target,update,system,port21from panels import panels22from report import chart,report23__scan__ = scan_management.Scan()24__update__ = system.update()25__port__ = port.Port()26__portgroup__ = port.PortGroup()27urlpatterns = [28    url(r'^panels$', panels.panels),29    url(r'^panels/get_count_info$', panels.get_count_info),30    url(r'^asset$', asset.asset),31    url(r'^asset/edit$', asset.edit),32    url(r'^asset/delete_index$', asset.delete_index),33    url(r'^asset/export$', asset.export),34    url(r'^asset/export_csv$', asset.download),35    url(r'^asset/get_subdomain_body$', asset.get_subdomain_body),36    url(r'^report$', report.report),37    url(r'^report/new$', report.new),38    url(r'^report/edit$', report.edit),39    url(r'^report/add_chart$', chart.add_chart),40    url(r'^report/chart_delete$', chart.chart_delete),41    url(r'^report/manual_report$', report.manual_report),42    url(r'^report/detail$', report.detail),43    url(r'^report/report_detail$', report.report_detail),44    url(r'^report/delete$', report.delete),45    url(r'^search$', search.search),46    url(r'^result$', result.result, name='search'),47    url(r'^result/download$', result.download),48    url(r'^result/count$', result.count),49    url(r'^result/page$', result.page),50    url(r'^result/get_host_content$', result.get_host_content),51    url(r'^threats$', threats.threats,name='threats'),52    url(r'^threats/detail$', threats.detail),53    url(r'^threats/clear$', threats.clear),54    url(r'^threats/get_vuls$', threats.get_vuls),55    url(r'^threats/exp_info$', threats.exp_info),56    url(r'^threats/poc_exp$', threats.poc_exp),57    url(r'^threats/poc_vertify_progress$', threats.poc_vertify_progress),58    url(r'^threats/export$', threats.export),59    url(r'^pocs$', pocs.pocs),60    url(r'^pocs/scan_start$', pocs.scan_start),61    url(r'^pocs/pocs_scan_status$', pocs.pocs_scan_status),62    url(r'^scancfgs$', __scan__.asset_scan),63    url(r'^scan/get_scan_status$', __scan__.get_scan_status),64    url(r'^scan/start_scan$', __scan__.start_scan),65    url(r'^scan/stop_scan$', __scan__.stop_scan),66    url(r'^network_settings$', scan_settings.network_settings),67    url(r'^scan_settings$', scan_settings.scan_settings),68    url(r'^fgap_settings', scan_settings.fgap_settings),69    url(r'^system_info$', system_info.info),70    url(r'^systems/get_systems$', system_info.get_systems),71    url(r'^port_management$', __port__.port_management),72    url(r'^port_management/new$', __port__.add_port),73    url(r'^port_management/edit$', __port__.edit_port),74    url(r'^port_management/delete$', __port__.delete_port),75    url(r'^port_management/get_port_group$', __port__.get_port_group),76    url(r'^port_group_management$', __portgroup__.port_group_management),77    url(r'^port_group_management/new$', __portgroup__.add_port_group),78    url(r'^port_group_management/edit$', __portgroup__.edit),79    url(r'^port_group_management/delete$', __portgroup__.delete),80    url(r'^port_group_management/get_port$', __portgroup__.get_port),81    url(r'^target_management$', target.management),82    url(r'^target_management/new$', target.new),83    url(r'^target_management/edit$', target.edit),84    url(r'^target_management/delete$', target.delete),85    url(r'^system/update$', update.update),86    url(r'^system/upgrade$', update.upgrade),87    url(r'^system/get_upgrade_status$', __update__.get_upgrade_status),88    url(r'^system/modify_upgrade_url$', update.modify_upgrade_url),89    url(r'^system/compare_versions$', __update__.compare_versions),90    url(r'^system/download_file$', __update__.download_file),91    url(r'^system/download_progress$', __update__.download_progress),92    url(r'^$', panels.panels, name="main")...

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.