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

How to use list_port_forwardings method in tempest

Best Python code snippet using tempest_python

manage.py

Source:manage.py

...72                if port not in ports:73                    ports.append(port)74                    break75    return ports76def list_port_forwardings():77    command = 'sudo iptables -L PREROUTING -t nat'78    output = check_output(command, shell=True).decode()79    port_forwardings = []80    for line in output.splitlines():81        if line.startswith('REDIRECT'):82            part_with_forwarding = line[line.find('tcp dpt:') + len(('tcp dpt:')):]83            parts = part_with_forwarding.partition(' redir ports ')84            source_port = {'http': '80', 'https': '443'}[parts[0]]85            destination_port = parts[2]86            port_forwardings.append([source_port, destination_port])87    return port_forwardings88def create_port_forwarding(source, destination):89    logger.info(f'Forwarding {source} â {destination}...')90    command = 'sudo iptables ' \91        '-t nat ' \92        '-A PREROUTING ' \93        '-i eth0 ' \94        '-p tcp ' \95        f'--destination-port {source} ' \96        '-j REDIRECT ' \97        f'--to-port {destination}'98    logger.debug(command)99    check_call(command, shell=True)100def delete_port_forwarding(source, destination):101    logger.info(f'Removing forwarding {source} â {destination}')102    command = 'sudo iptables ' \103        '-t nat ' \104        '-D PREROUTING ' \105        '-i eth0 ' \106        '-p tcp ' \107        f'--destination-port {source} ' \108        '-j REDIRECT ' \109        f'--to-port {destination}'110    logger.debug(command)111    check_call(command, shell=True)112def get_prepared_project(name):113    project = Project(name=name)114    project.prepare()115    return project116def main(argv):117    category_argument_parser = ArgumentParser()118    category_argument_parser.add_argument('command')119    namespace, arguments = category_argument_parser.parse_known_args(argv[1:])120    command_switch = {121        'run': lambda name, version: get_prepared_project(name).run_release(version),122        'deploy': lambda name, version: get_prepared_project(name).deploy_release(version),123        'update': lambda name, version: get_prepared_project(name).update_release(version),124        'sync': lambda name: get_prepared_project(name).synchronize_with_production(),125        'certs': lambda name: get_prepared_project(name).update_certificates(),126    }127    try:128        command_function = command_switch[namespace.command]129    except KeyError:130        raise SystemError(_(f'Unknown command {namespace.command}.'))131    command_argument_parser = ArgumentParser()132    command_argument_parser.add_argument('name', help=_('Project name.'))133    if namespace.command in ('run', 'deploy', 'update'):134        command_argument_parser.add_argument('version', help=_('Project version.'))135    command_arguments = command_argument_parser.parse_args(arguments)136    command_function(**vars(command_arguments))137class Registry:138    ENTRIES_MAX_COUNT = 10139    def __init__(self, path):140        self.path = path141    def prepare(self):142        logger.info(_('Preparing registry...'))143        if not os.path.isfile(self.path):144            logger.debug(f'Create â{self.path}â')145            with open(self.path, 'w') as file:146                file.write('[]')147    def read(self):148        logger.info(_('Reading registry...'))149        with open(self.path, 'r') as file:150            return json.load(file)151    def write(self, entries):152        logger.info(_('Writing registry...'))153        with open(self.path, 'w') as file:154            json.dump(entries, file)155    def update(self, entry):156        logger.info(_('Updating registry...'))157        entries = self.read()158        entries.insert(0, entry)159        if len(entries) > self.ENTRIES_MAX_COUNT:160            del entries[self.ENTRIES_MAX_COUNT - 1:]161        self.write(entries)162class Project:163    REPOSITORY_URL = 'https://github.com/vsevolod-skripnik/vash.git'164    def __init__(self, name: str):165        self.name = name166        # ~/projects/vash/manage.py167        self.manage_file_path = os.path.abspath(__file__)168        # ~/projects/vash169        self.repository_folder_path = os.path.sep.join(170            self.manage_file_path.split(os.path.sep)[:-1]171        )172        # ~/projects173        self.projects_folder_path = os.path.sep.join(174            self.repository_folder_path.split(os.path.sep)[:-1]175        )176        # ~/projects/{name}177        self.project_folder_path = f'{self.projects_folder_path}/{name}'178        # ~/projects/{name}/versions179        self.versions_folder_path = f'{self.project_folder_path}/versions'180        # ~/projects/{name}/resources181        self.resources_folder_path = f'{self.project_folder_path}/resources'182        # ~/projects/{name}/resources/{resource}183        self.resource_folder_paths = {184            'assets': f'{self.resources_folder_path}/assets',185            'certificates': f'{self.resources_folder_path}/certificates',186            'files': f'{self.resources_folder_path}/files',187            'pages': f'{self.resources_folder_path}/pages',188            'scripts': f'{self.resources_folder_path}/scripts',189            'styles': f'{self.resources_folder_path}/styles',190            'templates': f'{self.resources_folder_path}/templates',191        }192        # ~/projects/{name}/resources/pages/_html193        self.html_pages_folder_path = f'{self.resource_folder_paths["pages"]}/_html'194        # ~/projects/{name}/resources/certificates195        self.certificates_folder_path = self.resource_folder_paths['certificates']196        # ~/projects/{name}/registry.json197        registry_path = f'{self.project_folder_path}/registry.json'198        self.registry = Registry(path=registry_path)199        # Store them here for easy testing200        self.folder_paths = [201            self.project_folder_path,202            self.versions_folder_path,203            self.resources_folder_path,204        ]205        for resource_name, resource_folder_path in self.resource_folder_paths.items():206            self.folder_paths.append(resource_folder_path)207    def prepare_folders(self) -> list:208        logger.info(_('Preparing folders...'))209        created_folders = []210        for folder_path in self.folder_paths:211            if not os.path.isdir(folder_path):212                logger.debug(_(f'Create â{folder_path}â'))213                os.makedirs(folder_path)214                created_folders.append(folder_path)215        return created_folders216    def prepare(self):217        logging.info(_('Preparing project...'))218        self.prepare_folders()219        self.registry.prepare()220    def prepare_test_page(self):221        logging.info(_('Preparing index page...'))222        if not os.path.isdir(self.html_pages_folder_path):223            os.makedirs(self.html_pages_folder_path)224        page_path = f'{self.html_pages_folder_path}/index.html'225        if not os.path.isfile(page_path):226            page_content = (227                '<!DOCTYPE html>'228                '<html>'229                '  <head>'230                '    <meta charset="utf-8">'231                '     <title>Test title</title>'232                '  </head>'233                '  <body>'234                '    <p>Test paragraph</p>'235                '  </body>'236                '</html>'237            )238            with open(page_path, 'w') as file:239                file.write(page_content)240    def run_release(self, version: str):241        custom_environment = {242            'PROXY_PORT_80': '80',243            'PROXY_PORT_443': '443',244        }245        release = Release(246            project=self,247            version=version,248            service_names=SERVICE_NAMES_RUN,249        )250        release_data = release.run(251            do_build=True,252            as_daemon=False,253            custom_environment=custom_environment,254        )255        logger.info(_('Done. Perfect.'))256    def deploy_release(self, version: str):257        release = Release(258            project=self,259            version=version,260            service_names=SERVICE_NAMES_DEPLOY,261        )262        release_data = release.run(263            do_build=True,264            as_daemon=True,265        )266        self.prepare_test_page()267        page_url = f'https://127.0.0.1:{release_data["environment"]["PROXY_PORT_443"]}/'268        is_working_fine_on_port = test_page(page_url)269        if is_working_fine_on_port:270            saved_image_paths = release.save_images()271            for image_path in saved_image_paths:272                send_image(image_path)273            release.prune()274            logger.info(_('Done. Perfect.'))275        else:276            raise SystemError(_('Broken. Go check stuff'))277    def update_release(self, version):278        release = Release(279            project=self,280            version=version,281            service_names=SERVICE_NAMES_UPDATE,282        )283        try:284            release.prune()  # Delete previous release with same version285        except CalledProcessError:286            pass  # It's fine if it doesn't exist287        image_path = f'/tmp/{self.name}_proxy_{version}.tar'288        load_image(image_path)289        release_data = release.run(290            do_build=False,291            as_daemon=True,292        )293        self.prepare_test_page()294        page_url = f'https://127.0.0.1:{release_data["environment"]["PROXY_PORT_443"]}/'295        is_working_fine_on_port = test_page(page_url)296        if is_working_fine_on_port:297            port_forwardings = list_port_forwardings()298            create_port_forwarding(80, release_data['environment']['PROXY_PORT_80'])299            create_port_forwarding(443, release_data['environment']['PROXY_PORT_443'])300            for source_port, destination_port in port_forwardings:301                delete_port_forwarding(source_port, destination_port)302            domain = os.environ.get('PRODUCTION_DOMAIN') \303                or input(_('Production domain: '))304            page_url = f'https://{domain}/' \305                if domain else f'https://127.0.0.1:{release_data["environment"]["PROXY_PORT_443"]}/'306            is_working_fine_on_domain = test_page(page_url)307            if is_working_fine_on_domain:308                self.registry.update(release_data)309                logging.info(_('Done. Perfect.'))310            else:311                previous_releases = self.registry.read()...

