Method start

python/e2e/testharness/proxy.py:27–78

Launch the proxy server and return its URL.

(self)

        self._ca_file_path: str | None = None

    async def start(self) -> str:
        """Launch the proxy server and return its URL."""
        if self._proxy_url:
            return self._proxy_url

        # The harness server is in the shared test directory
        server_path = os.path.join(
            os.path.dirname(__file__), "..", "..", "..", "test", "harness", "server.ts"
        )
        server_path = os.path.abspath(server_path)

        # On Windows, use shell=True to find npx
        use_shell = platform.system() == "Windows"

        self._process = subprocess.Popen(
            ["npx", "tsx", server_path],
            stdout=subprocess.PIPE,
            stderr=None,  # Inherit stderr to parent for debugging
            text=True,
            cwd=os.path.dirname(server_path),
            shell=use_shell,
        )

        # Read until the server prints "Listening: http://..."; npm/npx may emit
        # wrapper output first on some platforms.
        line = ""
        match = None
        while True:
            line = self._process.stdout.readline()
            if not line:
                self._process.kill()
                raise RuntimeError("Failed to read proxy URL")
            match = re.search(r"Listening: (http://[^\s]+)", line.strip())
            if match:
                break

        self._proxy_url = match.group(1)
        metadata_match = re.search(r"(\{.*\})\s*$", line.strip())
        if not metadata_match:
            self._process.kill()
            raise RuntimeError(f"Proxy startup line missing CONNECT proxy metadata: {line}")
        try:
            metadata = json.loads(metadata_match.group(1))
        except json.JSONDecodeError as exc:
            self._process.kill()
            raise RuntimeError(f"Failed to parse proxy startup metadata: {line}") from exc
        self._connect_proxy_url = metadata.get("connectProxyUrl")
        self._ca_file_path = metadata.get("caFilePath")
        if not self._connect_proxy_url or not self._ca_file_path:
            self._process.kill()
            raise RuntimeError(f"Proxy startup metadata missing CONNECT proxy details: {line}")
        return self._proxy_url

    async def stop(self, skip_writing_cache: bool = False):
        """Gracefully shut down the proxy server.