diff --git a/news/6841.bugfix.md b/news/6841.bugfix.md new file mode 100644 index 00000000000..755e5824602 --- /dev/null +++ b/news/6841.bugfix.md @@ -0,0 +1 @@ +Avoid leaking temporary router dirtiness into linked/shared state event deltas. diff --git a/reflex/istate/shared.py b/reflex/istate/shared.py index a4d24de0657..d85f1d0221c 100644 --- a/reflex/istate/shared.py +++ b/reflex/istate/shared.py @@ -5,13 +5,13 @@ import logging from collections.abc import AsyncIterator from dataclasses import MISSING -from typing import TYPE_CHECKING, Self, TypeVar +from typing import TYPE_CHECKING, Any, Self, TypeVar from reflex_base.constants import ROUTER_DATA, ROUTER_VARS from reflex_base.event import Event, get_hydrate_event from reflex_base.registry import RegistrationContext from reflex_base.utils.exceptions import ReflexRuntimeError -from reflex_base.vars.base import _owner_state +from reflex_base.vars.base import _delta_value_key, _owner_state from reflex.istate.delta import _suppress_delta_recording from reflex.istate.manager.token import BaseStateToken @@ -114,14 +114,280 @@ async def _patch_state( linked_state.dirty_vars.update(linked_state.computed_vars) linked_state._mark_dirty() # Apply the updates into the existing state tree for rehydrate. + # For regular linked-state events this router dirtiness is temporary: + # it forces router-dependent computed vars to resolve for the patched + # tree, but should not leak into the event's final delta. root_state = original_state._get_root_state() + if ( + not full_delta + and not any( + root_state._var_dependencies.get(name) + for name in (*ROUTER_VARS, ROUTER_DATA) + ) + and all( + name in ROUTER_VARS or name == ROUTER_DATA + for name in root_state.dirty_vars + ) + and not root_state._potentially_dirty_states + ): + states_to_check = [root_state] + while states_to_check: + state = states_to_check.pop() + if state.computed_vars: + break + states_to_check.extend( + state.substates[name] + for name in state.dirty_substates | state._always_dirty_substates + if name in state.substates + ) + else: + # There is no router-dependent or otherwise pending computed + # work, so resolving this discarded delta cannot change state. + yield + return + dirty_state_snapshots: list[ + tuple[ + BaseState, + set[str], + set[str], + set[str], + bool, + ] + ] = [] + states_by_name: dict[str, BaseState] = {} + computed_var_cache_snapshots: dict[ + str, dict[str, tuple[bool, Any, bool, Any]] + ] = {} + if not full_delta: + states_to_snapshot = [root_state] + while states_to_snapshot: + state = states_to_snapshot.pop() + # Interval and always-dirty computed vars are valid output from + # this refresh and must not be mistaken for temporary router dirtiness. + computed_vars_to_preserve = state._expired_computed_vars().union( + state._always_dirty_computed_vars + ) + states_by_name[state.get_full_name()] = state + dirty_state_snapshots.append(( + state, + set(state.dirty_vars), + set(state.dirty_substates), + computed_vars_to_preserve, + state._was_touched, + )) + states_to_snapshot.extend(state.substates.values()) + computed_vars_to_snapshot = { + state.get_full_name(): ( + dirty_vars.intersection(state.computed_vars) + | computed_vars_to_preserve + | state._interval_computed_var_names + ) + for ( + state, + dirty_vars, + _, + computed_vars_to_preserve, + _, + ) in dirty_state_snapshots + } + # Track only caches that dirty propagation can invalidate, before + # marking router vars deletes those cached values. + pending_dependencies = [ + (state, name) + for ( + state, + dirty_vars, + _, + computed_vars_to_preserve, + _, + ) in dirty_state_snapshots + for name in dirty_vars | computed_vars_to_preserve + ] + pending_dependencies.extend((root_state, name) for name in ROUTER_VARS) + pending_dependencies.append((root_state, ROUTER_DATA)) + seen_dependencies: set[tuple[str, str]] = set() + while pending_dependencies: + state, name = pending_dependencies.pop() + dependency = (state.get_full_name(), name) + if dependency in seen_dependencies: + continue + seen_dependencies.add(dependency) + for target_name, computed_var_name in state._var_dependencies.get( + name, () + ): + target_state = states_by_name.get(target_name) + if target_state is None: + continue + target_vars = computed_vars_to_snapshot[target_name] + target_vars.add(computed_var_name) + pending_dependencies.append((target_state, computed_var_name)) + for ( + state, + _, + _, + _, + _, + ) in dirty_state_snapshots: + computed_var_cache_snapshots[state.get_full_name()] = { + name: ( + hasattr(state, state.computed_vars[name]._cache_attr), + getattr(state, state.computed_vars[name]._cache_attr, None), + hasattr(state, state.computed_vars[name]._last_updated_attr), + getattr( + state, state.computed_vars[name]._last_updated_attr, None + ), + ) + for name in computed_vars_to_snapshot[state.get_full_name()] + } root_state.dirty_vars.update(ROUTER_VARS) root_state.dirty_vars.add(ROUTER_DATA) root_state._mark_dirty() - # The delta is discarded: it is only resolved to refresh computed vars, - # so its values must not count as sent to the client. - with _suppress_delta_recording(): - await root_state._get_resolved_delta() + router_dirty_snapshots: list[ + tuple[ + BaseState, + set[str], + set[str], + set[str], + set[str], + set[str], + dict[str, tuple[bool, Any]], + dict[str, tuple[bool, Any, bool, Any]], + ] + ] = [] + if not full_delta: + for ( + state, + dirty_vars, + dirty_substates, + computed_vars_to_preserve, + _, + ) in dirty_state_snapshots: + computed_var_snapshots = computed_var_cache_snapshots[ + state.get_full_name() + ] + router_dirty_vars = ( + state.dirty_vars - dirty_vars - computed_vars_to_preserve + ) + router_computed_snapshots = { + name: ( + computed_var_snapshots[name][0], + _delta_value_key(computed_var_snapshots[name][1]) + if computed_var_snapshots[name][0] + else None, + ) + for name in router_dirty_vars.intersection(computed_var_snapshots) + } + router_dirty_snapshots.append(( + state, + dirty_vars, + dirty_substates, + router_dirty_vars, + state.dirty_substates - dirty_substates, + computed_vars_to_preserve, + router_computed_snapshots, + computed_var_snapshots, + )) + try: + # The delta is discarded: it is only resolved to refresh computed vars, + # so its values must not count as sent to the client. + with _suppress_delta_recording(): + await root_state._get_resolved_delta() + except BaseException: + if not full_delta: + for ( + state, + dirty_vars, + dirty_substates, + _, + was_touched, + ) in dirty_state_snapshots: + computed_var_snapshots = computed_var_cache_snapshots[ + state.get_full_name() + ] + state.dirty_vars = dirty_vars + state.dirty_substates = dirty_substates + state._was_touched = was_touched + for name, ( + had_cache, + cached_value, + had_last_updated, + last_updated, + ) in computed_var_snapshots.items(): + computed_var = state.computed_vars[name] + if had_cache: + setattr(state, computed_var._cache_attr, cached_value) + else: + with contextlib.suppress(AttributeError): + delattr(state, computed_var._cache_attr) + if had_last_updated: + setattr( + state, + computed_var._last_updated_attr, + last_updated, + ) + else: + with contextlib.suppress(AttributeError): + delattr(state, computed_var._last_updated_attr) + raise + else: + if not full_delta: + computed_refresh_states: list[BaseState] = [] + for ( + state, + dirty_vars, + dirty_substates, + router_dirty_vars, + _, + computed_vars_to_preserve, + router_computed_snapshots, + computed_var_snapshots, + ) in router_dirty_snapshots: + computed_vars_refreshed = set(computed_vars_to_preserve) | ( + state.dirty_vars - router_dirty_vars + ).intersection(state.computed_vars) + computed_vars_refreshed.update( + name + for name in state._interval_computed_var_names + if ( + computed_var_snapshots[name][2] + != hasattr( + state, + state.computed_vars[name]._last_updated_attr, + ) + or computed_var_snapshots[name][3] + != getattr( + state, + state.computed_vars[name]._last_updated_attr, + None, + ) + ) + ) + computed_vars_refreshed.update( + name + for name, ( + had_cache, + cached_value_key, + ) in router_computed_snapshots.items() + if hasattr(state, state.computed_vars[name]._cache_attr) + and ( + not had_cache + or cached_value_key + != _delta_value_key( + getattr(state, state.computed_vars[name]._cache_attr) + ) + ) + ) + if computed_vars_refreshed: + computed_refresh_states.append(state) + state.dirty_vars = ( + dirty_vars | computed_vars_to_preserve | computed_vars_refreshed + ) + state.dirty_substates = dirty_substates + for refreshed_state in computed_refresh_states: + state = refreshed_state + while state.parent_state is not None: + state.parent_state.dirty_substates.add(state.get_name()) + state = state.parent_state yield finally: original_parent_state.substates[state_name] = original_state diff --git a/tests/units/istate/manager/test_redis.py b/tests/units/istate/manager/test_redis.py index 5ed218e1ef4..96b616fe2bf 100644 --- a/tests/units/istate/manager/test_redis.py +++ b/tests/units/istate/manager/test_redis.py @@ -7,7 +7,7 @@ import time import uuid from collections import Counter -from collections.abc import AsyncGenerator +from collections.abc import AsyncGenerator, AsyncIterator from types import ModuleType from typing import Any from unittest.mock import AsyncMock, Mock @@ -424,6 +424,7 @@ async def test_oplock_contention_queue( state_manager_redis: StateManagerRedis, root_state: type[RedisTestState], event_log: list[dict[str, Any]], + monkeypatch: pytest.MonkeyPatch, ): """Test the oplock contention queue. @@ -431,6 +432,7 @@ async def test_oplock_contention_queue( state_manager_redis: The StateManagerRedis to test. root_state: The root state class. event_log: The redis event log. + monkeypatch: The pytest monkeypatch fixture. """ token = str(uuid.uuid4()) @@ -445,7 +447,6 @@ async def test_oplock_contention_queue( state_manager_2._oplock_enabled = True modify_started = asyncio.Event() - modify_2_started = asyncio.Event() modify_1_continue = asyncio.Event() modify_2_continue = asyncio.Event() @@ -460,7 +461,6 @@ async def modify_1(): async def modify_2(): await modify_started.wait() - modify_2_started.set() async with state_manager_2.modify_state( BaseStateToken(ident=token, cls=root_state), ) as new_state: @@ -470,7 +470,6 @@ async def modify_2(): async def modify_3(): await modify_started.wait() - modify_2_started.set() async with state_manager_2.modify_state( BaseStateToken(ident=token, cls=root_state), ) as new_state: @@ -481,8 +480,34 @@ async def modify_3(): task_1 = asyncio.create_task(modify_1()) task_2 = asyncio.create_task(modify_2()) task_3 = asyncio.create_task(modify_3()) + lock_key = state_manager_2._lock_key(BaseStateToken(ident=token, cls=root_state)) + queued_waiters = 0 + both_waiters_queued = asyncio.Event() + original_lock_waiter = StateManagerRedis._lock_waiter - await modify_2_started.wait() + @contextlib.asynccontextmanager + async def track_lock_waiter( + manager: StateManagerRedis, key: bytes + ) -> AsyncIterator[asyncio.Event]: + nonlocal queued_waiters + async with original_lock_waiter(manager, key) as lock_released_event: + if manager is state_manager_2 and key == lock_key: + queued_waiters += 1 + if queued_waiters == 2: + both_waiters_queued.set() + yield lock_released_event + + monkeypatch.setattr(StateManagerRedis, "_lock_waiter", track_lock_waiter) + + # Do not release the lock until both modifications have joined its queue. + try: + await asyncio.wait_for(both_waiters_queued.wait(), timeout=5) + except TimeoutError: + for task in (task_1, task_2, task_3): + task.cancel() + await asyncio.gather(task_1, task_2, task_3, return_exceptions=True) + await state_manager_2.close() + raise # Let modify 1 complete modify_1_continue.set() diff --git a/tests/units/istate/test_shared.py b/tests/units/istate/test_shared.py index e7cb2e73010..f5d127eace7 100644 --- a/tests/units/istate/test_shared.py +++ b/tests/units/istate/test_shared.py @@ -1,13 +1,19 @@ """Unit tests for shared state fan-out to other linked clients.""" import asyncio +import datetime import pickle from contextlib import asynccontextmanager +from typing import cast from unittest.mock import AsyncMock, Mock, patch import pytest +from reflex_base import constants +from reflex_base.constants.state import FIELD_MARKER import reflex as rx +from reflex.istate import shared as shared_module +from reflex.istate.data import RouterData, SessionData from reflex.istate.shared import ( SharedStateBaseInternal, _do_update_other_tokens, @@ -173,3 +179,463 @@ async def test_patch_state_recomputes_readers_after_restoring(): async with _patch_state(original_state=original, linked_state=linked): assert await reader.greeting == "linked" # pyright: ignore[reportAttributeAccessIssue] assert await reader.greeting == "private" # pyright: ignore[reportAttributeAccessIssue] + + +_PATCH_TEMPORARY_VAR = "temporary" +_PATCH_INTERVAL_VALUE_VAR = "interval_value" +_PATCH_ROUTER_INTERVAL_VALUE_VAR = "router_interval_value" +_PATCH_ROUTER_INTERVAL_DEPENDENT_VAR = "router_interval_dependent_value" +_PATCH_ROOT_VALUE_VAR = "root_value" +_PATCH_ROUTER_VALUE_VAR = "router_client_token" +_PATCH_EXISTING_SUBSTATE = "existing" + + +class _LinkedStatePatchRoot(BaseState): + """Root state for testing linked-state dirty propagation.""" + + value: int = 0 + + +class _LinkedStatePatchShared(_LinkedStatePatchRoot): + """Substate used to exercise _patch_state without full SharedState setup.""" + + counter: int = 0 + + @rx.var + def root_value(self) -> int: + return self.value + + +class _LinkedStatePatchIntervalRoot(BaseState): + """Root state for testing interval computed refreshes during patching.""" + + value: int = 0 + + +class _LinkedStatePatchIntervalShared(_LinkedStatePatchIntervalRoot): + """Substate used to exercise interval computed refreshes.""" + + counter: int = 0 + + @rx.var(interval=60) + def interval_value(self) -> int: + return self.value + + @rx.var(interval=60) + def router_interval_value(self) -> str: + return self.router.session.client_token + + @rx.var + def router_interval_dependent_value(self) -> str: + return self.router_interval_value + "-dependent" + + +class _LinkedStatePatchRouterRoot(BaseState): + """Root state with a computed value derived from the router.""" + + @rx.var + def router_client_token(self) -> str: + return self.router.session.client_token + + @rx.var + def unrelated_values(self) -> list[int]: + return list(range(1000)) + + +class _LinkedStatePatchRouterShared(_LinkedStatePatchRouterRoot): + """Substate used to exercise root router-dependent computations.""" + + counter: int = 0 + + +class _LinkedStatePatchNoComputedRoot(BaseState): + """Root state with no computed vars for the discarded-delta fast path.""" + + +class _LinkedStatePatchNoComputedShared(_LinkedStatePatchNoComputedRoot): + """Substate with no computed vars for the discarded-delta fast path.""" + + counter: int = 0 + + +@pytest.mark.asyncio +async def test_linked_state_event_does_not_dirty_root_state(): + """Linked-state events should not leak temporary router dirtiness.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + assert isinstance(private_state, _LinkedStatePatchShared) + assert isinstance(linked_state, _LinkedStatePatchShared) + + private_tree._clean() + + async with _patch_state(private_state, linked_state, full_delta=False): + linked_state.counter = 1 + + assert set(constants.ROUTER_VARS).isdisjoint(private_tree.dirty_vars) + assert constants.ROUTER_DATA not in private_tree.dirty_vars + assert private_tree.get_full_name() not in private_tree.get_delta() + + +@pytest.mark.asyncio +async def test_linked_state_patch_skips_discarded_delta_without_computed_vars(): + """Skip router refresh when no computed var or pending substate needs it.""" + private_tree = _LinkedStatePatchNoComputedRoot() + linked_tree = _LinkedStatePatchNoComputedRoot() + shared_state_name = _LinkedStatePatchNoComputedShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + resolve_delta = AsyncMock() + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_delta) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + + resolve_delta.assert_not_awaited() + assert not private_tree.dirty_vars + assert not private_tree.dirty_substates + + +@pytest.mark.asyncio +async def test_linked_state_patch_does_not_emit_unchanged_router_computed_var(): + """Refreshing a router computed var should not emit it when unchanged.""" + private_tree = _LinkedStatePatchRouterRoot() + linked_tree = _LinkedStatePatchRouterRoot() + + shared_state_name = _LinkedStatePatchRouterShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + original_value = private_tree.router_client_token + private_tree._clean() + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.router_client_token == original_value + assert _PATCH_ROUTER_VALUE_VAR not in private_tree.dirty_vars + assert private_tree.get_full_name() not in private_tree.get_delta() + + +@pytest.mark.asyncio +async def test_linked_state_patch_skips_unrelated_computed_value_keys( + monkeypatch: pytest.MonkeyPatch, +): + """Only router-dependent computed values need a cache comparison key.""" + private_tree = _LinkedStatePatchRouterRoot() + linked_tree = _LinkedStatePatchRouterRoot() + shared_state_name = _LinkedStatePatchRouterShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + unrelated_values = private_tree.unrelated_values + private_tree._clean() + delta_value_key = shared_module._delta_value_key + unrelated_key_calls = [] + + def track_unrelated_value_key(value): + if value is unrelated_values: + unrelated_key_calls.append(value) + return delta_value_key(value) + + monkeypatch.setattr(shared_module, "_delta_value_key", track_unrelated_value_key) + + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert unrelated_key_calls == [] + + +@pytest.mark.asyncio +async def test_linked_state_patch_restores_root_dirty_state_on_resolve_error(): + """Temporary root dirtiness should be cleaned if delta resolution fails.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + assert isinstance(private_state, _LinkedStatePatchShared) + assert isinstance(linked_state, _LinkedStatePatchShared) + + private_tree.value = 1 + private_tree.dirty_substates.add(_PATCH_EXISTING_SUBSTATE) + original_dirty_vars = set(private_tree.dirty_vars) + original_dirty_substates = set(private_tree.dirty_substates) + + async def raise_resolve_error(): + await asyncio.sleep(0) + msg = "delta resolution failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", raise_resolve_error) + + with pytest.raises(RuntimeError, match="delta resolution failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert private_tree.dirty_vars == original_dirty_vars + assert private_tree.dirty_substates == original_dirty_substates + assert private_tree.substates[shared_state_name] is private_state + assert linked_state.parent_state is linked_tree + + +@pytest.mark.asyncio +async def test_linked_state_patch_restores_descendant_dirty_state_on_resolve_error(): + """Temporary descendant dirtiness should be cleaned on resolution failure.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + async def raise_resolve_error(): + await asyncio.sleep(0) + linked_state.dirty_vars.add(_PATCH_TEMPORARY_VAR) + linked_state._mark_dirty() + msg = "descendant delta resolution failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", raise_resolve_error) + + with pytest.raises(RuntimeError, match="descendant delta resolution failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + pass + + assert linked_state.dirty_vars == set() + assert linked_state.dirty_substates == set() + assert private_tree.dirty_substates == set() + + +@pytest.mark.asyncio +async def test_linked_state_patch_restores_descendant_dirty_state_after_resolve(): + """Temporary descendant dirtiness should be cleaned after resolution.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + async def resolve_with_temporary_dirty_state(): + await asyncio.sleep(0) + linked_state.dirty_vars.add(_PATCH_TEMPORARY_VAR) + linked_state._mark_dirty() + return {} + + object.__setattr__( + private_tree, "_get_resolved_delta", resolve_with_temporary_dirty_state + ) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + + assert linked_state.dirty_vars == set() + assert linked_state.dirty_substates == set() + assert private_tree.dirty_substates == set() + + +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_interval_computed_refresh(): + """Interval computed vars refreshed during patching must be emitted.""" + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_with_dirty_child(): + linked_state._mark_dirty() + return await resolve_delta() + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_with_dirty_child) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + interval_delta = private_tree.get_delta() + + assert _PATCH_INTERVAL_VALUE_VAR in linked_state.dirty_vars + assert private_tree.dirty_substates == {linked_state.get_name()} + assert ( + _PATCH_INTERVAL_VALUE_VAR + FIELD_MARKER + in interval_delta[linked_state.get_full_name()] + ) + + +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_interval_refresh_during_async_router_var(): + """An interval var refreshed during async resolution must be emitted.""" + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = cast( + _LinkedStatePatchIntervalShared, private_tree.substates[shared_state_name] + ) + linked_state = cast( + _LinkedStatePatchIntervalShared, linked_tree.substates[shared_state_name] + ) + + assert linked_state.interval_value == linked_state.value + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_after_async_router_computation(): + await asyncio.sleep(0) + object.__setattr__(linked_state, "value", 1) + interval_var = _LinkedStatePatchIntervalShared.computed_vars[ + _PATCH_INTERVAL_VALUE_VAR + ] + setattr(linked_state, interval_var._last_updated_attr, datetime.datetime.min) + assert linked_state.interval_value == 1 + return await resolve_delta() + + object.__setattr__( + private_tree, + "_get_resolved_delta", + resolve_after_async_router_computation, + ) + + async with _patch_state(private_state, linked_state, full_delta=False): + interval_delta = private_tree.get_delta() + + assert _PATCH_INTERVAL_VALUE_VAR in linked_state.dirty_vars + assert ( + _PATCH_INTERVAL_VALUE_VAR + FIELD_MARKER + in interval_delta[linked_state.get_full_name()] + ) + + +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_downstream_of_refreshed_interval_var(): + """A downstream computed var of a refreshed interval var must be emitted.""" + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + private_tree.router = RouterData(session=SessionData(client_token="before")) + linked_tree.router = RouterData(session=SessionData(client_token="before")) + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = cast( + _LinkedStatePatchIntervalShared, private_tree.substates[shared_state_name] + ) + linked_state = cast( + _LinkedStatePatchIntervalShared, linked_tree.substates[shared_state_name] + ) + + assert linked_state.router_interval_dependent_value == "before-dependent" + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_after_router_computation(): + await asyncio.sleep(0) + private_tree.router = RouterData(session=SessionData(client_token="after")) + interval_var = _LinkedStatePatchIntervalShared.computed_vars[ + _PATCH_ROUTER_INTERVAL_VALUE_VAR + ] + setattr(linked_state, interval_var._last_updated_attr, datetime.datetime.min) + assert linked_state.router_interval_dependent_value == "after-dependent" + return await resolve_delta() + + object.__setattr__( + private_tree, + "_get_resolved_delta", + resolve_after_router_computation, + ) + + async with _patch_state(private_state, linked_state, full_delta=False): + interval_delta = private_tree.get_delta() + + assert _PATCH_ROUTER_INTERVAL_DEPENDENT_VAR in linked_state.dirty_vars + assert ( + _PATCH_ROUTER_INTERVAL_DEPENDENT_VAR + FIELD_MARKER + in interval_delta[linked_state.get_full_name()] + ) + + +@pytest.mark.asyncio +async def test_linked_state_patch_preserves_root_dependent_computed_refresh(): + """Computed vars invalidated through the root must remain in the delta.""" + private_tree = _LinkedStatePatchRoot() + linked_tree = _LinkedStatePatchRoot() + + shared_state_name = _LinkedStatePatchShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + + private_tree._clean() + + resolve_delta = private_tree._get_resolved_delta + + async def resolve_with_root_change(): + private_tree.value = 1 + return await resolve_delta() + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_with_root_change) + + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + root_dependent_delta = private_tree.get_delta() + + assert _PATCH_ROOT_VALUE_VAR in linked_state.dirty_vars + assert private_tree.dirty_substates == {linked_state.get_name()} + assert ( + _PATCH_ROOT_VALUE_VAR + FIELD_MARKER + in root_dependent_delta[linked_state.get_full_name()] + ) + + +@pytest.mark.asyncio +async def test_linked_state_patch_restores_computed_cache_on_resolve_error(): + """Failed resolution should not hide a partially refreshed computed value.""" + private_tree = _LinkedStatePatchIntervalRoot() + linked_tree = _LinkedStatePatchIntervalRoot() + shared_state_name = _LinkedStatePatchIntervalShared.get_name() + private_state = private_tree.substates[shared_state_name] + linked_state = linked_tree.substates[shared_state_name] + computed_var = _LinkedStatePatchIntervalShared.computed_vars[ + _PATCH_INTERVAL_VALUE_VAR + ] + cache_attr = computed_var._cache_attr + last_updated_attr = computed_var._last_updated_attr + cache_before = ( + hasattr(linked_state, cache_attr), + getattr(linked_state, cache_attr, None), + hasattr(linked_state, last_updated_attr), + getattr(linked_state, last_updated_attr, None), + ) + was_touched_before = linked_state._was_touched + + private_tree._clean() + resolve_delta = private_tree._get_resolved_delta + + async def resolve_then_fail(): + linked_state._mark_dirty() + linked_state._was_touched = True + await resolve_delta() + msg = "computed refresh failed" + raise RuntimeError(msg) + + object.__setattr__(private_tree, "_get_resolved_delta", resolve_then_fail) + + with pytest.raises(RuntimeError, match="computed refresh failed"): + async with _patch_state(private_state, linked_state, full_delta=False): + assert private_tree.substates[shared_state_name] is linked_state + + assert linked_state.dirty_vars == set() + assert linked_state._was_touched is was_touched_before + assert ( + hasattr(linked_state, cache_attr), + getattr(linked_state, cache_attr, None), + hasattr(linked_state, last_updated_attr), + getattr(linked_state, last_updated_attr, None), + ) == cache_before diff --git a/tests/units/utils/test_utils.py b/tests/units/utils/test_utils.py index 4ff41dac4cf..7aacad8027c 100644 --- a/tests/units/utils/test_utils.py +++ b/tests/units/utils/test_utils.py @@ -646,17 +646,17 @@ def test_initialize_requirements_txt_preserves_existing_requirements(tmp_path): assert not pyproject_file.exists() -def test_validate_app_name(tmp_path, mocker: MockerFixture): +def test_validate_app_name(tmp_path, monkeypatch: pytest.MonkeyPatch): """Test that an error is raised if the app name is reflex or if the name is not according to python package naming conventions. Args: tmp_path: Test working dir. - mocker: Pytest mocker object. + monkeypatch: Pytest monkeypatch fixture. """ reflex = tmp_path / "reflex" reflex.mkdir() - mocker.patch("os.getcwd", return_value=str(reflex)) + monkeypatch.chdir(reflex) with pytest.raises(SystemExit): prerequisites.validate_app_name()