DX-1019: Read Your Writes

tests/commands/asyncio/bitmap/test_bitfield.py

@@ -29,11 +29,9 @@ async def test_set(async_redis: Redis) -> None:
     ) == [97]
 
     # With string offset.
-    assert await (
-        async_redis.bitfield("string_for_bitfield_set")
-        .set(encoding="u8", offset="#1", value=115)
-        .execute()
-    ) == [101]
+    assert await async_redis.bitfield("string_for_bitfield_set").set(
+        encoding="u8", offset="#1", value=115
+    ).execute() == [101]
 
     assert await execute_on_http(
         "BITFIELD", "string_for_bitfield_set", "GET", "u8", "#1"
@@ -43,22 +41,18 @@ async def test_set(async_redis: Redis) -> None:
 @mark.asyncio
 async def test_incrby(async_redis: Redis) -> None:
     # With integer offset.
-    assert await (
-        async_redis.bitfield("string_for_bitfield_incrby")
-        .incrby(encoding="u8", offset=0, increment=1)
-        .execute()
-    ) == [117]
+    assert await async_redis.bitfield("string_for_bitfield_incrby").incrby(
+        encoding="u8", offset=0, increment=1
+    ).execute() == [117]
 
     assert await execute_on_http(
         "BITFIELD", "string_for_bitfield_incrby", "GET", "u8", "0"
     ) == [117]
 
     # With string offset.
-    assert await (
-        async_redis.bitfield("string_for_bitfield_incrby")
-        .incrby(encoding="u8", offset="#1", increment=2)
-        .execute()
-    ) == [103]
+    assert await async_redis.bitfield("string_for_bitfield_incrby").incrby(
+        encoding="u8", offset="#1", increment=2
+    ).execute() == [103]
 
     assert await execute_on_http(
         "BITFIELD", "string_for_bitfield_incrby", "GET", "u8", "#1"
@@ -67,12 +61,9 @@ async def test_incrby(async_redis: Redis) -> None:
 
 @mark.asyncio
 async def test_chained_commands(async_redis: Redis) -> None:
-    assert await (
-        async_redis.bitfield("string_for_bitfield_chained_commands")
-        .set(encoding="u8", offset=0, value=97)
-        .incrby(encoding="u8", offset=0, increment=1)
-        .execute()
-    ) == [116, 98]
+    assert await async_redis.bitfield("string_for_bitfield_chained_commands").set(
+        encoding="u8", offset=0, value=97
+    ).incrby(encoding="u8", offset=0, increment=1).execute() == [116, 98]
 
     assert await execute_on_http(
         "BITFIELD", "string_for_bitfield_chained_commands", "GET", "u8", "0"
@@ -81,13 +72,12 @@ async def test_chained_commands(async_redis: Redis) -> None:
 
 @mark.asyncio
 async def test_overflow(async_redis: Redis) -> None:
-    assert await (
-        async_redis.bitfield("string_for_bitfield_overflow")
-        .incrby(encoding="i8", offset=100, increment=100)
-        .overflow("SAT")
-        .incrby(encoding="i8", offset=100, increment=100)
-        .execute()
-    ) == [100, 127]
+    assert await async_redis.bitfield("string_for_bitfield_overflow").incrby(
+        encoding="i8", offset=100, increment=100
+    ).overflow("SAT").incrby(encoding="i8", offset=100, increment=100).execute() == [
+        100,
+        127,
+    ]
 
     assert await execute_on_http(
         "BITFIELD", "string_for_bitfield_overflow", "GET", "i8", "100"

tests/commands/asyncio/bitmap/test_bitfield_ro.py

@@ -18,9 +18,6 @@ async def test_get(async_redis: Redis) -> None:
 
 @mark.asyncio
 async def test_chained_commands(async_redis: Redis) -> None:
-    assert await (
-        async_redis.bitfield_ro("string")
-        .get(encoding="u8", offset=0)
-        .get(encoding="u8", offset="#1")
-        .execute()
-    ) == [116, 101]
+    assert await async_redis.bitfield_ro("string").get(encoding="u8", offset=0).get(
+        encoding="u8", offset="#1"
+    ).execute() == [116, 101]

tests/commands/asyncio/test_asyncio_pipeline.py

@@ -8,9 +8,9 @@
 async def flush_db(async_redis: Redis):
     await async_redis.delete("rocket", "space", "marine")
 
+
 @pytest.mark.asyncio
 async def test_pipeline(async_redis: Redis):
-
     pipeline = async_redis.pipeline()
 
     pipeline.incr("rocket")
@@ -26,9 +26,9 @@ async def test_pipeline(async_redis: Redis):
     res = await pipeline.exec()
     assert res == [1, 2, 1, 3, 2, 4, "4", "2", None]
 
+
 @pytest.mark.asyncio
 async def test_multi(async_redis: Redis):
-
     pipeline = async_redis.multi()
 
     pipeline.incr("rocket")
@@ -45,9 +45,9 @@ async def test_multi(async_redis: Redis):
     res = await pipeline.exec()
     assert res == [1, 2, 1, 3, 2, 4, "4", "2", None]
 
+
 @pytest.mark.asyncio
 async def test_context_manager_usage(async_redis: Redis):
-
     async with async_redis.pipeline() as pipeline:
         pipeline.incr("rocket")
         pipeline.incr("rocket")
@@ -62,7 +62,7 @@ async def test_context_manager_usage(async_redis: Redis):
         pipeline.set("foo", "bar")
 
     assert result == [1, 2, 1, 3, 2, 4]
-    assert len(pipeline._command_stack) == 0 # pipeline is empty
+    assert len(pipeline._command_stack) == 0  # pipeline is empty
 
     # redis still works after pipeline is done
     get_result = await async_redis.get("rocket")
@@ -76,6 +76,7 @@ async def test_context_manager_usage(async_redis: Redis):
     res = await get_pipeline.exec()
     assert res == ["4", "2", None]
 
+
 @pytest.mark.asyncio
 async def test_context_manager_raise(async_redis: Redis):
     """
@@ -88,6 +89,7 @@ async def test_context_manager_raise(async_redis: Redis):
             pipeline.incr("rocket")
             raise Exception("test")
 
+
 @pytest.mark.asyncio
 async def test_run_pipeline_twice(async_redis: Redis):
     """

tests/commands/test_pipeline.py

@@ -7,8 +7,8 @@
 def flush_db(redis: Redis):
     redis.delete("rocket", "space", "marine")
 
-def test_pipeline(redis: Redis):
 
+def test_pipeline(redis: Redis):
     pipeline = redis.pipeline()
 
     pipeline.incr("rocket")
@@ -23,8 +23,8 @@ def test_pipeline(redis: Redis):
     res = pipeline.exec()
     assert res == [1, 2, 1, 3, 2, 4, "4", "2", None]
 
-def test_multi(redis: Redis):
 
+def test_multi(redis: Redis):
     pipeline = redis.multi()
 
     pipeline.incr("rocket")
@@ -41,8 +41,8 @@ def test_multi(redis: Redis):
     res = pipeline.exec()
     assert res == [1, 2, 1, 3, 2, 4, "4", "2", None]
 
-def test_context_manager_usage(redis: Redis):
 
+def test_context_manager_usage(redis: Redis):
     with redis.pipeline() as pipeline:
         pipeline.incr("rocket")
         pipeline.incr("rocket")
@@ -57,7 +57,7 @@ def test_context_manager_usage(redis: Redis):
         pipeline.set("foo", "bar")
 
     assert result == [1, 2, 1, 3, 2, 4]
-    assert len(pipeline._command_stack) == 0 # pipeline is empty
+    assert len(pipeline._command_stack) == 0  # pipeline is empty
 
     # redis still works after pipeline is done
     result = redis.get("rocket")
@@ -71,6 +71,7 @@ def test_context_manager_usage(redis: Redis):
     res = get_pipeline.exec()
     assert res == ["4", "2", None]
 
+
 def test_context_manager_raise(redis: Redis):
     """
     Check that exceptions in context aren't silently ignored
@@ -82,6 +83,7 @@ def test_context_manager_raise(redis: Redis):
             pipeline.incr("rocket")
             raise Exception("test")
 
+
 def test_run_pipeline_twice(redis: Redis):
     """
     Runs a pipeline twice

tests/conftest.py

@@ -122,12 +122,26 @@ def pytest_configure():
 
 
 @pytest_asyncio.fixture
-async def async_redis():
-    async with AsyncRedis.from_env(allow_telemetry=False) as redis:
-        yield redis
+async def async_redis(request):
+    if hasattr(request, "param"):
+        opts = request.param
+    else:
+        opts = {}
+
+    opts["allow_telemetry"] = False
+
+    async with AsyncRedis.from_env(**opts) as r:
+        yield r
 
 
 @pytest.fixture
-def redis():
-    with Redis.from_env(allow_telemetry=False) as redis:
-        yield redis
+def redis(request):
+    if hasattr(request, "param"):
+        opts = request.param
+    else:
+        opts = {}
+
+    opts["allow_telemetry"] = False
+
+    with Redis.from_env(**opts) as r:
+        yield r

tests/test_read_your_writes.py

@@ -0,0 +1,141 @@
+import pytest
+
+from upstash_redis import AsyncRedis, Redis
+
+
+@pytest.mark.parametrize("redis", [{"read_your_writes": True}], indirect=True)
+def test_should_update_sync_token_on_basic_request(redis: Redis):
+    initial_token = redis._sync_token
+    redis.set("key", "value")
+    updated_token = redis._sync_token
+    assert initial_token != updated_token
+
+
+@pytest.mark.parametrize("async_redis", [{"read_your_writes": True}], indirect=True)
+@pytest.mark.asyncio
+async def test_should_update_sync_token_on_basic_request_async(async_redis: AsyncRedis):
+    initial_token = async_redis._sync_token
+    await async_redis.set("key", "value")
+    updated_token = async_redis._sync_token
+    assert initial_token != updated_token
+
+
+@pytest.mark.parametrize("redis", [{"read_your_writes": True}], indirect=True)
+def test_should_update_sync_token_on_pipeline(redis: Redis):
+    initial_token = redis._sync_token
+
+    pipeline = redis.pipeline()
+    pipeline.set("key", "value")
+    pipeline.set("key2", "value2")
+    pipeline.exec()
+
+    updated_token = redis._sync_token
+    assert initial_token != updated_token
+
+
+@pytest.mark.parametrize("async_redis", [{"read_your_writes": True}], indirect=True)
+@pytest.mark.asyncio()
+async def test_should_update_sync_token_on_pipeline_async(async_redis: AsyncRedis):
+    initial_token = async_redis._sync_token
+
+    pipeline = async_redis.pipeline()
+    pipeline.set("key", "value")
+    pipeline.set("key2", "value2")
+    await pipeline.exec()
+
+    updated_token = async_redis._sync_token
+    assert initial_token != updated_token
+
+
+@pytest.mark.parametrize("redis", [{"read_your_writes": True}], indirect=True)
+def test_should_update_sync_token_on_multiexec(redis: Redis):
+    initial_token = redis._sync_token
+
+    multi = redis.multi()
+    multi.set("key", "value")
+    multi.set("key2", "value2")
+    multi.exec()
+
+    updated_token = redis._sync_token
+    assert initial_token != updated_token
+
+
+@pytest.mark.parametrize("async_redis", [{"read_your_writes": True}], indirect=True)
+@pytest.mark.asyncio
+async def test_should_update_sync_token_on_multiexec_async(async_redis: AsyncRedis):
+    initial_token = async_redis._sync_token
+
+    multi = async_redis.multi()
+    multi.set("key", "value")
+    multi.set("key2", "value2")
+    await multi.exec()
+
+    updated_token = async_redis._sync_token
+    assert initial_token != updated_token
+
+
+@pytest.mark.parametrize("redis", [{"read_your_writes": True}], indirect=True)
+def test_updates_after_successful_lua_script_call(redis: Redis):
+    initial_token = redis._sync_token
+
+    redis.eval(
+        """
+    redis.call('SET', 'mykey', 'myvalue')
+    return 1
+    """
+    )
+
+    updated_token = redis._sync_token
+    assert updated_token != initial_token
+
+
+@pytest.mark.parametrize("async_redis", [{"read_your_writes": True}], indirect=True)
+@pytest.mark.asyncio
+async def test_updates_after_successful_lua_script_call_async(async_redis: AsyncRedis):
+    initial_token = async_redis._sync_token
+
+    await async_redis.eval(
+        """
+    redis.call('SET', 'mykey', 'myvalue')
+    return 1
+    """
+    )
+
+    updated_token = async_redis._sync_token
+    assert updated_token != initial_token
+
+
+@pytest.mark.parametrize("redis", [{"read_your_writes": False}], indirect=True)
+def test_should_not_update_sync_state_with_opt_out_ryw(redis: Redis):
+    initial_token = redis._sync_token
+    redis.set("key", "value")
+    updated_token = redis._sync_token
+    assert updated_token == initial_token
+
+
+@pytest.mark.parametrize("async_redis", [{"read_your_writes": False}], indirect=True)
+@pytest.mark.asyncio
+async def test_should_not_update_sync_state_with_opt_out_ryw_async(
+    async_redis: AsyncRedis
+):
+    initial_token = async_redis._sync_token
+    await async_redis.set("key", "value")
+    updated_token = async_redis._sync_token
+    assert updated_token == initial_token
+
+
+def test_should_update_sync_state_with_default_behavior(redis: Redis):
+    initial_token = redis._sync_token
+    redis.set("key", "value")
+    updated_token = redis._sync_token
+    assert updated_token != initial_token
+
+
+@pytest.mark.asyncio
+async def test_should_update_sync_state_with_default_behavior_async(
+    async_redis: AsyncRedis
+):
+    initial_token = async_redis._sync_token
+    await async_redis.set("key", "value")
+    updated_token = async_redis._sync_token
+    assert updated_token != initial_token

upstash_redis/__init__.py

@@ -1,5 +1,6 @@
 __version__ = "1.1.0"
 
+from upstash_redis.asyncio.client import Redis as AsyncRedis
 from upstash_redis.client import Redis
 
-__all__ = ["Redis"]
+__all__ = ["AsyncRedis", "Redis"]