fix(tests): use dependency_overrides for get_task_executor, not patch()
FastAPI resolves Depends(get_task_executor) against the function object it
captured when each route was decorated, at import time. patch('src.main.
get_task_executor') therefore never reaches an already-registered route —
confirmed empirically with a minimal FastAPI app before touching this repo's
tests. Every one of these tests fell through to the real dependency, which
called the real psycopg2.connect() and failed on postgres-shared's DNS (a
Docker-network name unreachable from a host process either way, so this
reproduced identically with and without host networking).
app.dependency_overrides is FastAPI's own supported mechanism for this, and
was already used correctly elsewhere in this suite (test_task_delete.py).
Added a shared override_task_executor fixture in conftest.py and switched
every get_task_executor patch() call site in test_api_comprehensive.py to it.
Source (src/main.py) is unchanged — Depends() is the idiomatic, correct
pattern; the tests were using an ineffective substitute for it.
This commit is contained in:
+17
-1
@@ -20,7 +20,7 @@ os.environ["GITEA_USER"] = "test-librarian"
|
|||||||
os.environ["GITEA_PASSWORD"] = "test-gitea-token"
|
os.environ["GITEA_PASSWORD"] = "test-gitea-token"
|
||||||
os.environ["REDIS_HOST"] = "redis-shared"
|
os.environ["REDIS_HOST"] = "redis-shared"
|
||||||
|
|
||||||
from src.main import app
|
from src.main import app, get_task_executor
|
||||||
from src.config import Settings, get_settings
|
from src.config import Settings, get_settings
|
||||||
|
|
||||||
|
|
||||||
@@ -44,6 +44,22 @@ def auth_headers(api_key: str) -> dict:
|
|||||||
return {"Authorization": f"Bearer {api_key}"}
|
return {"Authorization": f"Bearer {api_key}"}
|
||||||
|
|
||||||
|
|
||||||
|
# FastAPI resolves `Depends(get_task_executor)` against the function object it
|
||||||
|
# captured when each route was decorated, at import time. `unittest.mock.patch`
|
||||||
|
# on the module attribute `src.main.get_task_executor` therefore never reaches
|
||||||
|
# an already-registered route — the route keeps calling the original function.
|
||||||
|
# `app.dependency_overrides` is FastAPI's own supported mechanism for this
|
||||||
|
# (already used correctly in test_task_delete.py); this fixture centralizes it
|
||||||
|
# so call sites just need the executor mock they want installed.
|
||||||
|
@pytest.fixture
|
||||||
|
def override_task_executor() -> Generator[MagicMock, None, None]:
|
||||||
|
"""Install a mock TaskExecutor as the live dependency for this test only."""
|
||||||
|
mock_executor = MagicMock()
|
||||||
|
app.dependency_overrides[get_task_executor] = lambda: mock_executor
|
||||||
|
yield mock_executor
|
||||||
|
app.dependency_overrides.pop(get_task_executor, None)
|
||||||
|
|
||||||
|
|
||||||
# Synchronous test client
|
# Synchronous test client
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def client() -> Generator[TestClient, None, None]:
|
def client() -> Generator[TestClient, None, None]:
|
||||||
|
|||||||
+155
-167
@@ -3,7 +3,7 @@ Comprehensive API tests to improve coverage of main.py.
|
|||||||
"""
|
"""
|
||||||
import pytest
|
import pytest
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
from unittest.mock import patch, MagicMock
|
from unittest.mock import MagicMock
|
||||||
import json
|
import json
|
||||||
|
|
||||||
|
|
||||||
@@ -12,106 +12,101 @@ import json
|
|||||||
class TestTaskCRUDOperations:
|
class TestTaskCRUDOperations:
|
||||||
"""Comprehensive CRUD tests for task endpoints."""
|
"""Comprehensive CRUD tests for task endpoints."""
|
||||||
|
|
||||||
def test_list_tasks_empty(self, client: TestClient, auth_headers: dict):
|
def test_list_tasks_empty(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test listing tasks when none exist."""
|
"""Test listing tasks when none exist."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchall.return_value = []
|
||||||
mock_cursor.fetchall.return_value = []
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.get("/tasks", headers=auth_headers)
|
response = client.get("/tasks", headers=auth_headers)
|
||||||
|
|
||||||
if response.status_code == 200:
|
if response.status_code == 200:
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert "tasks" in data
|
assert "tasks" in data
|
||||||
assert data["count"] == 0
|
assert data["count"] == 0
|
||||||
|
|
||||||
def test_list_tasks_with_filters(self, client: TestClient, auth_headers: dict):
|
def test_list_tasks_with_filters(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test listing tasks with enabled and service filters."""
|
"""Test listing tasks with enabled and service filters."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchall.return_value = []
|
||||||
mock_cursor.fetchall.return_value = []
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.get(
|
response = client.get(
|
||||||
"/tasks?enabled=true&service=scheduler",
|
"/tasks?enabled=true&service=scheduler",
|
||||||
headers=auth_headers
|
headers=auth_headers
|
||||||
)
|
)
|
||||||
|
|
||||||
# Should execute filtered query
|
# Should execute filtered query
|
||||||
assert mock_cursor.execute.called
|
assert mock_cursor.execute.called
|
||||||
call_args = str(mock_cursor.execute.call_args)
|
call_args = str(mock_cursor.execute.call_args)
|
||||||
assert "enabled" in call_args.lower() or response.status_code in [200, 500]
|
assert "enabled" in call_args.lower() or response.status_code in [200, 500]
|
||||||
|
|
||||||
def test_get_task_details_not_found(self, client: TestClient, auth_headers: dict):
|
def test_get_task_details_not_found(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test getting details for non-existent task."""
|
"""Test getting details for non-existent task."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchone.return_value = None
|
||||||
mock_cursor.fetchone.return_value = None
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.get("/tasks/nonexistent", headers=auth_headers)
|
response = client.get("/tasks/nonexistent", headers=auth_headers)
|
||||||
|
|
||||||
assert response.status_code == 404
|
assert response.status_code == 404
|
||||||
|
|
||||||
def test_update_task(self, client: TestClient, auth_headers: dict):
|
def test_update_task(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test updating a task."""
|
"""Test updating a task."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchone.return_value = {
|
||||||
mock_cursor.fetchone.return_value = {
|
"task_name": "test",
|
||||||
"task_name": "test",
|
"priority": 60
|
||||||
"priority": 60
|
}
|
||||||
}
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.put(
|
response = client.put(
|
||||||
"/tasks/test",
|
"/tasks/test",
|
||||||
headers=auth_headers,
|
headers=auth_headers,
|
||||||
json={"priority": 60}
|
json={"priority": 60}
|
||||||
)
|
)
|
||||||
|
|
||||||
# Should have attempted update
|
# Should have attempted update
|
||||||
assert mock_cursor.execute.called
|
assert mock_cursor.execute.called
|
||||||
|
|
||||||
def test_update_task_not_found(self, client: TestClient, auth_headers: dict):
|
def test_update_task_not_found(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test updating non-existent task."""
|
"""Test updating non-existent task."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchone.return_value = None
|
||||||
mock_cursor.fetchone.return_value = None
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.put(
|
response = client.put(
|
||||||
"/tasks/nonexistent",
|
"/tasks/nonexistent",
|
||||||
headers=auth_headers,
|
headers=auth_headers,
|
||||||
json={"priority": 60}
|
json={"priority": 60}
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 404
|
assert response.status_code == 404
|
||||||
|
|
||||||
def test_update_task_no_fields(self, client: TestClient, auth_headers: dict):
|
def test_update_task_no_fields(self, client: TestClient, auth_headers: dict):
|
||||||
"""Test updating task with no valid fields."""
|
"""Test updating task with no valid fields."""
|
||||||
@@ -123,39 +118,37 @@ class TestTaskCRUDOperations:
|
|||||||
|
|
||||||
assert response.status_code == 400
|
assert response.status_code == 400
|
||||||
|
|
||||||
def test_delete_task(self, client: TestClient, auth_headers: dict):
|
def test_delete_task(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test deleting a task."""
|
"""Test deleting a task."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchone.return_value = ("test_task",)
|
||||||
mock_cursor.fetchone.return_value = ("test_task",)
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.delete("/tasks/test_task", headers=auth_headers)
|
response = client.delete("/tasks/test_task", headers=auth_headers)
|
||||||
|
|
||||||
if response.status_code == 200:
|
if response.status_code == 200:
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert "deleted successfully" in data["message"].lower()
|
assert "deleted successfully" in data["message"].lower()
|
||||||
|
|
||||||
def test_delete_task_not_found(self, client: TestClient, auth_headers: dict):
|
def test_delete_task_not_found(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test deleting non-existent task."""
|
"""Test deleting non-existent task."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchone.return_value = None
|
||||||
mock_cursor.fetchone.return_value = None
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.delete("/tasks/nonexistent", headers=auth_headers)
|
response = client.delete("/tasks/nonexistent", headers=auth_headers)
|
||||||
|
|
||||||
assert response.status_code == 404
|
assert response.status_code == 404
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.api
|
@pytest.mark.api
|
||||||
@@ -163,40 +156,38 @@ class TestTaskCRUDOperations:
|
|||||||
class TestTriggerEndpoint:
|
class TestTriggerEndpoint:
|
||||||
"""Tests for task trigger endpoint."""
|
"""Tests for task trigger endpoint."""
|
||||||
|
|
||||||
def test_trigger_disabled_task(self, client: TestClient, auth_headers: dict):
|
def test_trigger_disabled_task(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test triggering a disabled task."""
|
"""Test triggering a disabled task."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchone.return_value = {
|
||||||
mock_cursor.fetchone.return_value = {
|
"task_name": "test",
|
||||||
"task_name": "test",
|
"enabled": False
|
||||||
"enabled": False
|
}
|
||||||
}
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.post("/tasks/test/trigger", headers=auth_headers)
|
response = client.post("/tasks/test/trigger", headers=auth_headers)
|
||||||
|
|
||||||
assert response.status_code == 400
|
assert response.status_code == 400
|
||||||
|
|
||||||
def test_trigger_nonexistent_task(self, client: TestClient, auth_headers: dict):
|
def test_trigger_nonexistent_task(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test triggering a task that doesn't exist."""
|
"""Test triggering a task that doesn't exist."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchone.return_value = None
|
||||||
mock_cursor.fetchone.return_value = None
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.post("/tasks/nonexistent/trigger", headers=auth_headers)
|
response = client.post("/tasks/nonexistent/trigger", headers=auth_headers)
|
||||||
|
|
||||||
assert response.status_code == 404
|
assert response.status_code == 404
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.api
|
@pytest.mark.api
|
||||||
@@ -254,58 +245,55 @@ class TestLegacyEndpoints:
|
|||||||
class TestExecutionFiltering:
|
class TestExecutionFiltering:
|
||||||
"""Tests for execution history filtering."""
|
"""Tests for execution history filtering."""
|
||||||
|
|
||||||
def test_filter_by_service(self, client: TestClient, auth_headers: dict):
|
def test_filter_by_service(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test filtering executions by service."""
|
"""Test filtering executions by service."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchall.return_value = []
|
||||||
mock_cursor.fetchall.return_value = []
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.get(
|
response = client.get(
|
||||||
"/executions?service=scheduler",
|
"/executions?service=scheduler",
|
||||||
headers=auth_headers
|
headers=auth_headers
|
||||||
)
|
)
|
||||||
|
|
||||||
assert mock_cursor.execute.called
|
assert mock_cursor.execute.called
|
||||||
|
|
||||||
def test_filter_by_status(self, client: TestClient, auth_headers: dict):
|
def test_filter_by_status(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test filtering executions by status."""
|
"""Test filtering executions by status."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchall.return_value = []
|
||||||
mock_cursor.fetchall.return_value = []
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.get(
|
response = client.get(
|
||||||
"/executions?status=success",
|
"/executions?status=success",
|
||||||
headers=auth_headers
|
headers=auth_headers
|
||||||
)
|
)
|
||||||
|
|
||||||
assert mock_cursor.execute.called
|
assert mock_cursor.execute.called
|
||||||
|
|
||||||
def test_custom_limit(self, client: TestClient, auth_headers: dict):
|
def test_custom_limit(self, client: TestClient, auth_headers: dict, override_task_executor: MagicMock):
|
||||||
"""Test custom limit for executions."""
|
"""Test custom limit for executions."""
|
||||||
with patch('src.main.get_task_executor') as mock_executor:
|
mock_conn = MagicMock()
|
||||||
mock_conn = MagicMock()
|
mock_cursor = MagicMock()
|
||||||
mock_cursor = MagicMock()
|
mock_cursor.fetchall.return_value = []
|
||||||
mock_cursor.fetchall.return_value = []
|
|
||||||
|
|
||||||
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
|
||||||
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
override_task_executor.get_db_connection.return_value.__enter__ = MagicMock(return_value=mock_conn)
|
||||||
mock_executor.return_value.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
override_task_executor.get_db_connection.return_value.__exit__ = MagicMock(return_value=None)
|
||||||
|
|
||||||
response = client.get("/executions?limit=50", headers=auth_headers)
|
response = client.get("/executions?limit=50", headers=auth_headers)
|
||||||
|
|
||||||
if response.status_code == 200:
|
if response.status_code == 200:
|
||||||
data = response.json()
|
data = response.json()
|
||||||
assert data["limit"] == 50
|
assert data["limit"] == 50
|
||||||
|
|||||||
Reference in New Issue
Block a user