Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use describe_connection method in localstack

Best Python code snippet using localstack_python

connection_index.py

Source:connection_index.py

...5from karadoc.common.model import file_index6from karadoc.common.run import load_populate7from karadoc.common.stream import load_stream_file8from karadoc.common.table_utils import populate_exists, stream_file_exists9def describe_connection(job, connection_name, source_details, full_table_name, direction, env: Optional[str]):10    try:11        conn = conf.get_connection_conf(connection_name, env)12        common_args = {13            "job_type": type(job).__name__,14            "full_table_name": full_table_name,15            "conn_name": connection_name,16            "conn_type": conn["type"],17            "direction": direction,18            "job_disabled": str(job.disable),19            "conn_disabled": str(conn.get("disable", False)),20        }21        if conn["type"].endswith(".jdbc"):22            return {23                **common_args,24                "host": conn["host"],25                "username": conn["user"],26                "database": conn["database"],27                "table": source_details.get("table"),28            }29        elif conn["type"].endswith(".mssql"):30            return {31                **common_args,32                "host": conn["host"],33                "database": conn["database"],34                "table": source_details.get("table"),35            }36        elif conn["type"].endswith(".cosmosdb"):37            return {38                **common_args,39                "host": conn["host"],40                "database": source_details.get("database"),41                "table": source_details.get("collection"),42            }43        elif conn["type"].endswith(".wasbs"):44            return {45                **common_args,46                "host": conn["storage_account"],47                "container": conn["container_name"],48                "table": source_details.get("path"),49            }50        elif conn["type"].endswith(".azure_table"):51            return {52                **common_args,53                "host": conn["storage_account"],54                "table": source_details.get("table"),55            }56        elif conn["type"].endswith(".event_hub"):57            return {58                **common_args,59                "host": conn["HubName"],60                "username": source_details.get("consumer_group"),61                "table": source_details.get("entity_path"),62            }63        elif conn["type"].endswith(".sftp"):64            return {65                **common_args,66                "host": conn["host"],67                "username": conn["username"],68            }69        elif conn["type"].endswith(".bigquery") or conn["type"].endswith(".bigquery_2"):70            return {71                **common_args,72                "container": conn["bucket_name"],73                "database": conn["project_name"],74                "table": source_details.get("table"),75            }76        else:77            return common_args78    except Exception as e:79        return {80            "full_table_name": full_table_name,81            "conn_name": connection_name,82            "conn_type": "error",83            "direction": direction,84            "host": str(e),85            "disabled": str(job.disable),86        }87def __build_connections_generator(env: Optional[str]) -> Iterator[Dict[str, str]]:  # NOSONAR88    # TODO: refactor this method by making a generic load_action_files method89    for (schema, table, path) in list(file_index.list_schema_table_folders()):90        full_table_name = "%s.%s" % (schema, table)91        if populate_exists(full_table_name):92            job = load_populate(full_table_name)93            for source_name, source_details in job.external_inputs.items():94                connection_name = source_details[CONNECTION_GROUP]95                yield describe_connection(96                    job, connection_name, source_details, full_table_name, direction="input", env=env97                )98            for dest_name, dest_details in job.external_outputs.items():99                connection_name = dest_details[CONNECTION_GROUP]100                yield describe_connection(101                    job, connection_name, dest_details, full_table_name, direction="output", env=env102                )103        if stream_file_exists(full_table_name):104            job = load_stream_file(full_table_name)105            for source_name, source_details in job.external_inputs.items():106                connection_name = source_details[CONNECTION_GROUP]107                yield describe_connection(108                    job, connection_name, source_details, full_table_name, direction="input", env=env109                )110            if job.external_output is not None:111                yield describe_connection(112                    job,113                    job.external_output[CONNECTION_GROUP],114                    job.external_output,115                    full_table_name,116                    direction="output",117                    env=env,118                )119def list_connections_df(env: Optional[str]) -> DataFrame:120    """Builds a Spark DataFrame listing all external inputs/outputs used by POPULATE.py files, with their respective121    source/dest and connection properties.122    :param env: Retrieves the connections properties for the specified environment123    :return: a Spark DataFrame124    """125    from karadoc.common import Job...

problem.py

Source:problem.py

...101      for point2 in self.points:102        points_join.append((point1, point2))103    return reduce(group, [describe_distance(distance) for distance in points_join])104  def describe_connections(self):105    def describe_connection(connection):106      start=self.points.index(connection[0])107      end=self.points.index(connection[1])108      return f"(connected LOCATION{start} LOCATION{end})"109    return reduce(group, [describe_connection(connection) for connection in self.connections])110  def describe_metric(self):111    metrics=['totalTime','totalBattery','totalDistance','recharges']112    if self.description['metric'] in metrics:113      return f"(:metric minimize ({self.description['metric']}))"114    else:115      return ""116  def __repr__(self):117    return self.__str__118  def __str__(self):119    return (120      f"(define (problem probPlanetary) (:domain planetary) (:objects\n"121      f" {self.get_robots()} - robot\n"122      f" {self.get_locations()} - location\n"123      f" {self.get_speeds()} - speedType\n"...

scratch.py

Source:scratch.py

...18vc = VpnClient()19vc.stop_vpn('all')20vc.start_vpn('uk')21le = LeakTestExplorer()22le.describe_connection()23print('stopping uk vpn')24vc.stop_vpn('uk')25print('connecting to inda')26vc.start_vpn('india')27print('closing driver')28le.close_driver()29print('New connection')30le = LeakTestExplorer()31# le = LeakTestExplorer(le.driver.refresh())32print('Describing')33le.describe_connection()34# import subprocess35#36# cs="sudo systemctl status shadowsocks-libev-local@ssClientConfig_mumbai"37# cmd = 'sudo /bin/systemctl status {cs}.service'.format(cs=cs)...

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.