Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use append_row_to_raw_collection method in localstack

Best Python code snippet using localstack_python

metric_aggregator.py

Source:metric_aggregator.py

...158        for subkey, val in val.items():159            if isinstance(val, dict) and val.get("invoked"):160                if val["invoked"] > 0 and not metric_recorder_external[key][subkey]["invoked"]:161                    print(f"found invocation mismatch: {key}.{subkey}")162def append_row_to_raw_collection(collection_raw_csv_file_name, row, arch):163    with open(collection_raw_csv_file_name, "a") as fd:164        writer = csv.writer(fd)165        row.append(arch)166        writer.writerow(row)167def aggregate_recorded_raw_data(168    base_dir: str, collection_raw_csv: Optional[str] = None, collect_for_arch: Optional[str] = ""169) -> dict:170    pathlist = Path(base_dir).rglob("metric-report-raw-data-*.csv")171    recorded = _init_service_metric_counter()172    for path in pathlist:173        print(f"checking {str(path)}")174        with open(path, "r") as csv_obj:175            csv_dict_reader = csv.reader(csv_obj)176            # skip the header177            next(csv_dict_reader)178            for row in csv_dict_reader:179                if collection_raw_csv:180                    arch = ""181                    if "arm64" in str(path):182                        arch = "arm64"183                    elif "amd64" in str(path):184                        arch = "amd64"185                    # only aggregate all if we did not set a specific target to collect186                    if not collect_for_arch:187                        append_row_to_raw_collection(collection_raw_csv, copy.deepcopy(row), arch)188                    elif collect_for_arch in str(path):189                        append_row_to_raw_collection(collection_raw_csv, copy.deepcopy(row), arch)190                metric: Metric = Metric(*row)191                if metric.xfail == "True":192                    print(f"test {metric.node_id} marked as xfail")193                    continue194                if collect_for_arch and collect_for_arch not in str(path):195                    continue196                service = recorded[metric.service]197                ops = service[metric.operation]198                errors = ops.setdefault("errors", {})199                if metric.exception:200                    exception = metric.exception201                    errors[exception] = ops.get(exception, 0) + 1202                elif int(metric.response_code) >= 300:203                    for expected_error in ops.get("errors", {}).keys():...

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.