wehub-resource-sync
7a0da7932b
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Has been cancelled
CodeQL Advanced / Analyze (javascript-typescript) (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Has been cancelled
Docker Tests (Consolidated) / Accessibility Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Unit Tests (push) Has been cancelled
Docker Tests (Consolidated) / LLM Example Tests (push) Has been cancelled
Docker Tests (Consolidated) / Production Image Smoke Test (push) Has been cancelled
Docker Tests (Consolidated) / Infrastructure Tests (push) Has been cancelled
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Has been cancelled
Backwards Compatibility / Verify Encryption Constants (push) Has been cancelled
Backwards Compatibility / PyPI Version Compatibility (push) Has been cancelled
Backwards Compatibility / Database Migration Tests (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Docker Tests (Consolidated) / detect-changes (push) Has been cancelled
Docker Tests (Consolidated) / Build Test Image (push) Has been cancelled
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Has been cancelled
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Has been cancelled
1052 lines
36 KiB
Python
1052 lines
36 KiB
Python
"""
|
|
Comprehensive tests for news/flask_api.py
|
|
|
|
Tests cover:
|
|
- safe_error_message function
|
|
- get_user_id function
|
|
- News feed endpoint
|
|
- Subscription endpoints
|
|
- Folder management endpoints
|
|
- Error handling
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
from flask import Flask, jsonify
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
"""Create a Flask app for testing."""
|
|
app = Flask(__name__)
|
|
app.config["SECRET_KEY"] = "test-secret-key"
|
|
app.config["WTF_CSRF_ENABLED"] = False
|
|
app.config["TESTING"] = True
|
|
|
|
# Import and register the blueprint
|
|
from local_deep_research.news.flask_api import news_api_bp
|
|
|
|
app.register_blueprint(news_api_bp, url_prefix="/news/api")
|
|
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Create a test client."""
|
|
return app.test_client()
|
|
|
|
|
|
class TestSafeErrorMessage:
|
|
"""Tests for safe_error_message function."""
|
|
|
|
def test_value_error(self):
|
|
"""Test handling of ValueError."""
|
|
from local_deep_research.news.flask_api import safe_error_message
|
|
|
|
error = ValueError("sensitive internal message")
|
|
result = safe_error_message(error, "test context")
|
|
|
|
assert result == "Invalid input provided"
|
|
assert "sensitive" not in result
|
|
|
|
def test_key_error(self):
|
|
"""Test handling of KeyError."""
|
|
from local_deep_research.news.flask_api import safe_error_message
|
|
|
|
error = KeyError("missing_key")
|
|
result = safe_error_message(error, "test context")
|
|
|
|
assert result == "Required data missing"
|
|
|
|
def test_type_error(self):
|
|
"""Test handling of TypeError."""
|
|
from local_deep_research.news.flask_api import safe_error_message
|
|
|
|
error = TypeError("type mismatch")
|
|
result = safe_error_message(error, "test context")
|
|
|
|
assert result == "Invalid data format"
|
|
|
|
def test_generic_error(self):
|
|
"""Test handling of generic error."""
|
|
from local_deep_research.news.flask_api import safe_error_message
|
|
|
|
error = RuntimeError("internal error")
|
|
result = safe_error_message(error, "doing something")
|
|
|
|
assert "An error occurred" in result
|
|
assert "doing something" in result
|
|
|
|
def test_generic_error_no_context(self):
|
|
"""Test handling of generic error without context."""
|
|
from local_deep_research.news.flask_api import safe_error_message
|
|
|
|
error = RuntimeError("internal error")
|
|
result = safe_error_message(error)
|
|
|
|
assert result == "An error occurred"
|
|
|
|
|
|
class TestGetUserId:
|
|
"""Tests for get_user_id function."""
|
|
|
|
def test_get_user_id_authenticated(self, app):
|
|
"""Test getting user ID when authenticated."""
|
|
from local_deep_research.news.flask_api import get_user_id
|
|
|
|
with app.app_context():
|
|
# current_user is imported inside get_user_id, so patch at source
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.current_user",
|
|
return_value="testuser",
|
|
):
|
|
result = get_user_id()
|
|
assert result == "testuser"
|
|
|
|
def test_get_user_id_not_authenticated(self, app):
|
|
"""Test getting user ID when not authenticated."""
|
|
from local_deep_research.news.flask_api import get_user_id
|
|
|
|
with app.app_context():
|
|
with patch(
|
|
"local_deep_research.web.auth.decorators.current_user",
|
|
return_value=None,
|
|
):
|
|
result = get_user_id()
|
|
assert result is None
|
|
|
|
|
|
class TestNewsBlueprintImport:
|
|
"""Tests for news blueprint import."""
|
|
|
|
def test_blueprint_exists(self):
|
|
"""Test that news API blueprint exists."""
|
|
from local_deep_research.news.flask_api import news_api_bp
|
|
|
|
assert news_api_bp is not None
|
|
assert news_api_bp.name == "news_api"
|
|
assert news_api_bp.url_prefix == "/api"
|
|
|
|
|
|
class TestNewsFeedEndpoint:
|
|
"""Tests for news feed endpoint."""
|
|
|
|
def test_feed_endpoint_exists(self, client):
|
|
"""Test that feed endpoint exists."""
|
|
try:
|
|
response = client.get("/news/api/feed")
|
|
# Route exists - any response is valid (may require auth)
|
|
assert response.status_code == 401, response.status_code
|
|
except Exception:
|
|
# If app context fails, that's okay - we're testing route existence
|
|
pass
|
|
|
|
|
|
class TestSubscriptionEndpoints:
|
|
"""Tests for subscription endpoints."""
|
|
|
|
def test_subscribe_no_data(self, client):
|
|
"""Test subscribe endpoint without data."""
|
|
try:
|
|
response = client.post(
|
|
"/news/api/subscribe",
|
|
content_type="application/json",
|
|
)
|
|
# Route exists - any response is valid
|
|
assert response.status_code == 401, response.status_code
|
|
except Exception:
|
|
pass
|
|
|
|
def test_subscribe_invalid_json(self, client):
|
|
"""Test subscribe endpoint with invalid JSON."""
|
|
try:
|
|
response = client.post(
|
|
"/news/api/subscribe",
|
|
data="not json",
|
|
content_type="application/json",
|
|
)
|
|
# Route exists - any response is valid
|
|
assert response.status_code == 401, response.status_code
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class TestFolderEndpoints:
|
|
"""Tests for folder management endpoints."""
|
|
|
|
def test_folders_endpoint_requires_auth(self, client):
|
|
"""Test that folders endpoint requires authentication."""
|
|
response = client.get("/news/api/subscription/folders")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 401, response.status_code
|
|
|
|
def test_create_folder_requires_auth(self, client):
|
|
"""Test that create folder endpoint requires authentication."""
|
|
response = client.post(
|
|
"/news/api/subscription/folders",
|
|
json={"name": "Test Folder"},
|
|
content_type="application/json",
|
|
)
|
|
|
|
# Should require auth
|
|
assert response.status_code == 401, response.status_code
|
|
|
|
|
|
class TestSchedulerEndpoints:
|
|
"""Tests for scheduler endpoints."""
|
|
|
|
def test_scheduler_status_requires_auth(self, client):
|
|
"""Test that scheduler status endpoint exists."""
|
|
try:
|
|
response = client.get("/news/api/scheduler/status")
|
|
# Route may or may not exist - any response is valid
|
|
assert response.status_code == 401, response.status_code
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
class TestRecommenderEndpoints:
|
|
"""Tests for recommender endpoints."""
|
|
|
|
def test_recommender_status_requires_auth(self, client):
|
|
"""Test that recommender status endpoint requires authentication."""
|
|
response = client.get("/news/api/recommender/status")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestSubscriptionRunEndpoint:
|
|
"""Tests for subscription run endpoint."""
|
|
|
|
def test_run_subscription_requires_auth(self, client):
|
|
"""Test that run subscription endpoint requires authentication."""
|
|
response = client.post("/news/api/subscription/sub123/run")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestGetSubscription:
|
|
"""Tests for get subscription endpoint."""
|
|
|
|
def test_get_subscription_requires_auth(self, client):
|
|
"""Test that get subscription endpoint requires authentication."""
|
|
response = client.get("/news/api/subscription/sub123")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestUpdateSubscription:
|
|
"""Tests for update subscription endpoint."""
|
|
|
|
def test_update_subscription_requires_auth(self, client):
|
|
"""Test that update subscription endpoint requires authentication."""
|
|
response = client.put(
|
|
"/news/api/subscription/sub123",
|
|
json={"name": "Updated"},
|
|
content_type="application/json",
|
|
)
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestDeleteSubscription:
|
|
"""Tests for delete subscription endpoint."""
|
|
|
|
def test_delete_subscription_requires_auth(self, client):
|
|
"""Test that delete subscription endpoint requires authentication."""
|
|
response = client.delete("/news/api/subscription/sub123")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestNewsCardInteractions:
|
|
"""Tests for news card interaction endpoints."""
|
|
|
|
def test_dismiss_card_requires_auth(self, client):
|
|
"""Test that dismiss card endpoint requires authentication."""
|
|
response = client.post("/news/api/card/card123/dismiss")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
def test_bookmark_card_requires_auth(self, client):
|
|
"""Test that bookmark card endpoint requires authentication."""
|
|
response = client.post("/news/api/card/card123/bookmark")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
def test_rate_card_requires_auth(self, client):
|
|
"""Test that rate card endpoint requires authentication."""
|
|
response = client.post(
|
|
"/news/api/card/card123/rate",
|
|
json={"rating": 5},
|
|
content_type="application/json",
|
|
)
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestSubscriptionsList:
|
|
"""Tests for subscriptions list endpoint."""
|
|
|
|
def test_subscriptions_list_requires_auth(self, client):
|
|
"""Test that subscriptions list endpoint requires authentication."""
|
|
response = client.get("/news/api/subscriptions")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestNotificationEndpoints:
|
|
"""Tests for notification endpoints."""
|
|
|
|
def test_test_notification_requires_auth(self, client):
|
|
"""Test that test notification endpoint requires authentication."""
|
|
response = client.post(
|
|
"/news/api/notifications/test",
|
|
json={"service_url": "mailto://test@example.com"},
|
|
content_type="application/json",
|
|
)
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestRefreshEndpoint:
|
|
"""Tests for refresh endpoint."""
|
|
|
|
def test_refresh_feed_requires_auth(self, client):
|
|
"""Test that refresh feed endpoint requires authentication."""
|
|
response = client.post("/news/api/refresh")
|
|
|
|
# Should require auth
|
|
assert response.status_code == 404, response.status_code
|
|
|
|
|
|
class TestSchedulerControlRequired:
|
|
"""Tests for scheduler_control_required decorator (PR #2035).
|
|
|
|
The decorator gates global scheduler control endpoints behind
|
|
the news.scheduler.allow_api_control setting, returning 403
|
|
when disabled.
|
|
"""
|
|
|
|
def test_decorator_allows_when_setting_enabled(self, app):
|
|
"""Decorated function executes when setting is True."""
|
|
from local_deep_research.news.flask_api import (
|
|
scheduler_control_required,
|
|
)
|
|
|
|
with app.app_context():
|
|
with patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=True,
|
|
):
|
|
|
|
@scheduler_control_required
|
|
def dummy_view():
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
response, status_code = dummy_view()
|
|
assert status_code == 200
|
|
|
|
def test_decorator_blocks_when_setting_disabled(self, app):
|
|
"""Decorated function returns 403 when setting is False."""
|
|
from local_deep_research.news.flask_api import (
|
|
scheduler_control_required,
|
|
)
|
|
|
|
with app.test_request_context():
|
|
with patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
):
|
|
|
|
@scheduler_control_required
|
|
def dummy_view():
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
response, status_code = dummy_view()
|
|
assert status_code == 403
|
|
|
|
def test_decorator_returns_error_message_when_disabled(self, app):
|
|
"""403 response includes informative error message."""
|
|
from local_deep_research.news.flask_api import (
|
|
scheduler_control_required,
|
|
)
|
|
|
|
with app.test_request_context():
|
|
with patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
):
|
|
|
|
@scheduler_control_required
|
|
def dummy_view():
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
response, status_code = dummy_view()
|
|
data = response.get_json()
|
|
assert "error" in data
|
|
assert "disabled" in data["error"].lower()
|
|
|
|
def test_decorator_checks_correct_setting_key(self, app):
|
|
"""Decorator checks news.scheduler.allow_api_control setting."""
|
|
from local_deep_research.news.flask_api import (
|
|
scheduler_control_required,
|
|
)
|
|
|
|
with app.app_context():
|
|
with patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=True,
|
|
) as mock_setting:
|
|
|
|
@scheduler_control_required
|
|
def dummy_view():
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
dummy_view()
|
|
mock_setting.assert_called_once_with(
|
|
"news.scheduler.allow_api_control", False
|
|
)
|
|
|
|
def test_decorator_preserves_function_name(self):
|
|
"""Decorator preserves the wrapped function's name."""
|
|
from local_deep_research.news.flask_api import (
|
|
scheduler_control_required,
|
|
)
|
|
|
|
@scheduler_control_required
|
|
def my_endpoint():
|
|
pass
|
|
|
|
assert my_endpoint.__name__ == "my_endpoint"
|
|
|
|
def test_decorator_passes_args_to_wrapped_function(self, app):
|
|
"""Decorator forwards positional and keyword arguments."""
|
|
from local_deep_research.news.flask_api import (
|
|
scheduler_control_required,
|
|
)
|
|
|
|
with app.app_context():
|
|
with patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=True,
|
|
):
|
|
|
|
@scheduler_control_required
|
|
def dummy_view(a, b, key=None):
|
|
return jsonify({"a": a, "b": b, "key": key}), 200
|
|
|
|
response, status_code = dummy_view(1, 2, key="val")
|
|
assert status_code == 200
|
|
data = response.get_json()
|
|
assert data["a"] == 1
|
|
assert data["b"] == 2
|
|
assert data["key"] == "val"
|
|
|
|
def test_decorator_does_not_call_wrapped_when_disabled(self, app):
|
|
"""Wrapped function is never invoked when setting is disabled."""
|
|
from local_deep_research.news.flask_api import (
|
|
scheduler_control_required,
|
|
)
|
|
|
|
call_tracker = {"called": False}
|
|
|
|
with app.test_request_context():
|
|
with patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
):
|
|
|
|
@scheduler_control_required
|
|
def dummy_view():
|
|
call_tracker["called"] = True
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
dummy_view()
|
|
assert call_tracker["called"] is False
|
|
|
|
|
|
class TestSchedulerEndpointGating:
|
|
"""Integration tests verifying scheduler endpoints are properly gated.
|
|
|
|
Tests that mutating scheduler endpoints (start, stop, check-now,
|
|
cleanup-now) return 403 when the allow_api_control setting is
|
|
disabled, while read-only endpoints (status, users, stats) remain
|
|
accessible.
|
|
"""
|
|
|
|
GATED_ENDPOINTS = [
|
|
"/news/api/scheduler/start",
|
|
"/news/api/scheduler/stop",
|
|
"/news/api/scheduler/check-now",
|
|
"/news/api/scheduler/cleanup-now",
|
|
]
|
|
|
|
NON_GATED_ENDPOINTS = [
|
|
"/news/api/scheduler/status",
|
|
"/news/api/scheduler/users",
|
|
"/news/api/scheduler/stats",
|
|
]
|
|
|
|
def _auth_session(self, client):
|
|
"""Inject a valid session so login_required passes."""
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = "testuser"
|
|
|
|
def test_gated_endpoints_return_403_when_disabled(self, client, app):
|
|
"""Mutating scheduler endpoints return 403 when control disabled."""
|
|
self._auth_session(client)
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
):
|
|
mock_db.connections = {"testuser": True}
|
|
|
|
for endpoint in self.GATED_ENDPOINTS:
|
|
response = client.post(endpoint)
|
|
assert response.status_code == 403, (
|
|
f"{endpoint} should return 403 when control disabled, "
|
|
f"got {response.status_code}"
|
|
)
|
|
|
|
def test_read_only_endpoints_not_gated(self, client, app):
|
|
"""Read-only scheduler endpoints are not blocked by the setting."""
|
|
self._auth_session(client)
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
):
|
|
mock_db.connections = {"testuser": True}
|
|
|
|
for endpoint in self.NON_GATED_ENDPOINTS:
|
|
response = client.get(endpoint)
|
|
# Should NOT be 403 — these are read-only and not gated
|
|
assert response.status_code != 403, (
|
|
f"{endpoint} should not be gated by allow_api_control, "
|
|
f"but got 403"
|
|
)
|
|
|
|
|
|
class TestSchedulerStatsScoping:
|
|
"""Tests for scheduler stats endpoint data scoping (follow-up to PR #2035).
|
|
|
|
Verifies that GET /scheduler/stats and /scheduler/users only expose
|
|
the current user's data unless allow_api_control is enabled.
|
|
"""
|
|
|
|
def _auth_session(self, client, username="testuser"):
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = username
|
|
|
|
def _mock_scheduler(self):
|
|
"""Create a mock scheduler with multi-user data."""
|
|
scheduler = MagicMock()
|
|
scheduler.is_running = True
|
|
scheduler.user_sessions = {
|
|
"alice": {
|
|
"last_activity": MagicMock(
|
|
isoformat=lambda: "2026-01-01T00:00:00"
|
|
),
|
|
"scheduled_jobs": {"job1"},
|
|
},
|
|
"bob": {
|
|
"last_activity": MagicMock(
|
|
isoformat=lambda: "2026-01-02T00:00:00"
|
|
),
|
|
"scheduled_jobs": {"job2", "job3"},
|
|
},
|
|
}
|
|
scheduler._credential_store.retrieve.return_value = "secret"
|
|
|
|
# Mock APScheduler jobs with username as first arg
|
|
alice_job = MagicMock()
|
|
alice_job.id = "alice_check_42"
|
|
alice_job.name = "Alice check"
|
|
alice_job.args = ("alice", 42)
|
|
alice_job.next_run_time = None
|
|
alice_job.trigger = "interval"
|
|
|
|
bob_job = MagicMock()
|
|
bob_job.id = "bob_document_processing"
|
|
bob_job.name = "Bob doc"
|
|
bob_job.args = ("bob", 1)
|
|
bob_job.next_run_time = None
|
|
bob_job.trigger = "interval"
|
|
|
|
overdue_alice = MagicMock()
|
|
overdue_alice.id = "overdue_alice_99_ts"
|
|
overdue_alice.name = "Overdue Alice"
|
|
overdue_alice.args = ("alice", 99)
|
|
overdue_alice.next_run_time = None
|
|
overdue_alice.trigger = "date"
|
|
|
|
scheduler.scheduler = MagicMock()
|
|
scheduler.scheduler.get_jobs.return_value = [
|
|
alice_job,
|
|
bob_job,
|
|
overdue_alice,
|
|
]
|
|
|
|
return scheduler
|
|
|
|
def test_stats_no_side_effect(self, client, app):
|
|
"""GET /scheduler/stats must NOT call _schedule_user_subscriptions."""
|
|
self._auth_session(client, "alice")
|
|
mock_sched = self._mock_scheduler()
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"alice": True}
|
|
response = client.get("/news/api/scheduler/stats")
|
|
assert response.status_code == 200
|
|
mock_sched._schedule_user_subscriptions.assert_not_called()
|
|
|
|
def test_stats_scoped_to_current_user(self, client, app):
|
|
"""Stats returns only current user's session when control disabled."""
|
|
self._auth_session(client, "alice")
|
|
mock_sched = self._mock_scheduler()
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"alice": True}
|
|
response = client.get("/news/api/scheduler/stats")
|
|
data = response.get_json()
|
|
|
|
# Only alice's session, not bob's
|
|
assert "alice" in data["user_sessions"]
|
|
assert "bob" not in data["user_sessions"]
|
|
|
|
def test_stats_scoped_jobs_to_current_user(self, client, app):
|
|
"""Stats returns only current user's APScheduler jobs when control disabled."""
|
|
self._auth_session(client, "alice")
|
|
mock_sched = self._mock_scheduler()
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"alice": True}
|
|
response = client.get("/news/api/scheduler/stats")
|
|
data = response.get_json()
|
|
|
|
job_ids = [j["id"] for j in data["apscheduler_jobs"]]
|
|
assert "alice_check_42" in job_ids
|
|
assert "overdue_alice_99_ts" in job_ids
|
|
assert "bob_document_processing" not in job_ids
|
|
|
|
def test_stats_shows_all_when_control_enabled(self, client, app):
|
|
"""Stats returns all users and jobs when allow_api_control is True."""
|
|
self._auth_session(client, "alice")
|
|
mock_sched = self._mock_scheduler()
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=True,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"alice": True}
|
|
response = client.get("/news/api/scheduler/stats")
|
|
data = response.get_json()
|
|
|
|
assert "alice" in data["user_sessions"]
|
|
assert "bob" in data["user_sessions"]
|
|
|
|
job_ids = [j["id"] for j in data["apscheduler_jobs"]]
|
|
assert "bob_document_processing" in job_ids
|
|
|
|
def test_users_scoped_to_current_user(self, client, app):
|
|
"""GET /scheduler/users returns only current user when control disabled."""
|
|
self._auth_session(client, "alice")
|
|
mock_sched = self._mock_scheduler()
|
|
mock_sched.get_user_sessions_summary.return_value = [
|
|
{"user_id": "alice", "last_activity": "2026-01-01T00:00:00"},
|
|
{"user_id": "bob", "last_activity": "2026-01-02T00:00:00"},
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"alice": True}
|
|
response = client.get("/news/api/scheduler/users")
|
|
data = response.get_json()
|
|
|
|
assert data["active_users"] == 1
|
|
user_ids = [u["user_id"] for u in data["users"]]
|
|
assert "alice" in user_ids
|
|
assert "bob" not in user_ids
|
|
|
|
def test_users_shows_all_when_control_enabled(self, client, app):
|
|
"""GET /scheduler/users returns all users when control enabled."""
|
|
self._auth_session(client, "alice")
|
|
mock_sched = self._mock_scheduler()
|
|
mock_sched.get_user_sessions_summary.return_value = [
|
|
{"user_id": "alice", "last_activity": "2026-01-01T00:00:00"},
|
|
{"user_id": "bob", "last_activity": "2026-01-02T00:00:00"},
|
|
]
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=True,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"alice": True}
|
|
response = client.get("/news/api/scheduler/users")
|
|
data = response.get_json()
|
|
|
|
assert data["active_users"] == 2
|
|
user_ids = [u["user_id"] for u in data["users"]]
|
|
assert "alice" in user_ids
|
|
assert "bob" in user_ids
|
|
|
|
def test_error_message_no_env_var_name(self, app):
|
|
"""403 error message must not expose internal env var names."""
|
|
from local_deep_research.news.flask_api import (
|
|
scheduler_control_required,
|
|
)
|
|
|
|
with app.test_request_context():
|
|
with app.app_context():
|
|
with patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
):
|
|
with patch(
|
|
"local_deep_research.news.flask_api.request",
|
|
) as mock_req:
|
|
mock_req.remote_addr = "127.0.0.1"
|
|
|
|
@scheduler_control_required
|
|
def dummy_view():
|
|
return jsonify({"status": "ok"}), 200
|
|
|
|
response, status_code = dummy_view()
|
|
data = response.get_json()
|
|
assert "LDR_NEWS_SCHEDULER" not in data["error"]
|
|
assert "administrator" in data["error"].lower()
|
|
|
|
|
|
class TestIsJobOwnedByUser:
|
|
"""Tests for _is_job_owned_by_user helper."""
|
|
|
|
def test_match_by_args(self):
|
|
"""Job with matching username in args is owned."""
|
|
from local_deep_research.news.flask_api import _is_job_owned_by_user
|
|
|
|
job = MagicMock()
|
|
job.args = ("alice", 1)
|
|
job.id = "alice_sub_1"
|
|
scheduler = MagicMock(spec=[]) # no user_sessions attr
|
|
assert _is_job_owned_by_user(job, "alice", scheduler) is True
|
|
|
|
def test_no_match_different_user(self):
|
|
"""Job with different username in args is not owned."""
|
|
from local_deep_research.news.flask_api import _is_job_owned_by_user
|
|
|
|
job = MagicMock()
|
|
job.args = ("bob", 1)
|
|
job.id = "bob_sub_1"
|
|
scheduler = MagicMock(spec=[])
|
|
assert _is_job_owned_by_user(job, "alice", scheduler) is False
|
|
|
|
def test_prefix_collision_prevented(self):
|
|
"""User 'alice' cannot see jobs for 'alice_admin'."""
|
|
from local_deep_research.news.flask_api import _is_job_owned_by_user
|
|
|
|
job = MagicMock()
|
|
job.args = ("alice_admin", 42)
|
|
job.id = "alice_admin_42"
|
|
scheduler = MagicMock(spec=[])
|
|
assert _is_job_owned_by_user(job, "alice", scheduler) is False
|
|
|
|
def test_match_by_scheduled_jobs_set(self):
|
|
"""Job in user's scheduled_jobs set is matched even without args."""
|
|
from local_deep_research.news.flask_api import _is_job_owned_by_user
|
|
|
|
job = MagicMock()
|
|
job.args = ()
|
|
job.id = "some_job_id"
|
|
scheduler = MagicMock()
|
|
scheduler.user_sessions = {
|
|
"alice": {"scheduled_jobs": {"some_job_id"}},
|
|
}
|
|
assert _is_job_owned_by_user(job, "alice", scheduler) is True
|
|
|
|
def test_overdue_job_matched_by_args(self):
|
|
"""Overdue job with username in args is matched."""
|
|
from local_deep_research.news.flask_api import _is_job_owned_by_user
|
|
|
|
job = MagicMock()
|
|
job.args = ("alice", 1)
|
|
job.id = "overdue_alice_1"
|
|
scheduler = MagicMock(spec=[])
|
|
assert _is_job_owned_by_user(job, "alice", scheduler) is True
|
|
|
|
def test_system_job_no_match(self):
|
|
"""System job with empty args is not owned by any user."""
|
|
from local_deep_research.news.flask_api import _is_job_owned_by_user
|
|
|
|
job = MagicMock()
|
|
job.args = ()
|
|
job.id = "cleanup_inactive_users"
|
|
scheduler = MagicMock()
|
|
scheduler.user_sessions = {
|
|
"alice": {"scheduled_jobs": set()},
|
|
}
|
|
assert _is_job_owned_by_user(job, "alice", scheduler) is False
|
|
|
|
|
|
class TestSchedulerStatusScoping:
|
|
"""Integration tests for /scheduler/status endpoint scoping."""
|
|
|
|
def _auth_session(self, client, username):
|
|
"""Inject a valid session so login_required passes."""
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = username
|
|
|
|
def _mock_scheduler(self):
|
|
"""Create a multi-user mock scheduler with jobs."""
|
|
scheduler = MagicMock()
|
|
scheduler.is_running = True
|
|
scheduler.config = {"check_interval": 300}
|
|
|
|
# Two users with sessions
|
|
scheduler.user_sessions = {
|
|
"alice": {"scheduled_jobs": {"alice_sub_1", "alice_sub_2"}},
|
|
"bob": {"scheduled_jobs": {"bob_sub_1"}},
|
|
}
|
|
|
|
# APScheduler jobs with explicit .args
|
|
alice_job1 = MagicMock()
|
|
alice_job1.id = "alice_sub_1"
|
|
alice_job1.name = "Alice Job 1"
|
|
alice_job1.args = ("alice", 1)
|
|
alice_job1.next_run_time = None
|
|
|
|
alice_job2 = MagicMock()
|
|
alice_job2.id = "alice_sub_2"
|
|
alice_job2.name = "Alice Job 2"
|
|
alice_job2.args = ("alice", 2)
|
|
alice_job2.next_run_time = None
|
|
|
|
bob_job = MagicMock()
|
|
bob_job.id = "bob_sub_1"
|
|
bob_job.name = "Bob Job 1"
|
|
bob_job.args = ("bob", 1)
|
|
bob_job.next_run_time = None
|
|
|
|
system_job = MagicMock()
|
|
system_job.id = "cleanup_inactive_users"
|
|
system_job.name = "Cleanup"
|
|
system_job.args = ()
|
|
system_job.next_run_time = None
|
|
|
|
scheduler.scheduler.get_jobs.return_value = [
|
|
alice_job1,
|
|
alice_job2,
|
|
bob_job,
|
|
system_job,
|
|
]
|
|
|
|
return scheduler
|
|
|
|
def test_status_scoped_to_current_user(self, client, app):
|
|
"""Alice sees only her own data when allow_api_control is False."""
|
|
self._auth_session(client, "alice")
|
|
mock_sched = self._mock_scheduler()
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"alice": True}
|
|
|
|
response = client.get("/news/api/scheduler/status")
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
|
|
# Alice sees only her own data
|
|
assert data["active_users"] == 1
|
|
assert data["total_scheduled_jobs"] == 2
|
|
assert data["scheduled_jobs"] == 2
|
|
# Only alice's jobs in apscheduler_jobs
|
|
job_ids = [j["id"] for j in data.get("apscheduler_jobs", [])]
|
|
assert "alice_sub_1" in job_ids
|
|
assert "alice_sub_2" in job_ids
|
|
assert "bob_sub_1" not in job_ids
|
|
assert "cleanup_inactive_users" not in job_ids
|
|
|
|
def test_status_shows_all_when_admin(self, client, app):
|
|
"""All data visible when allow_api_control is True."""
|
|
self._auth_session(client, "alice")
|
|
mock_sched = self._mock_scheduler()
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=True,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"alice": True}
|
|
|
|
response = client.get("/news/api/scheduler/status")
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
|
|
# Admin sees all users
|
|
assert data["active_users"] == 2
|
|
assert data["total_scheduled_jobs"] == 3
|
|
# All jobs visible
|
|
job_ids = [j["id"] for j in data.get("apscheduler_jobs", [])]
|
|
assert "bob_sub_1" in job_ids
|
|
assert "cleanup_inactive_users" in job_ids
|
|
|
|
def test_status_unknown_user_sees_zeros(self, client, app):
|
|
"""User not in sessions sees zero counts and empty jobs."""
|
|
self._auth_session(client, "charlie")
|
|
mock_sched = self._mock_scheduler()
|
|
|
|
with (
|
|
patch(
|
|
"local_deep_research.news.flask_api.get_env_setting",
|
|
return_value=False,
|
|
),
|
|
patch(
|
|
"local_deep_research.web.auth.decorators.db_manager",
|
|
) as mock_db,
|
|
patch(
|
|
"local_deep_research.scheduler.background.get_background_job_scheduler",
|
|
return_value=mock_sched,
|
|
),
|
|
):
|
|
mock_db.connections = {"charlie": True}
|
|
|
|
response = client.get("/news/api/scheduler/status")
|
|
assert response.status_code == 200
|
|
data = response.get_json()
|
|
|
|
assert data["active_users"] == 0
|
|
assert data["total_scheduled_jobs"] == 0
|
|
assert data["scheduled_jobs"] == 0
|
|
assert data.get("apscheduler_jobs", []) == []
|
|
|
|
|
|
class TestErrorHandling:
|
|
"""Tests for error handling in endpoints."""
|
|
|
|
def test_endpoints_handle_exceptions(self, client):
|
|
"""Test that endpoints exist and handle requests."""
|
|
# Test that routes are registered
|
|
endpoints = [
|
|
("/news/api/feed", "GET"),
|
|
("/news/api/subscriptions", "GET"),
|
|
("/news/api/subscription/folders", "GET"),
|
|
]
|
|
|
|
for endpoint, method in endpoints:
|
|
try:
|
|
if method == "GET":
|
|
response = client.get(endpoint)
|
|
else:
|
|
response = client.post(endpoint)
|
|
|
|
# Any response is acceptable
|
|
assert response.status_code == 401, response.status_code
|
|
except Exception:
|
|
# If dependencies fail, that's okay - routes may exist
|
|
pass
|