capture_notimplemented_responses.py

import csv
import json
import logging
import re
import sys
import time
import traceback
from datetime import timedelta
from pathlib import Path
from typing import Optional, TypedDict

import botocore.config
import requests
from botocore.exceptions import (
    ClientError,
    ConnectTimeoutError,
    EndpointConnectionError,
    ReadTimeoutError,
)
from botocore.parsers import ResponseParserError
from rich.console import Console

from localstack.aws.connect import connect_externally_to
from localstack.aws.mocking import Instance, generate_request
from localstack.aws.spec import ServiceCatalog

logging.basicConfig(level=logging.INFO)
service_models = ServiceCatalog()

c = Console()

STATUS_TIMEOUT_ERROR = 901
STATUS_PARSING_ERROR = 902
STATUS_CONNECTION_ERROR = 903
# dict of operations that should be skipped for a service, currently only contains s3.PostObject (which we added for internal use)
PHANTOM_OPERATIONS = {"s3": ["PostObject"]}

# will only include available services
response = requests.get("http://localhost:4566/_localstack/health").content.decode("utf-8")
latest_services_pro = [k for k in json.loads(response).get("services").keys()]

exclude_services = {"azure"}
latest_services_pro = [s for s in latest_services_pro if s not in exclude_services]
latest_services_pro.sort()


class RowEntry(TypedDict, total=False):
    service: str
    operation: str
    status_code: int
    error_code: str
    error_message: str
    is_implemented: bool


def simulate_call(service: str, op: str) -> RowEntry:
    """generates a mock request based on the service and operation model and sends it to the API"""
    client = connect_externally_to.get_client(
        service,
        aws_access_key_id="test",
        aws_secret_access_key="test",
        config=botocore.config.Config(
            parameter_validation=False,
            retries={"max_attempts": 0, "total_max_attempts": 1},
            connect_timeout=90,
            read_timeout=120,
            inject_host_prefix=False,
        ),
    )

    service_model = service_models.get(service)
    op_model = service_model.operation_model(op)
    parameters = generate_request(op_model) or {}
    result = _make_api_call(client, service, op, parameters)
    error_msg = result.get("error_message", "")

    if result.get("error_code", "") == "InternalError":
        # some deeper investigation necessary, check for some common errors here and retry
        if service == "apigateway" and "Unexpected HTTP method" in error_msg:
            # moto currently raises exception in some requests, if the http method is not supported, meaning it is not implemented
            result["status_code"] = 501  # reflect that this is not implemented
        elif (
            "localstack.aws.protocol.parser.ProtocolParserError: Unable to parse request (not well-formed (invalid token)"
            in error_msg
        ):
            # parsing errors might be due to invalid parameter values
            # try to re-create params
            logging.debug(
                "ProtocolParserError detected, old parameters used: %s\nre-running request %s.%s with new parameters",
                parameters,
                service,
                op,
            )
            parameters = generate_request(op_model) or {}
            result = _make_api_call(client, service, op, parameters)
        elif "TypeError" in error_msg and "got an unexpected keyword argument" in error_msg:
            # sometimes LocalStack's autogenerated API is not yet up-to-date, which could cause
            # if we see the 'unexpected keyword argument' error, we can re-try after removing the unknown argument
            while match := re.search("got an unexpected keyword argument '(.*)'", error_msg):
                keyword = match.groups()[0]
                # find the corresponding parameter for the unexpected keyword
                if argument := next(
                    (
                        arg
                        for arg in list(parameters.keys())
                        if keyword.replace("_", "") == arg.casefold()
                    ),
                    None,
                ):
                    logging.warning(
                        "Got 'TypeError' with unexpected keyword argument: '%s' for %s.%s. "
                        "Re-trying without keyword ...",
                        keyword,
                        service,
                        op,
                    )
                    # remove the unexpected keyword and try again
                    parameters.pop(argument)
                    result = _make_api_call(client, service, op, parameters)

                    if result.get("error_code", "") != "InternalError":
                        break
                    if argument in parameters:
                        # sometimes the parameter seem to be automatically added again by boto
                        # happened eg for ClientToken in ec2.ProvisionIpamPoolCidr
                        logging.warning(
                            "unexpected keyword '%s' was added to the parameters again for: %s.%s",
                            argument,
                            service,
                            op,
                        )
                        break
                    error_msg = result.get("error_message", "")
                else:
                    # keyword argument not found in the parameters
                    break
    elif result.get("error_code", "") == "UnsupportedOperation" and service == "stepfunctions":
        # the stepfunction lib returns 500 for not implemented + UnsupportedOperation
        result["status_code"] = 501  # reflect that this is not implemented
    if result.get("status_code") in [0, 901, 902, 903]:
        # something went wrong, we do not know exactly what/why - just try again one more time
        logging.debug(
            "Detected invalid status code %i for %s.%s. Re-running request with new parameters",
            result.get("status_code"),
            service,
            op,
        )
        parameters = generate_request(op_model) or {}  # should be generate_parameters I guess
        result = _make_api_call(client, service, op, parameters)
    return result


