42 lines
1.5 KiB
Python
42 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from imagefind.accelerator import AcceleratorService
|
|
from imagefind.config import Settings
|
|
from imagefind.inference import InferenceSupervisor, IsolatedEmbeddingService
|
|
|
|
|
|
def test_hash_embeddings_remain_in_process_without_starting_worker(tmp_path: Path):
|
|
settings = Settings(data_dir=tmp_path, embedding_backend="hash")
|
|
settings.prepare()
|
|
accelerator = AcceleratorService(settings)
|
|
supervisor = InferenceSupervisor(settings, accelerator, idle_seconds=5)
|
|
embeddings = IsolatedEmbeddingService(settings, accelerator, supervisor)
|
|
|
|
vector = embeddings.encode_text("本地测试")
|
|
assert len(vector) == 512
|
|
assert supervisor.status()["running"] is False
|
|
supervisor.close()
|
|
|
|
|
|
def test_inference_worker_is_lazy_and_exits_after_idle_window(tmp_path: Path):
|
|
settings = Settings(data_dir=tmp_path, embedding_backend="auto")
|
|
settings.prepare()
|
|
accelerator = AcceleratorService(settings)
|
|
supervisor = InferenceSupervisor(settings, accelerator, idle_seconds=5)
|
|
try:
|
|
assert supervisor.status()["running"] is False
|
|
assert supervisor.call("reset", component="visual") is None
|
|
started = supervisor.status()
|
|
assert started["running"] is True
|
|
assert started["pid"]
|
|
|
|
deadline = time.monotonic() + 8
|
|
while supervisor.status()["running"] and time.monotonic() < deadline:
|
|
time.sleep(0.1)
|
|
assert supervisor.status()["running"] is False
|
|
finally:
|
|
supervisor.close()
|