Set up EngineService for testing
(cls)
| 33 | |
| 34 | @classmethod |
| 35 | def setUpClass(cls): |
| 36 | """Set up EngineService for testing""" |
| 37 | try: |
| 38 | # Create engine args for testing |
| 39 | engine_args = EngineArgs( |
| 40 | model=MODEL_NAME, |
| 41 | max_model_len=8192, |
| 42 | tensor_parallel_size=1, |
| 43 | engine_worker_queue_port=int(os.getenv("FD_ENGINE_QUEUE_PORT", "6778")), |
| 44 | cache_queue_port=int(os.getenv("FD_CACHE_QUEUE_PORT", "6779")), |
| 45 | ) |
| 46 | |
| 47 | # Create and start the engine service |
| 48 | cls.cfg = engine_args.create_engine_config() |
| 49 | cls.engine = EngineService(cls.cfg, start_queue=True, use_async_llm=True) |
| 50 | |
| 51 | # Start the engine service |
| 52 | cls.engine.start() |
| 53 | |
| 54 | except Exception as e: |
| 55 | print(f"Setting up EngineService failed: {e}") |
| 56 | raise |
| 57 | |
| 58 | @classmethod |
| 59 | def tearDownClass(cls): |
nothing calls this directly
no test coverage detected