Test that previously bound MutableProxy instances can be rebound correctly.
(
token: str, attached_mock_event_context: EventContext
)
| 4956 | |
| 4957 | @pytest.mark.asyncio |
| 4958 | async def test_rebind_mutable_proxy( |
| 4959 | token: str, attached_mock_event_context: EventContext |
| 4960 | ) -> None: |
| 4961 | """Test that previously bound MutableProxy instances can be rebound correctly.""" |
| 4962 | state_manager = attached_mock_event_context.state_manager |
| 4963 | |
| 4964 | async with state_manager.modify_state( |
| 4965 | BaseStateToken(ident=token, cls=MutableProxyState) |
| 4966 | ) as state: |
| 4967 | state.router = RouterData.from_router_data({ |
| 4968 | "query": {}, |
| 4969 | "token": token, |
| 4970 | "sid": "test_sid", |
| 4971 | }) |
| 4972 | assert isinstance(state, MutableProxyState) |
| 4973 | assert isinstance(state.data, MutableProxy) |
| 4974 | assert not isinstance(state.data, ImmutableMutableProxy) |
| 4975 | state_proxy = StateProxy(state) |
| 4976 | assert isinstance(state_proxy.data, ImmutableMutableProxy) |
| 4977 | async with state_proxy: |
| 4978 | # This assigns an ImmutableMutableProxy to data["a"]. |
| 4979 | state_proxy.data["a"] = state_proxy.data["b"] |
| 4980 | assert isinstance(state_proxy.data["a"], ImmutableMutableProxy) |
| 4981 | assert state_proxy.data["a"] is not state_proxy.data["b"] |
| 4982 | assert state_proxy.data["a"].__wrapped__ is state_proxy.data["b"].__wrapped__ |
| 4983 | |
| 4984 | # Rebinding with a non-proxy should return a MutableProxy object (not ImmutableMutableProxy). |
| 4985 | assert isinstance(state_proxy.__wrapped__.data["a"], MutableProxy) |
| 4986 | assert not isinstance(state_proxy.__wrapped__.data["a"], ImmutableMutableProxy) |
| 4987 | |
| 4988 | # Flush any oplock. |
| 4989 | await state_manager.close() |
| 4990 | |
| 4991 | new_state_proxy = StateProxy(state) |
| 4992 | assert state_proxy is not new_state_proxy |
| 4993 | assert new_state_proxy.data["a"]._self_state is new_state_proxy |
| 4994 | assert state_proxy.data["a"]._self_state is state_proxy |
| 4995 | assert state_proxy.__wrapped__.data["a"]._self_state is state_proxy.__wrapped__ |
| 4996 | |
| 4997 | async with state_proxy: |
| 4998 | state_proxy.data["a"].append(3) |
| 4999 | |
| 5000 | async with state_manager.modify_state( |
| 5001 | BaseStateToken(ident=token, cls=MutableProxyState) |
| 5002 | ) as state: |
| 5003 | assert isinstance(state, MutableProxyState) |
| 5004 | assert state.data["a"] == [2, 3] |
| 5005 | # Object identity persists across serialization, so data["b"] is also mutated. |
| 5006 | assert state.data["b"] == [2, 3] |
| 5007 | |
| 5008 | |
| 5009 | def test_override_base_method_skips_event_handler_wrapping(): |
nothing calls this directly
no test coverage detected