(self)
| 462 | multiprocessing.set_start_method("spawn", force=True) |
| 463 | |
| 464 | async def test_serve(self): # noqa: C901 |
| 465 | serve_process = multiprocessing.Process(target=run_serve, args=(self.config,)) |
| 466 | serve_process.start() |
| 467 | await asyncio.sleep(10) |
| 468 | |
| 469 | state_manager = StateManager( |
| 470 | path=self.config.checkpoint_job_dir, |
| 471 | explorer_name=self.config.explorer.name, |
| 472 | ) |
| 473 | |
| 474 | # wait for explorer initialization |
| 475 | for i in range(50): |
| 476 | try: |
| 477 | server_url = state_manager.load_explorer_server_url() |
| 478 | except Exception: |
| 479 | server_url = None |
| 480 | if server_url: |
| 481 | break |
| 482 | await asyncio.sleep(3) |
| 483 | if not server_url: |
| 484 | raise RuntimeError("Explorer server URL not found.") |
| 485 | # wait for server setup |
| 486 | for i in range(10): |
| 487 | try: |
| 488 | async with httpx.AsyncClient() as client: |
| 489 | response = await client.get(f"{server_url}/health") |
| 490 | if response.status_code == 200: |
| 491 | break |
| 492 | except Exception: |
| 493 | pass |
| 494 | await asyncio.sleep(2) |
| 495 | |
| 496 | task_num = 10 |
| 497 | apps = [] |
| 498 | for i in range(task_num): |
| 499 | app_process = multiprocessing.Process( |
| 500 | target=run_agent, args=(server_url, self.config.model.model_path, i % 2 == 0) |
| 501 | ) |
| 502 | apps.append(app_process) |
| 503 | app_process.start() |
| 504 | |
| 505 | for app in apps: |
| 506 | app.join(timeout=60) |
| 507 | self.assertFalse(app.is_alive()) |
| 508 | |
| 509 | finish_step = None |
| 510 | proxy_client = TrinityClient(proxy_url=server_url) |
| 511 | for i in range(20): |
| 512 | metrics = await proxy_client.get_metrics_async() |
| 513 | metrics_keys = list(metrics.keys()) |
| 514 | self.assertIn("explore_step_num", metrics_keys) |
| 515 | self.assertIn("rollout/total_experience_count", metrics_keys) |
| 516 | self.assertIn("rollout/model_0/total_request_count", metrics_keys) |
| 517 | self.assertIn("rollout/model_3/model_version", metrics_keys) |
| 518 | if not finish_step and metrics["rollout/total_experience_count"] == task_num: |
| 519 | finish_step = metrics["explore_step_num"] |
| 520 | await proxy_client.commit_async() |
| 521 | if finish_step and metrics["explore_step_num"] >= finish_step + 1: |
nothing calls this directly
no test coverage detected