def _make_api_call(client, service: str, op: str, parameters: Optional[Instance]):
    result = RowEntry(service=service, operation=op, status_code=0)
    try:
        response = client._make_api_call(op, parameters)
        result["status_code"] = response["ResponseMetadata"]["HTTPStatusCode"]
    except ClientError as ce:
        result["status_code"] = ce.response["ResponseMetadata"]["HTTPStatusCode"]
        result["error_code"] = ce.response.get("Error", {}).get("Code", "Unknown?")
        result["error_message"] = ce.response.get("Error", {}).get("Message", "Unknown?")
    except (ReadTimeoutError, ConnectTimeoutError) as e:
        logging.warning("Reached timeout for %s.%s. Assuming it is implemented.", service, op)
        logging.exception(e)
        result["status_code"] = STATUS_TIMEOUT_ERROR
        result["error_message"] = traceback.format_exception(e)
    except EndpointConnectionError as e:
        # TODO: investigate further;for now assuming not implemented
        logging.warning("Connection failed for %s.%s. Assuming it is not implemented.", service, op)
        logging.exception(e)
        result["status_code"] = STATUS_CONNECTION_ERROR
        result["error_message"] = traceback.format_exception(e)
    except ResponseParserError as e:
        # TODO: this is actually a bit tricky and might have to be handled on a service by service basis again
        logging.warning("Parsing issue for %s.%s. Assuming it isn't implemented.", service, op)
        logging.exception(e)
        logging.warning("%s.%s: used parameters %s", service, op, parameters)
        result["status_code"] = STATUS_PARSING_ERROR
        result["error_message"] = traceback.format_exception(e)
    except Exception as e:
        logging.warning("Unknown Exception for %s.%s", service, op)
        logging.exception(e)
        logging.warning("%s.%s: used parameters %s", service, op, parameters)
        result["error_message"] = traceback.format_exception(e)
    return result


def map_to_notimplemented(row: RowEntry) -> bool:
    """
    Some simple heuristics to check the API responses and classify them into implemented/notimplemented

    Ideally they all should behave the same way when receiving requests for not yet implemented endpoints
    (501 with a "not yet implemented" message)

    :param row: the RowEntry
    :return: True if we assume it is not implemented, False otherwise
    """
    if row["status_code"] in [STATUS_PARSING_ERROR]:
        # parsing issues are nearly always due to something not being implemented or activated
        return True
    if row["status_code"] in [STATUS_TIMEOUT_ERROR]:
        #  timeout issue, interpreted as implemented until there's a better heuristic
        return False
    if row["status_code"] == STATUS_CONNECTION_ERROR:
        return True
    if (
        row["service"] == "cloudfront"
        and row["status_code"] == 500
        and row.get("error_code") == "500"
        and row.get("error_message", "").lower() == "internal server error"
    ):
        return True
    if row["service"] == "dynamodb" and row.get("error_code") == "UnknownOperationException":
        return True
    if row["service"] == "lambda" and row["status_code"] == 404 and row.get("error_code") == "404":
        return True
    if (
        row["service"]
        in [
            "route53",
            "s3control",
        ]
        and row["status_code"] == 404
        and row.get("error_code") == "404"
        and row.get("error_message") is not None
        and "not found" == row.get("error_message", "").lower()
    ):
        return True
    if (
        row["service"] in ["xray", "batch", "glacier", "resource-groups", "apigateway"]
        and row["status_code"] == 404
        and row.get("error_message") is not None
        and "The requested URL was not found on the server" in row.get("error_message")
    ):
        return True
    if (
        row["status_code"] == 501
        and row.get("error_message") is not None
        and "not yet implemented" in row.get("error_message", "")
    ):
        return True
    if row.get("error_message") is not None and "not yet implemented" in row.get(
        "error_message", ""
    ):
        return True
    if row["status_code"] == 501:
        return True
    if (
        row["status_code"] == 500
        and row.get("error_code") == "500"
        and not row.get("error_message")
    ):
        return True
    return False