test_floating_ips_port_forwarding_client.py

Source:test_floating_ips_port_forwarding_client.py

...62            bytes_body,63            201,64            floatingip_id=self.FAKE_FLOATINGIP_ID,65            **self.FAKE_PORT_FORWARDING_REQUEST)66    def _test_list_port_forwardings(self, bytes_body=False):67        self.check_service_client_function(68            self.floating_ips_port_forwarding_client.69            list_port_forwardings,70            "tempest.lib.common.rest_client.RestClient.get",71            self.FAKE_PORT_FORWARDINGS,72            bytes_body,73            200,74            floatingip_id=self.FAKE_FLOATINGIP_ID)75    def _test_show_port_forwardings(self, bytes_body=False):76        self.check_service_client_function(77            self.floating_ips_port_forwarding_client.78            show_port_forwarding,79            "tempest.lib.common.rest_client.RestClient.get",80            self.FAKE_PORT_FORWARDING_RESPONSE,81            bytes_body,82            200,83            floatingip_id=self.FAKE_FLOATINGIP_ID,84            port_forwarding_id=self.FAKE_PORT_FORWARDING_ID)85    def _test_delete_port_forwarding(self):86        self.check_service_client_function(87            self.floating_ips_port_forwarding_client.88            delete_port_forwarding,89            "tempest.lib.common.rest_client.RestClient.delete",90            {},91            status=204,92            floatingip_id=self.FAKE_FLOATINGIP_ID,93            port_forwarding_id=self.FAKE_PORT_FORWARDING_ID)94    def _test_update_port_forwarding(self, bytes_body=False):95        update_kwargs = {96            "internal_port": "27"97        }98        resp_body = {99            "port_forwarding": copy.deepcopy(100                self.FAKE_PORT_FORWARDING_RESPONSE['port_forwarding']101            )102        }103        resp_body["port_forwarding"].update(update_kwargs)104        self.check_service_client_function(105            self.floating_ips_port_forwarding_client.update_port_forwarding,106            "tempest.lib.common.rest_client.RestClient.put",107            resp_body,108            bytes_body,109            200,110            floatingip_id=self.FAKE_FLOATINGIP_ID,111            port_forwarding_id=self.FAKE_PORT_FORWARDING_ID,112            **update_kwargs)113    def test_list_port_forwardings_with_str_body(self):114        self._test_list_port_forwardings()115    def test_list_port_forwardings_with_bytes_body(self):116        self._test_list_port_forwardings(bytes_body=True)117    def test_show_port_forwardings_with_str_body(self):118        self._test_show_port_forwardings()119    def test_show_port_forwardings_with_bytes_body(self):120        self._test_show_port_forwardings(bytes_body=True)121    def test_create_port_forwarding_with_str_body(self):122        self._test_create_port_forwarding()123    def test_create_port_forwarding_with_bytes_body(self):124        self._test_create_port_forwarding(bytes_body=True)125    def test_update_port_forwarding_with_str_body(self):126        self._test_update_port_forwarding()127    def test_update_port_forwarding_with_bytes_body(self):...

floating_ips_port_forwarding_client.py

Source:floating_ips_port_forwarding_client.py

...54        """55        uri = '/floatingips/%s/port_forwardings/%s' % (56            floatingip_id, port_forwarding_id)57        return self.delete_resource(uri)58    def list_port_forwardings(self, floatingip_id, **filters):59        """Lists floating Ip port forwardings.60        For a full list of available parameters, please refer to the official61        API reference:62        https://docs.openstack.org/api-ref/network/v2/index.html#list-floating-ip-port-forwardings-detail63        """64        uri = '/floatingips/%s/port_forwardings' % floatingip_id...

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.