Mercurial > libervia-backend

comparison tests/unit/test_memory_sqla.py @ 4374:90d476a80ce9

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests (unit/sqla): Add test for `keyword` filtering: rel 458
author Goffi <goffi@goffi.org>
date Fri, 06 Jun 2025 10:45:54 +0200
parents
children
comparison
equal deleted inserted replaced
4373:5e48ae998976 4374:90d476a80ce9
1 #!/usr/bin/env python3
2
3 # Libervia: an XMPP client
4 # Copyright (C) 2009-2025 Jérôme Poisson (goffi@goffi.org)
5
6 # This program is free software: you can redistribute it and/or modify
7 # it under the terms of the GNU Affero General Public License as published by
8 # the Free Software Foundation, either version 3 of the License, or
9 # (at your option) any later version.
10
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU Affero General Public License for more details.
15
16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18
19 from unittest.mock import AsyncMock, MagicMock
20
21 from pytest_twisted import ensureDeferred as ed
22 from sqlalchemy.dialects import sqlite
23
24 from libervia.backend.memory import sqla
25
26
27 class TestSQLA:
28
29 @ed
30 async def test_keyword_filtering(self, monkeypatch):
31 monkeypatch.setattr(sqla, "profiles", {"test_profile": 1})
32 storage = sqla.Storage()
33
34 # We mock the SQLAlchemy session and result.
35 mock_session_context = AsyncMock()
36 mock_session_context.__aenter__ = AsyncMock()
37 mock_session_context.__aexit__ = AsyncMock()
38
39 storage.session = MagicMock(return_value=mock_session_context)
40
41 mock_session_instance = mock_session_context.__aenter__.return_value
42 mock_result = MagicMock()
43 mock_scalars = MagicMock()
44 mock_unique = MagicMock()
45 mock_unique.all.return_value = []
46 mock_scalars.unique.return_value = mock_unique
47 mock_result.scalars.return_value = mock_scalars
48 mock_session_instance.execute = AsyncMock(return_value=mock_result)
49
50 await storage.history_get(
51 from_jid=None,
52 to_jid=None,
53 filters={"keyword": "test"},
54 profile="test_profile"
55 )
56
57 # We check that the SQL query is as expected.
58 executed_stmt = mock_session_instance.execute.call_args[0][0]
59 compiled = executed_stmt.compile(
60 dialect=sqlite.dialect(),
61 compile_kwargs={"literal_binds": True}
62 )
63 sql = str(compiled)
64 assert "json_each" in sql
65 assert "keywords" in sql
66 assert "test" in sql
67 assert "EXISTS" in sql