Add unit test cases for new registry/context/processor modules

Add unit test cases for reflex.istate.manager.token

Author: masenf

tests/units/istate/manager/test_token.py

@@ -0,0 +1,178 @@
+"""Tests for StateToken, BaseStateToken, and from_legacy_token."""
+
+import io
+import pickle
+
+import pytest
+
+from reflex.istate.manager.token import BaseStateToken, StateToken
+
+
+def test_state_token_str():
+    """__str__ encodes ident and cls into 'ident/module.Class' format."""
+    token = StateToken(ident="abc-123", cls=int)
+    assert str(token) == "abc-123/builtins.int"
+
+
+def test_state_token_str_escapes_slashes():
+    """Slashes in ident or cls name are percent-encoded."""
+    token = StateToken(ident="a/b", cls=int)
+    result = str(token)
+    assert "%2F" in result
+    assert "/" in result
+
+
+def test_state_token_with_cls():
+    """with_cls returns a new token with updated cls, leaving the original unchanged."""
+    token = StateToken(ident="tok", cls=int)
+    new = token.with_cls(bool)
+    assert new.cls is bool
+    assert new.ident == "tok"
+    assert token.cls is int
+
+
+def test_state_token_serialize_deserialize_roundtrip():
+    """serialize/deserialize with data= round-trips through pickle."""
+    value = {"key": [1, 2, 3]}
+    data = StateToken.serialize(value)
+    assert isinstance(data, bytes)
+    assert StateToken.deserialize(data=data) == value
+
+
+def test_state_token_deserialize_from_fp():
+    """Deserialize with fp= reads from a file-like object."""
+    value = "hello"
+    buf = io.BytesIO(pickle.dumps(value))
+    assert StateToken.deserialize(fp=buf) == value
+
+
+def test_state_token_deserialize_neither_raises():
+    """Deserialize with neither data nor fp raises ValueError."""
+    with pytest.raises(ValueError, match="Only one"):
+        StateToken.deserialize()
+
+
+def test_state_token_get_and_reset_touched_state():
+    """Default implementation always returns True."""
+    assert StateToken.get_and_reset_touched_state("anything") is True
+
+
+def test_base_state_token_str(clean_registration_context):
+    """__str__ uses 'ident_full_name' format.
+
+    Args:
+        clean_registration_context: A fresh, empty registration context.
+    """
+    from reflex.state import BaseState
+
+    class TokState(BaseState):
+        pass
+
+    token = BaseStateToken(ident="client-abc", cls=TokState)
+    result = str(token)
+    assert result.startswith("client-abc_")
+    assert TokState.get_full_name() in result
+
+
+def test_base_state_token_with_cls(clean_registration_context):
+    """with_cls returns a BaseStateToken (not a plain StateToken).
+
+    Args:
+        clean_registration_context: A fresh, empty registration context.
+    """
+    from reflex.state import BaseState
+
+    class A(BaseState):
+        pass
+
+    class B(BaseState):
+        pass
+
+    token = BaseStateToken(ident="tok", cls=A)
+    new = token.with_cls(B)
+    assert isinstance(new, BaseStateToken)
+    assert new.cls is B
+
+
+def test_base_state_token_serialize_deserialize(clean_registration_context):
+    """BaseStateToken serialization uses BaseState._serialize/_deserialize.
+
+    Args:
+        clean_registration_context: A fresh, empty registration context.
+    """
+    from reflex.state import BaseState
+
+    class SerState(BaseState):
+        x: int = 42
+
+    state = SerState()
+    data = BaseStateToken.serialize(state)
+    assert isinstance(data, bytes)
+    restored = BaseStateToken.deserialize(data=data)
+    assert isinstance(restored, SerState)
+    assert restored.x == 42
+
+
+def test_base_state_token_get_and_reset_touched(clean_registration_context):
+    """get_and_reset_touched_state returns the touched flag and resets it.
+
+    Args:
+        clean_registration_context: A fresh, empty registration context.
+    """
+    from reflex.state import BaseState
+
+    class TouchState(BaseState):
+        x: int = 0
+
+    state = TouchState()
+    state._was_touched = True
+    assert BaseStateToken.get_and_reset_touched_state(state) is True
+    assert state._was_touched is False
+    assert BaseStateToken.get_and_reset_touched_state(state) is False
+
+
+def test_from_legacy_token(clean_registration_context):
+    """from_legacy_token parses 'ident_state.path' into a BaseStateToken.
+
+    Args:
+        clean_registration_context: A fresh, empty registration context.
+    """
+    from reflex.state import BaseState
+
+    class LegacyRoot(BaseState):
+        pass
+
+    full_name = LegacyRoot.get_full_name()
+    legacy_str = f"my-client-token_{full_name}"
+
+    token = BaseStateToken.from_legacy_token(legacy_str, root_state=LegacyRoot)
+    assert token.ident == "my-client-token"
+    assert token.cls is LegacyRoot
+
+
+def test_from_legacy_token_substate(clean_registration_context):
+    """from_legacy_token resolves a substate path.
+
+    Args:
+        clean_registration_context: A fresh, empty registration context.
+    """
+    from reflex.state import BaseState
+
+    class LegRoot(BaseState):
+        pass
+
+    class LegChild(LegRoot):
+        pass
+
+    full_name = LegChild.get_full_name()
+    legacy_str = f"tok_{full_name}"
+
+    token = BaseStateToken.from_legacy_token(legacy_str, root_state=LegRoot)
+    assert token.ident == "tok"
+    assert token.cls is LegChild
+
+
+def test_from_legacy_token_none_root_raises():
+    """from_legacy_token with root_state=None raises ValueError."""
+    with pytest.raises(ValueError, match="Root state must be provided"):
+        BaseStateToken.from_legacy_token("tok_some.state", root_state=None)

