Spaces:
Paused
Paused
File size: 1,019 Bytes
4ae946d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 |
from unittest.mock import MagicMock
import pytest
from sqlalchemy.orm import Session
from app.modules.analytics.router import get_temporal_flow
@pytest.mark.asyncio
async def test_get_temporal_flow_limit():
# Mock DB session
mock_db = MagicMock(spec=Session)
# Mock result proxy
mock_result = MagicMock()
mock_result.__iter__.return_value = []
mock_db.execute.return_value = mock_result
# Call the function
await get_temporal_flow(days=30, limit=5000, db=mock_db)
# Verify db.execute was called
args, kwargs = mock_db.execute.call_args
query_obj = args[0]
query_params = args[1]
query_str = str(query_obj)
print(f"Executed query: {query_str}")
print(f"Query params: {query_params}")
# Assert LIMIT clause is present
assert "LIMIT :limit" in query_str, "Query should contain LIMIT clause"
# Assert parameters
assert query_params["limit"] == 5000, "Limit parameter should be passed correctly"
assert "cutoff_date" in query_params
|