def run_script(services: list[str], path: None):
    """send requests against all APIs"""
    print(
        f"writing results to '{path}implementation_coverage_full.csv' and '{path}implementation_coverage_aggregated.csv'..."
    )
    with (
        open(f"{path}implementation_coverage_full.csv", "w") as csvfile,
        open(f"{path}implementation_coverage_aggregated.csv", "w") as aggregatefile,
    ):
        full_w = csv.DictWriter(
            csvfile,
            fieldnames=[
                "service",
                "operation",
                "status_code",
                "error_code",
                "error_message",
                "is_implemented",
            ],
        )
        aggregated_w = csv.DictWriter(
            aggregatefile,
            fieldnames=["service", "implemented_count", "full_count", "percentage"],
        )

        full_w.writeheader()
        aggregated_w.writeheader()

        total_count = 0
        for service_name in services:
            service = service_models.get(service_name)
            for op_name in service.operation_names:
                if op_name in PHANTOM_OPERATIONS.get(service_name, []):
                    continue
                total_count += 1

        time_start = time.perf_counter_ns()
        counter = 0
        responses = {}
        for service_name in services:
            c.print(f"\n=====  {service_name} =====")
            service = service_models.get(service_name)
            for op_name in service.operation_names:
                if op_name in PHANTOM_OPERATIONS.get(service_name, []):
                    continue
                counter += 1
                c.print(
                    f"{100 * counter/total_count:3.1f}% | Calling endpoint {counter:4.0f}/{total_count}: {service_name}.{op_name}"
                )

                # here's the important part (the actual service call!)
                response = simulate_call(service_name, op_name)

                responses.setdefault(service_name, {})[op_name] = response
                is_implemented = str(not map_to_notimplemented(response))
                full_w.writerow(response | {"is_implemented": is_implemented})

            # calculate aggregate for service
            all_count = len(responses[service_name].values())
            implemented_count = len(
                [r for r in responses[service_name].values() if not map_to_notimplemented(r)]
            )
            implemented_percentage = implemented_count / all_count

            aggregated_w.writerow(
                {
                    "service": response["service"],
                    "implemented_count": implemented_count,
                    "full_count": all_count,
                    "percentage": f"{implemented_percentage * 100:.1f}",
                }
            )
        time_end = time.perf_counter_ns()
        delta = timedelta(microseconds=(time_end - time_start) / 1000.0)
        c.print(f"\n\nDone.\nTotal time to completion: {delta}")


def calculate_percentages():
    aggregate = {}

    implemented_aggregate = {}
    aggregate_list = []

    with open("./output-notimplemented.csv", "r") as fd:
        reader = csv.DictReader(fd, fieldnames=["service", "operation", "implemented"])
        for line in reader:
            if line["implemented"] == "implemented":
                continue
            aggregate.setdefault(line["service"], {}).setdefault(line["operation"], line)

        for service in aggregate.keys():
            vals = aggregate[service].values()
            all_count = len(vals)
            implemented_count = len([v for v in vals if v["implemented"] == "True"])
            implemented_aggregate[service] = implemented_count / all_count
            aggregate_list.append(
                {
                    "service": service,
                    "count": all_count,
                    "implemented": implemented_count,
                    "percentage": implemented_count / all_count,
                }
            )

    aggregate_list.sort(key=lambda k: k["percentage"])

    with open("implementation_coverage_aggregated.csv", "w") as csv_fd:
        writer = csv.DictWriter(
            csv_fd, fieldnames=["service", "percentage", "implemented", "count"]
        )
        writer.writeheader()

        for agg in aggregate_list:
            agg["percentage"] = f"{agg['percentage'] * 100:.1f}"
            writer.writerow(agg)


# @click.command()
def main():
    path = "./"
    if len(sys.argv) > 1 and Path(sys.argv[1]).is_dir():
        path = sys.argv[1]
        if not path.endswith("/"):
            path += "/"
    run_script(latest_services_pro, path=path)


if __name__ == "__main__":
    main()