tests/units/reflex_core/__init__.py

@@ -0,0 +1,94 @@
+"""Tests for BaseContext."""
+
+import dataclasses
+
+import pytest
+from reflex_core._internal.context.base import BaseContext
+
+
+@dataclasses.dataclass(frozen=True, kw_only=True, slots=True)
+class _TestContext(BaseContext):
+    """Minimal BaseContext subclass for unit testing."""
+
+    label: str = "test"
+
+
+def test_get_without_set_raises():
+    """get() raises LookupError when no context is set."""
+    with pytest.raises(LookupError):
+        _TestContext.get()
+
+
+def test_set_and_get():
+    """set() makes the context retrievable via get()."""
+    ctx = _TestContext(label="a")
+    token = _TestContext.set(ctx)
+    try:
+        assert _TestContext.get() is ctx
+    finally:
+        _TestContext.reset(token)
+
+
+def test_reset_restores_previous():
+    """reset() restores the previously active context."""
+    outer = _TestContext(label="outer")
+    outer_tok = _TestContext.set(outer)
+    try:
+        inner = _TestContext(label="inner")
+        inner_tok = _TestContext.set(inner)
+        assert _TestContext.get() is inner
+        _TestContext.reset(inner_tok)
+        assert _TestContext.get() is outer
+    finally:
+        _TestContext.reset(outer_tok)
+
+
+def test_context_manager_enter_exit():
+    """__enter__ sets the context and __exit__ removes it."""
+    ctx = _TestContext(label="cm")
+    with ctx as entered:
+        assert entered is ctx
+        assert _TestContext.get() is ctx
+    with pytest.raises(LookupError):
+        _TestContext.get()
+
+
+def test_context_manager_nesting():
+    """Nested context managers restore the outer context on inner exit."""
+    outer = _TestContext(label="outer")
+    inner = _TestContext(label="inner")
+    with outer:
+        assert _TestContext.get().label == "outer"
+        with inner:
+            assert _TestContext.get().label == "inner"
+        assert _TestContext.get().label == "outer"
+
+
+def test_double_enter_raises():
+    """Entering the same context instance twice raises RuntimeError."""
+    ctx = _TestContext(label="double")
+    with ctx, pytest.raises(RuntimeError, match="already attached"):
+        ctx.__enter__()
+
+
+def test_ensure_context_attached():
+    """ensure_context_attached raises when not entered, succeeds when entered."""
+    ctx = _TestContext(label="ensure")
+    with pytest.raises(RuntimeError, match="must be entered"):
+        ctx.ensure_context_attached()
+    with ctx:
+        ctx.ensure_context_attached()
+
+
+def test_subclasses_have_independent_context_vars():
+    """Two BaseContext subclasses do not share their ContextVar."""
+
+    @dataclasses.dataclass(frozen=True, kw_only=True, slots=True)
+    class _OtherContext(BaseContext):
+        value: int = 0
+
+    ctx_a = _TestContext(label="a")
+    ctx_b = _OtherContext(value=42)
+    with ctx_a, ctx_b:
+        assert _TestContext.get().label == "a"
+        assert _OtherContext.get().value == 42