Repository: qdrant/qdrant
File: tests/consensus_tests/auth_tests/test_jwt_access.py (Python)

Code before:
    "readyz": EndpointAccess(True, True, True, "GET /readyz", "grpc.health.v1.Health/Check", everything=True),
    "healthz": EndpointAccess(True, True, True, "GET /healthz", "grpc.health.v1.Health/Check", everything=True),
    "livez": EndpointAccess(True, True, True, "GET /livez", "grpc.health.v1.Health/Check", everything=True),
    "telemetry": EndpointAccess(True, True, True, "GET /telemetry"),
    "metrics": EndpointAccess(True, False, True, "GET /metrics", coll_r=False),
    "get_issues": EndpointAccess(True, True, True, "GET /issues"),
    "clear_issues": EndpointAccess(False, False, True, "DELETE /issues"),
    "get_logger_config": EndpointAccess(True, False, True, "GET /logger", coll_r=False),
    "update_logger_config": EndpointAccess(False, False, True, "POST /logger"),
}


def test_all_actions_have_tests():
    # for each action
    for action_name in ACTION_ACCESS.keys():
        # a test_{action_name} exists in this file
        test_name = f"test_{action_name}"
        assert (
                test_name in globals()
        ), f"An action is not tested: `{test_name}` was not found in this file"


def test_all_rest_endpoints_are_covered():
    # Load the JSON content from the openapi.json file
    project_root = Path(__file__).parent.parent.parent.parent
    with open(project_root / "docs" / "redoc" / "master" / "openapi.json", "r") as file:
        openapi_data = json.load(file)

    # Extract all endpoint paths
    endpoint_paths = []
    for path in openapi_data["paths"].keys():
        for method in openapi_data["paths"][path]:
            method = method.upper()
            endpoint_paths.append(f"{method} {path}")

    # check that all endpoints are covered in ACTION_ACCESS
    covered_endpoints = set(v.rest_endpoint for v in ACTION_ACCESS.values())
    for endpoint in endpoint_paths:
        assert (
                endpoint in covered_endpoints
        ), f"REST endpoint `{endpoint}` not found in any of the `ACTION_ACCESS` REST endpoints"


class MetadataInterceptor(ClientInterceptor):
    """A test interceptor that injects invocation metadata."""

    def __init__(self, metadata: List[Tuple[str, str]]):
        self._metadata = metadata

    def intercept(self, method, request_or_iterator, call_details: ClientCallDetails):
        """Add invocation metadata to request."""

Review comment (suggestion):
Maybe use `get_project_root()` here?

Also, with `pathlib`, you don't need `with open`:
```python
file_path = Path("…")
data = json.loads(file_path.read_text())
```

Code after:
    "readyz": EndpointAccess(True, True, True, "GET /readyz", "grpc.health.v1.Health/Check", everything=True),
    "healthz": EndpointAccess(True, True, True, "GET /healthz", "grpc.health.v1.Health/Check", everything=True),
    "livez": EndpointAccess(True, True, True, "GET /livez", "grpc.health.v1.Health/Check", everything=True),
    "telemetry": EndpointAccess(True, True, True, "GET /telemetry"),
    "metrics": EndpointAccess(True, False, True, "GET /metrics", coll_r=False),
    "get_issues": EndpointAccess(True, True, True, "GET /issues"),
    "clear_issues": EndpointAccess(False, False, True, "DELETE /issues"),
    "get_logger_config": EndpointAccess(True, False, True, "GET /logger", coll_r=False),
    "update_logger_config": EndpointAccess(False, False, True, "POST /logger"),
}


def test_all_actions_have_tests():
    # for each action
    for action_name in ACTION_ACCESS.keys():
        # a test_{action_name} exists in this file
        test_name = f"test_{action_name}"
        assert (
                test_name in globals()
        ), f"An action is not tested: `{test_name}` was not found in this file"


def test_all_rest_endpoints_are_covered():
    # Load the JSON content from the openapi.json file
    openapi_path = PROJECT_ROOT / "docs" / "redoc" / "master" / "openapi.json"
    openapi_data = json.loads(openapi_path.read_text())

    # Extract all endpoint paths
    endpoint_paths = []
    for path in openapi_data["paths"].keys():
        for method in openapi_data["paths"][path]:
            method = method.upper()
            endpoint_paths.append(f"{method} {path}")

    # check that all endpoints are covered in ACTION_ACCESS
    covered_endpoints = set(v.rest_endpoint for v in ACTION_ACCESS.values())
    for endpoint in endpoint_paths:
        assert (
                endpoint in covered_endpoints
        ), f"REST endpoint `{endpoint}` not found in any of the `ACTION_ACCESS` REST endpoints"


class MetadataInterceptor(ClientInterceptor):
    """A test interceptor that injects invocation metadata."""

    def __init__(self, metadata: List[Tuple[str, str]]):
        self._metadata = metadata

    def intercept(self, method, request_or_iterator, call_details: ClientCallDetails):
        """Add invocation metadata to request."""
        new_details = call_details._replace(metadata=self._metadata)