Test cache invalidation update insert delete (#4)

Author: janbjorge

db/init_db.sh

@@ -15,5 +15,6 @@ psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname testdb <<-EOSQL
     INSERT INTO sysconf (key, value) VALUES
     ('app_name', 'MyApplication'),
     ('app_version', '1.0.0'),
-    ('maintenance_mode', 'false');
+    ('maintenance_mode', 'false'),
+    ('updated_at', '2024-02-19 23:01:54.609243+00');
 EOSQL

tests/test_integration.py

@@ -1,11 +1,16 @@
 import asyncio
 import collections
+import datetime
 
 import asyncpg
 import pytest
 from pgcachewatch import cli, decorators, listeners, models, strategies
 
 
+def utcnow() -> datetime.datetime:
+    return datetime.datetime.now(tz=datetime.timezone.utc)
+
+
 async def test_1_install_triggers(
     monkeypatch: pytest.MonkeyPatch,
     pgconn: asyncpg.Connection,
@@ -54,7 +59,106 @@ async def fetch_sysconf() -> list:
     assert statistics["hit"] == N - 1
 
 
-async def test_3_uninstall_triggers(
+async def test_3_cache_invalidation_update(
+    pgconn: asyncpg.Connection,
+    pgpool: asyncpg.Pool,
+) -> None:
+    statistics = collections.Counter[str]()
+    listener = await listeners.PGEventQueue.create(
+        models.PGChannel("ch_pgcachewatch_table_change"),
+        pgconn=pgconn,
+    )
+
+    @decorators.cache(
+        strategy=strategies.Gready(listener=listener),
+        statistics_callback=lambda x: statistics.update([x]),
+    )
+    async def fetch_sysconf() -> list:
+        return await pgpool.fetch("SELECT * FROM sysconf")
+
+    async def blast() -> list:
+        before = await fetch_sysconf()
+        while (rv := await fetch_sysconf()) == before:
+            await asyncio.sleep(0.001)
+        return rv
+
+    blast_task = asyncio.create_task(blast())
+    await pgpool.execute(
+        "UPDATE sysconf set value = $1 where key = 'updated_at'",
+        utcnow().isoformat(),
+    )
+    await asyncio.wait_for(blast_task, 1)
+    # First fetch and update
+    assert statistics["miss"] == 2
+
+
+async def test_3_cache_invalidation_insert(
+    pgconn: asyncpg.Connection,
+    pgpool: asyncpg.Pool,
+) -> None:
+    statistics = collections.Counter[str]()
+    listener = await listeners.PGEventQueue.create(
+        models.PGChannel("ch_pgcachewatch_table_change"),
+        pgconn=pgconn,
+    )
+
+    @decorators.cache(
+        strategy=strategies.Gready(listener=listener),
+        statistics_callback=lambda x: statistics.update([x]),
+    )
+    async def fetch_sysconf() -> list:
+        return await pgpool.fetch("SELECT * FROM sysconf")
+
+    async def blast() -> list:
+        before = await fetch_sysconf()
+        while (rv := await fetch_sysconf()) == before:
+            await asyncio.sleep(0.001)
+        return rv
+
+    blast_task = asyncio.create_task(blast())
+    await pgpool.execute(
+        "INSERT INTO sysconf (key, value) VALUES ($1, $2);",
+        utcnow().isoformat(),
+        utcnow().isoformat(),
+    )
+    await asyncio.wait_for(blast_task, 1)
+    # First fetch and insert
+    assert statistics["miss"] == 2
+
+
+async def test_3_cache_invalidation_delete(
+    pgconn: asyncpg.Connection,
+    pgpool: asyncpg.Pool,
+) -> None:
+    statistics = collections.Counter[str]()
+    listener = await listeners.PGEventQueue.create(
+        models.PGChannel("ch_pgcachewatch_table_change"),
+        pgconn=pgconn,
+    )
+
+    @decorators.cache(
+        strategy=strategies.Gready(listener=listener),
+        statistics_callback=lambda x: statistics.update([x]),
+    )
+    async def fetch_sysconf() -> list:
+        return await pgpool.fetch("SELECT * FROM sysconf")
+
+    async def blast() -> list:
+        before = await fetch_sysconf()
+        while (rv := await fetch_sysconf()) == before:
+            await asyncio.sleep(0.001)
+        return rv
+
+    blast_task = asyncio.create_task(blast())
+    await pgpool.execute(
+        "DELETE FROM sysconf WHERE key ~ '^\\d{4}-\\d{2}-\\d{2}';",
+    )
+    await asyncio.wait_for(blast_task, 1)
+    # First fetch and insert
+    assert statistics["miss"] == 2
+
+
+async def test_4_uninstall_triggers(
     monkeypatch: pytest.MonkeyPatch,
     pgconn: asyncpg.Connection,
 ) -> None: