Extended test context that manages two clients connected to the same CLI server.
| 29 | |
| 30 | |
| 31 | class MultiClientContext: |
| 32 | """Extended test context that manages two clients connected to the same CLI server.""" |
| 33 | |
| 34 | def __init__(self): |
| 35 | self.cli_path: str = "" |
| 36 | self.home_dir: str = "" |
| 37 | self.work_dir: str = "" |
| 38 | self.proxy_url: str = "" |
| 39 | self._proxy: CapiProxy | None = None |
| 40 | self._client1: CopilotClient | None = None |
| 41 | self._client2: CopilotClient | None = None |
| 42 | |
| 43 | async def setup(self): |
| 44 | from .testharness.context import get_cli_path_for_tests |
| 45 | |
| 46 | self.cli_path = get_cli_path_for_tests() |
| 47 | self.home_dir = os.path.realpath(tempfile.mkdtemp(prefix="copilot-multi-config-")) |
| 48 | self.work_dir = os.path.realpath(tempfile.mkdtemp(prefix="copilot-multi-work-")) |
| 49 | |
| 50 | self._proxy = CapiProxy() |
| 51 | self.proxy_url = await self._proxy.start() |
| 52 | |
| 53 | github_token = ( |
| 54 | "fake-token-for-e2e-tests" if os.environ.get("GITHUB_ACTIONS") == "true" else None |
| 55 | ) |
| 56 | |
| 57 | # Client 1 uses TCP mode so a second client can connect to the same server |
| 58 | self._client1 = CopilotClient( |
| 59 | connection=RuntimeConnection.for_tcp( |
| 60 | path=self.cli_path, connection_token="py-tcp-shared-test-token" |
| 61 | ), |
| 62 | working_directory=self.work_dir, |
| 63 | env=self.get_env(), |
| 64 | github_token=github_token, |
| 65 | ) |
| 66 | |
| 67 | # Trigger connection by creating and disconnecting an init session |
| 68 | init_session = await self._client1.create_session( |
| 69 | on_permission_request=PermissionHandler.approve_all |
| 70 | ) |
| 71 | await init_session.disconnect() |
| 72 | |
| 73 | # Read the actual port from client 1 and create client 2 |
| 74 | actual_port = self._client1.runtime_port |
| 75 | assert actual_port is not None, "Client 1 should have an actual port after connecting" |
| 76 | |
| 77 | self._client2 = CopilotClient( |
| 78 | connection=RuntimeConnection.for_uri( |
| 79 | f"localhost:{actual_port}", connection_token="py-tcp-shared-test-token" |
| 80 | ) |
| 81 | ) |
| 82 | |
| 83 | async def teardown(self, test_failed: bool = False): |
| 84 | if self._client2: |
| 85 | try: |
| 86 | await self._client2.stop() |
| 87 | except Exception: |
| 88 | pass |