mirror of
https://github.com/barkeser2002/offline-db.git
synced 2026-09-25 04:19:53 +03:00
Barış Keserandgoogle-labs-jules[bot]
aa01c1ea2b
This commit adds a new `slow_query_logger` function in `core/utils.py` that wraps Django database executions and logs queries taking longer than 100ms. It uses `time.monotonic()` for robust timing. The logger is globally registered in `core/apps.py` via the `connection_created` signal with `weak=False` to ensure all new database connections are properly instrumented. Unit tests have been added to `core/tests/test_slow_query.py` and the development plan checkbox is updated. Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
import pytest
|
|
import time
|
|
import logging
|
|
from unittest.mock import patch
|
|
from django.db import connection
|
|
from core.utils import slow_query_logger
|
|
|
|
@pytest.fixture
|
|
def mock_execute():
|
|
def execute(sql, params, many, context):
|
|
return "result"
|
|
return execute
|
|
|
|
def test_slow_query_logger_fast_query(mock_execute, caplog):
|
|
# Setup caplog
|
|
caplog.set_level(logging.WARNING)
|
|
|
|
# Call with a fast query simulation
|
|
with patch('time.monotonic', side_effect=[0.0, 0.05]):
|
|
result = slow_query_logger(mock_execute, "SELECT 1", [], False, {})
|
|
|
|
assert result == "result"
|
|
assert "Slow query detected" not in caplog.text
|
|
|
|
def test_slow_query_logger_slow_query(mock_execute, caplog):
|
|
# Setup caplog
|
|
caplog.set_level(logging.WARNING)
|
|
|
|
# Call with a slow query simulation (> 100ms)
|
|
with patch('time.monotonic', side_effect=[0.0, 0.15]):
|
|
result = slow_query_logger(mock_execute, "SELECT 1", [], False, {})
|
|
|
|
assert result == "result"
|
|
assert "Slow query detected" in caplog.text
|
|
assert "SELECT 1" in caplog.text
|
|
|
|
@pytest.mark.django_db
|
|
def test_apps_ready():
|
|
# Verify the execute wrapper is attached to new connections via signal
|
|
from django.db import connection as thread_connection
|
|
# Force connection creation if not already created
|
|
thread_connection.ensure_connection()
|
|
assert slow_query_logger in thread_connection.execute_wrappers
|