Spawns a TCP CLI server with an explicit connection token.
| 20 | |
| 21 | |
| 22 | class ConnectionTokenContext: |
| 23 | """Spawns a TCP CLI server with an explicit connection token.""" |
| 24 | |
| 25 | def __init__(self, token: str | None): |
| 26 | self.token = token |
| 27 | self.cli_path: str = "" |
| 28 | self.home_dir: str = "" |
| 29 | self.work_dir: str = "" |
| 30 | self.proxy_url: str = "" |
| 31 | self._proxy: CapiProxy | None = None |
| 32 | self._client: CopilotClient | None = None |
| 33 | |
| 34 | async def setup(self): |
| 35 | from .testharness.context import get_cli_path_for_tests |
| 36 | |
| 37 | self.cli_path = get_cli_path_for_tests() |
| 38 | self.home_dir = tempfile.mkdtemp(prefix="copilot-token-config-") |
| 39 | self.work_dir = tempfile.mkdtemp(prefix="copilot-token-work-") |
| 40 | |
| 41 | self._proxy = CapiProxy() |
| 42 | self.proxy_url = await self._proxy.start() |
| 43 | |
| 44 | github_token = ( |
| 45 | "fake-token-for-e2e-tests" if os.environ.get("GITHUB_ACTIONS") == "true" else None |
| 46 | ) |
| 47 | |
| 48 | self._client = CopilotClient( |
| 49 | connection=RuntimeConnection.for_tcp(path=self.cli_path, connection_token=self.token), |
| 50 | working_directory=self.work_dir, |
| 51 | env=self.get_env(), |
| 52 | github_token=github_token, |
| 53 | ) |
| 54 | |
| 55 | # Trigger the spawn + connect handshake so the server is listening. |
| 56 | await self._client.start() |
| 57 | |
| 58 | async def teardown(self): |
| 59 | if self._client: |
| 60 | try: |
| 61 | await self._client.stop() |
| 62 | except Exception: |
| 63 | # Best-effort cleanup; ignore stop errors during teardown. |
| 64 | pass |
| 65 | self._client = None |
| 66 | if self._proxy: |
| 67 | await self._proxy.stop(skip_writing_cache=True) |
| 68 | self._proxy = None |
| 69 | if self.home_dir and os.path.exists(self.home_dir): |
| 70 | shutil.rmtree(self.home_dir, ignore_errors=True) |
| 71 | if self.work_dir and os.path.exists(self.work_dir): |
| 72 | shutil.rmtree(self.work_dir, ignore_errors=True) |
| 73 | |
| 74 | def get_env(self) -> dict: |
| 75 | env = os.environ.copy() |
| 76 | env.update( |
| 77 | { |
| 78 | "COPILOT_API_URL": self.proxy_url, |
| 79 | "COPILOT_HOME": self.home_dir, |
no outgoing calls
searching dependent graphs…