def find_matching_response()

in runtime_scripts/replay.py [0:0]

• 38 lines of code
• 13 McCabe index (conditional complexity)

    def find_matching_response(self, request_flow: http.HTTPFlow):
        """
        Find a matching response from the recorded flows for the given request.

        Args:
            request_flow (http.HTTPFlow): The incoming HTTP request flow.

        Returns:
            dict: The matching recorded flow, or None if not found.
        """
        request_method = request_flow.request.method.encode("utf-8")
        request_url = request_flow.request.url

        # First, check in other_flows
        for flow in self.other_flows:
            recorded_method = flow["request"]["method"]
            recorded_url = self.construct_url(flow)
            if recorded_method == request_method and recorded_url == request_url:
                return flow

        # Now check in attachment_flows
        for flow in self.attachment_flows:
            recorded_method = flow["request"]["method"]
            recorded_url = self.construct_url(flow)
            # Strip request URL to match
            request_url_match = request_url.split("?")[0]
            # In recorded URL perform ID replacement
            for key, value in REPLACEMENT_VARS.items():
                recorded_url = recorded_url.replace(key, value)
            if recorded_method == request_method and recorded_url == request_url_match:
                return flow

        # Then, check in filtered_flows
        request_content = request_flow.request.content
        content_type = request_flow.request.headers.get("Content-Type", "")
        flow_entry = self.find_matching_flow(
            request_method, request_url, request_content, content_type
        )
        if flow_entry is None:
            return None

        matching_flow = flow_entry["flow"]

        if not matching_flow["response"]:
            return matching_flow

        matching_flow_headers = self.convert_headers_to_dict(
            matching_flow["request"]["headers"]
        )
        if self.is_multipart_form_data(
            matching_flow_headers
        ) or self.is_x_www_form_urlencoded(matching_flow_headers):
            # We do not want to tamper original matching flow, thus create a copy
            matching_flow = deepcopy(matching_flow)
            self.replace_unique_ids(request_flow, matching_flow)

        # If content is ignored, undo
        if matching_flow["response"]["content"] == "IGNORED":
            flow_entry["marked"] = False
            matching_flow["response"]["content"] = json.dumps({})

        return matching_flow