zerodds-py 1.0.0-rc.3

PyO3 bindings for the ZeroDDS DCPS API
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

"""Tests for §6.6 — ReadCondition + QueryCondition + state bitmask constants.

Verifies construction, attaching to a WaitSet, and SQL filter validation
for QueryCondition.
"""

from __future__ import annotations

import pytest

import zerodds

pytestmark = pytest.mark.skipif(
    not getattr(zerodds, "_CORE_AVAILABLE", False),
    reason="zerodds._core not compiled — maturin develop needed",
)


# ---------------------------------------------------------------------------
# State bitmask constants (§6.6, DDS 1.4 §2.2.2.5.1.4)
# ---------------------------------------------------------------------------


def test_sample_state_constants():
    assert zerodds.sample_state.NOT_READ == 1
    assert zerodds.sample_state.READ == 2
    assert zerodds.sample_state.ANY == 3


def test_view_state_constants():
    assert zerodds.view_state.NEW == 1
    assert zerodds.view_state.NOT_NEW == 2
    assert zerodds.view_state.ANY == 3


def test_instance_state_constants():
    assert zerodds.instance_state.ALIVE == 1
    assert zerodds.instance_state.NOT_ALIVE_DISPOSED == 2
    assert zerodds.instance_state.NOT_ALIVE_NO_WRITERS == 4
    assert zerodds.instance_state.NOT_ALIVE == 6
    assert zerodds.instance_state.ANY == 7


# ---------------------------------------------------------------------------
# ReadCondition
# ---------------------------------------------------------------------------


def test_read_condition_constructible():
    """§6.6 — ReadCondition with three bitmasks."""
    rc = zerodds.ReadCondition(
        zerodds.sample_state.ANY,
        zerodds.view_state.ANY,
        zerodds.instance_state.ALIVE,
    )
    assert rc.get_sample_state_mask() == 3
    assert rc.get_view_state_mask() == 3
    assert rc.get_instance_state_mask() == 1


def test_read_condition_modes():
    """§6.6 — state_check_mode='any'|'always'|'never' affects the trigger."""
    rc_any = zerodds.ReadCondition(3, 3, 7, "any")
    assert rc_any.get_trigger_value() is True

    rc_always = zerodds.ReadCondition(0, 0, 0, "always")
    assert rc_always.get_trigger_value() is True

    rc_never = zerodds.ReadCondition(3, 3, 7, "never")
    assert rc_never.get_trigger_value() is False


def test_read_condition_unknown_mode_raises():
    with pytest.raises(RuntimeError, match="state_check_mode"):
        zerodds.ReadCondition(3, 3, 7, "elsewhere")


def test_read_condition_attaches_to_waitset():
    rc = zerodds.ReadCondition(
        zerodds.sample_state.NOT_READ,
        zerodds.view_state.NEW,
        zerodds.instance_state.ALIVE,
    )
    ws = zerodds.WaitSet()
    ws.attach_read_condition(rc)
    # WaitSet.wait raises TimeoutError because the rc.trigger == "any" mode
    # does not take effect on the offline path; we only test that attach works.


# ---------------------------------------------------------------------------
# QueryCondition
# ---------------------------------------------------------------------------


def test_query_condition_constructible_simple_filter():
    """§6.6 — QueryCondition with a valid SQL92 filter."""
    qc = zerodds.QueryCondition(
        zerodds.sample_state.ANY,
        zerodds.view_state.ANY,
        zerodds.instance_state.ALIVE,
        "x > 5",
        [],
    )
    assert qc is not None


def test_query_condition_with_parameters():
    """§6.6 — QueryCondition accepts positional query parameters (%0, %1, ...)."""
    qc = zerodds.QueryCondition(
        zerodds.sample_state.ANY,
        zerodds.view_state.ANY,
        zerodds.instance_state.ANY,
        "color = %0 AND x > %1",
        ["RED", "10"],
    )
    assert qc is not None


def test_query_condition_invalid_sql_raises():
    """§6.6 — invalid SQL expression → RuntimeError in the constructor."""
    with pytest.raises(RuntimeError, match="QueryCondition"):
        zerodds.QueryCondition(
            zerodds.sample_state.ANY,
            zerodds.view_state.ANY,
            zerodds.instance_state.ANY,
            "this is not (valid SQL",
            [],
        )


def test_query_condition_attaches_to_waitset():
    qc = zerodds.QueryCondition(
        zerodds.sample_state.ANY,
        zerodds.view_state.ANY,
        zerodds.instance_state.ALIVE,
        "x < 100",
        [],
    )
    ws = zerodds.WaitSet()
    ws.attach_query_condition(qc)