Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use aggregate_recorded_raw_data method in localstack

Best Python code snippet using localstack_python

metric_aggregator.py

Source:metric_aggregator.py

...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():204                        if expected_error in metric.response_data:205                            # assume we have a match206                            errors[expected_error] += 1207                            LOG.warning(208                                f"Exception assumed for {metric.service}.{metric.operation}: code {metric.response_code}"209                            )210                            break211                ops["invoked"] += 1212                if metric.snapshot == "True":213                    ops["snapshot"] = True  # TODO snapshot currently includes also "skip_verify"214                    ops["snapshot_skipped_paths"] = metric.snapshot_skipped_paths or ""215                if metric.aws_validated == "True":216                    ops["aws_validated"] = True217                if not metric.parameters:218                    params = ops.setdefault("parameters", {})219                    params["_none_"] = params.get("_none_", 0) + 1220                else:221                    for p in metric.parameters.split(","):222                        ops["parameters"][p] += 1223                test_list = ops.setdefault("tests", [])224                if metric.node_id not in test_list:225                    test_list.append(metric.node_id)226    return recorded227def main():228    if not len(sys.argv) >= 2 or not Path(sys.argv[1]).is_dir():229        print_usage()230        return231    base_dir = sys.argv[1]232    collect_for_arch = ""233    if len(sys.argv) == 3:234        collect_for_arch = sys.argv[2]235        if collect_for_arch not in ("amd64", "arm64"):236            print_usage()237            return238        print(239            f"Set target to '{collect_for_arch}' - will only aggregate for these test results. Raw collection of all files.\n"240        )241    # TODO: removed splitting of internal/external recorded calls, as some pro tests use 'internals' to connect to service242    metrics_path = os.path.join(base_dir, "parity_metrics")243    Path(metrics_path).mkdir(parents=True, exist_ok=True)244    dtime = datetime.datetime.utcnow().strftime("%Y-%m-%d-%H-%M-%s")245    collection_raw_csv = os.path.join(metrics_path, f"raw-collected-data-{dtime}.csv")246    with open(collection_raw_csv, "w") as fd:247        writer = csv.writer(fd)248        header = Metric.RAW_DATA_HEADER.copy()249        header.append("arch")250        writer.writerow(header)251    recorded_metrics = aggregate_recorded_raw_data(base_dir, collection_raw_csv, collect_for_arch)252    write_json(253        os.path.join(254            metrics_path,255            f"metric-report-{dtime}{collect_for_arch}.json",256        ),257        recorded_metrics,258    )259    # filename = os.path.join(metrics_path, f"metric-report-{dtime}{collect_for_arch}.md")260    # create_readable_report(filename, recorded_metrics)261    filename = os.path.join(metrics_path, f"metric-report-{dtime}{collect_for_arch}.html")262    create_simple_html_report(filename, recorded_metrics)263if __name__ == "__main__":...

coverage_docs_utility.py

Source:coverage_docs_utility.py

...179            details = service[row["operation"]]180            if not details["implemented"] and row["is_implemented"] == "True":181                details["implemented"] = True182                details["pro"] = True183    recorded_metrics = aggregate_recorded_raw_data(base_dir=path_to_raw_metrics)184    # create_metric_coverage_docs_internal(185    #     file_name=path_to_raw_metrics + "/metric-coverage_internal.md",186    #     metrics=recorded_metrics,187    #     impl_details=impl_details,188    #     coverage=coverage,189    # )190    create_metric_coverage_docs(191        file_name=path_to_raw_metrics + "/coverage.md",192        metrics=recorded_metrics,193        impl_details=impl_details,194    )195def print_usage():196    print("missing arguments")197    print(...

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.