tests (unit/sqla): Add test for keyword filtering:

rel 458

tests/unit/__init__.py

@@ -0,0 +1,4 @@
+from pytest_twisted import init_asyncio_reactor
+
+
+init_asyncio_reactor()

tests/unit/test_memory_sqla.py

@@ -0,0 +1,67 @@
+#!/usr/bin/env python3
+
+# Libervia: an XMPP client
+# Copyright (C) 2009-2025 Jérôme Poisson (goffi@goffi.org)
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from unittest.mock import AsyncMock, MagicMock
+
+from pytest_twisted import ensureDeferred as ed
+from sqlalchemy.dialects import sqlite
+
+from libervia.backend.memory import sqla
+
+
+class TestSQLA:
+
+    @ed
+    async def test_keyword_filtering(self, monkeypatch):
+        monkeypatch.setattr(sqla, "profiles", {"test_profile": 1})
+        storage = sqla.Storage()
+
+        # We mock the SQLAlchemy session and result.
+        mock_session_context = AsyncMock()
+        mock_session_context.__aenter__ = AsyncMock()
+        mock_session_context.__aexit__ = AsyncMock()
+
+        storage.session = MagicMock(return_value=mock_session_context)
+
+        mock_session_instance = mock_session_context.__aenter__.return_value
+        mock_result = MagicMock()
+        mock_scalars = MagicMock()
+        mock_unique = MagicMock()
+        mock_unique.all.return_value = []
+        mock_scalars.unique.return_value = mock_unique
+        mock_result.scalars.return_value = mock_scalars
+        mock_session_instance.execute = AsyncMock(return_value=mock_result)
+
+        await storage.history_get(
+            from_jid=None,
+            to_jid=None,
+            filters={"keyword": "test"},
+            profile="test_profile"
+        )
+
+        # We check that the SQL query is as expected.
+        executed_stmt = mock_session_instance.execute.call_args[0][0]
+        compiled = executed_stmt.compile(
+            dialect=sqlite.dialect(),
+            compile_kwargs={"literal_binds": True}
+        )
+        sql = str(compiled)
+        assert "json_each" in sql
+        assert "keywords" in sql
+        assert "test" in sql
+        assert "EXISTS" in sql