tests/test_tokenizer_threads.py

"""Two jobs tokenizing with the same model at once must not fail with "Already borrowed" (Hugging Face fast
tokenizers are not thread-safe; runtime.py gives each one a lock). Live Space, 2026-10-19."""
import threading

import pytest

import runtime  # noqa: F401  (installs the locks)


def test_one_tokenizer_shared_by_many_jobs():
    transformers = pytest.importorskip("transformers")
    try:
        tok = transformers.AutoTokenizer.from_pretrained("distilroberta-base")
    except Exception:
        pytest.skip("tokenizer not available offline")
    errors, text = [], ["The quick brown fox jumps over the lazy dog. " * 20] * 16

    def job(i):
        try:
            for _ in range(40):
                tok(text, padding="max_length" if i % 2 else "longest", truncation=True,
                    max_length=64 + 32 * (i % 3))
        except Exception as err:
            errors.append(repr(err))
    threads = [threading.Thread(target=job, args=(i,)) for i in range(8)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    assert errors == []