67 lines
2.4 KiB
Python
67 lines
2.4 KiB
Python
#!/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
|