oxillama-py 0.1.2

Python bindings for OxiLLaMa LLM inference engine
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90

"""Tests for CancellationToken — skipped when native extension is absent."""

from __future__ import annotations

import threading

import pytest


def _import_native():
    try:
        import oxillama_py.oxillama_py as _m  # type: ignore[import-untyped]

        return _m
    except ImportError:
        return None


_NATIVE = _import_native()
_SKIP = pytest.mark.skipif(
    _NATIVE is None, reason="Native extension not built (run `maturin develop`)"
)


@_SKIP
def test_cancellation_token_default_not_cancelled():
    ct = _NATIVE.CancellationToken()
    assert ct.is_cancelled() is False


@_SKIP
def test_cancellation_token_cancel():
    ct = _NATIVE.CancellationToken()
    ct.cancel()
    assert ct.is_cancelled() is True


@_SKIP
def test_cancellation_token_reset_after_cancel():
    ct = _NATIVE.CancellationToken()
    ct.cancel()
    assert ct.is_cancelled() is True
    ct.reset()
    assert ct.is_cancelled() is False


@_SKIP
def test_cancellation_token_multiple_cancels_idempotent():
    ct = _NATIVE.CancellationToken()
    ct.cancel()
    ct.cancel()
    assert ct.is_cancelled() is True


@_SKIP
def test_cancellation_token_thread_safety():
    """Cancelling from one thread should be visible from another."""
    ct = _NATIVE.CancellationToken()
    results = []

    def canceller():
        ct.cancel()

    def checker():
        # busy-wait up to 1 s
        import time

        deadline = time.monotonic() + 1.0
        while time.monotonic() < deadline:
            if ct.is_cancelled():
                results.append(True)
                return
            time.sleep(0.001)
        results.append(False)

    t_cancel = threading.Thread(target=canceller)
    t_check = threading.Thread(target=checker)
    t_check.start()
    t_cancel.start()
    t_cancel.join()
    t_check.join()

    assert results == [True]


@_SKIP
def test_cancellation_token_repr():
    ct = _NATIVE.CancellationToken()
    r = repr(ct)
    assert "CancellationToken" in r or "cancelled" in r.lower()