Function _cluster_config_lock

e2e/python/test_inference_routing.py:136–142

()

@contextmanager
def _cluster_config_lock() -> Iterator[None]:
    with open(_INFERENCE_CONFIG_LOCK, "a+", encoding="utf-8") as lock_file:
        fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
        try:
            yield
        finally:
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)


@pytest.fixture