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
791 lines
32 KiB
Python
791 lines
32 KiB
Python
"""
|
|
Tests for news/api.py
|
|
|
|
Tests cover:
|
|
- get_news_feed() function
|
|
- submit_feedback() function
|
|
- get_votes_for_cards() function
|
|
- _format_time_ago() function
|
|
- get_subscription() function
|
|
- get_subscriptions() function
|
|
- create_subscription() function
|
|
- delete_subscription() function
|
|
- update_subscription() function
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from datetime import datetime, timezone, timedelta
|
|
import json
|
|
|
|
|
|
class TestGetNewsFeed:
|
|
"""Tests for get_news_feed function."""
|
|
|
|
def test_get_news_feed_invalid_limit_raises_exception(self):
|
|
"""Test that invalid limit raises InvalidLimitException."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
from local_deep_research.news.exceptions import InvalidLimitException
|
|
|
|
with pytest.raises(InvalidLimitException):
|
|
get_news_feed(user_id="test", limit=0)
|
|
|
|
def test_get_news_feed_negative_limit_raises_exception(self):
|
|
"""Test that negative limit raises InvalidLimitException."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
from local_deep_research.news.exceptions import InvalidLimitException
|
|
|
|
with pytest.raises(InvalidLimitException):
|
|
get_news_feed(user_id="test", limit=-1)
|
|
|
|
def test_get_news_feed_returns_empty_when_no_items(self):
|
|
"""Test that empty result is returned when no news items."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = []
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_news_feed(user_id="test", limit=10)
|
|
|
|
assert "news_items" in result
|
|
assert len(result["news_items"]) == 0
|
|
|
|
def test_get_news_feed_filters_by_subscription_id(self):
|
|
"""Test that subscription_id filter is applied."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_query = mock_db.query.return_value.filter.return_value
|
|
mock_query.filter.return_value.order_by.return_value.limit.return_value.all.return_value = []
|
|
mock_session.return_value = mock_db
|
|
|
|
get_news_feed(user_id="test", limit=10, subscription_id="sub-123")
|
|
|
|
# Verify filter was applied (query was built)
|
|
assert mock_db.query.called
|
|
|
|
def test_get_news_feed_handles_json_parsing_errors(self):
|
|
"""Test handling of malformed JSON in research_meta."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
# Create mock result with invalid JSON
|
|
mock_result = MagicMock()
|
|
mock_result.id = "test-id"
|
|
mock_result.query = "news about test"
|
|
mock_result.title = None
|
|
mock_result.created_at = datetime.now(timezone.utc)
|
|
mock_result.completed_at = None
|
|
mock_result.duration_seconds = 10
|
|
mock_result.report_path = None
|
|
mock_result.report_content = None
|
|
mock_result.research_meta = "invalid json{"
|
|
mock_result.status = "completed"
|
|
|
|
mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
|
|
mock_result
|
|
]
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_news_feed(user_id="test", limit=10)
|
|
# Should handle error gracefully
|
|
assert "news_items" in result
|
|
|
|
def test_get_news_feed_extracts_links_from_content(self):
|
|
"""Test that links are extracted from report content."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
# Create mock result with links in content
|
|
mock_result = MagicMock()
|
|
mock_result.id = "test-id"
|
|
mock_result.query = "breaking news today"
|
|
mock_result.title = "Test News"
|
|
mock_result.created_at = datetime.now(timezone.utc)
|
|
mock_result.completed_at = datetime.now(timezone.utc)
|
|
mock_result.duration_seconds = 10
|
|
mock_result.report_path = None
|
|
mock_result.report_content = (
|
|
"Source Title\nURL: https://example.com/article\nMore content"
|
|
)
|
|
mock_result.research_meta = json.dumps(
|
|
{"is_news_search": True, "generated_headline": "Test Headline"}
|
|
)
|
|
mock_result.status = "completed"
|
|
|
|
mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
|
|
mock_result
|
|
]
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_news_feed(user_id="test", limit=10)
|
|
assert "news_items" in result
|
|
|
|
def test_get_news_feed_includes_generated_at_timestamp(self):
|
|
"""Test that generated_at timestamp is included."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = []
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_news_feed(user_id="test", limit=10)
|
|
|
|
assert "generated_at" in result
|
|
|
|
def test_get_news_feed_respects_focus_parameter(self):
|
|
"""Test that focus parameter is passed through."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = []
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_news_feed(user_id="test", limit=10, focus="technology")
|
|
|
|
assert result.get("focus") == "technology"
|
|
|
|
def test_get_news_feed_handles_database_error(self):
|
|
"""Test handling of database access errors."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
from local_deep_research.news.exceptions import DatabaseAccessException
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_session.side_effect = Exception("Database connection failed")
|
|
|
|
with pytest.raises((DatabaseAccessException, Exception)):
|
|
get_news_feed(user_id="test", limit=10)
|
|
|
|
def test_get_news_feed_skips_items_without_headline(self):
|
|
"""Test that items without headlines are skipped."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
# Create mock result without headline
|
|
mock_result = MagicMock()
|
|
mock_result.id = "test-id"
|
|
mock_result.query = "breaking news test"
|
|
mock_result.title = None # No title
|
|
mock_result.created_at = datetime.now(timezone.utc)
|
|
mock_result.completed_at = datetime.now(timezone.utc)
|
|
mock_result.duration_seconds = 10
|
|
mock_result.report_path = None
|
|
mock_result.report_content = "Some content"
|
|
mock_result.research_meta = json.dumps(
|
|
{
|
|
"is_news_search": True,
|
|
# No generated_headline
|
|
}
|
|
)
|
|
mock_result.status = "completed"
|
|
|
|
mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = [
|
|
mock_result
|
|
]
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_news_feed(user_id="test", limit=10)
|
|
assert "news_items" in result
|
|
|
|
def test_get_news_feed_skips_in_progress_items(self):
|
|
"""Test that in_progress items are skipped."""
|
|
from local_deep_research.news.api import get_news_feed
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter.return_value.order_by.return_value.limit.return_value.all.return_value = []
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_news_feed(user_id="test", limit=10)
|
|
# Verify query filters by completed status
|
|
assert "news_items" in result
|
|
|
|
|
|
class TestSubmitFeedback:
|
|
"""Tests for submit_feedback function."""
|
|
|
|
def test_submit_feedback_invalid_vote_raises_error(self):
|
|
"""Test that invalid vote type raises ValueError."""
|
|
from local_deep_research.news.api import submit_feedback
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with pytest.raises(ValueError) as exc_info:
|
|
submit_feedback("card-1", "testuser", "invalid")
|
|
|
|
assert "Invalid vote type" in str(exc_info.value)
|
|
|
|
def test_submit_feedback_creates_new_rating(self):
|
|
"""Test that new rating is created when none exists."""
|
|
from local_deep_research.news.api import submit_feedback
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
|
mock_db.query.return_value.filter_by.return_value.count.return_value = 0
|
|
mock_session.return_value = mock_db
|
|
|
|
result = submit_feedback("card-1", "testuser", "up")
|
|
|
|
assert result["success"] is True
|
|
assert mock_db.add.called
|
|
|
|
def test_submit_feedback_updates_existing_rating(self):
|
|
"""Test that existing rating is updated."""
|
|
from local_deep_research.news.api import submit_feedback
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
existing_rating = MagicMock()
|
|
existing_rating.rating_value = "down"
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = existing_rating
|
|
mock_db.query.return_value.filter_by.return_value.count.return_value = 1
|
|
mock_session.return_value = mock_db
|
|
|
|
result = submit_feedback("card-1", "testuser", "up")
|
|
|
|
assert existing_rating.rating_value == "up"
|
|
assert result["success"] is True
|
|
|
|
def test_submit_feedback_returns_vote_counts(self):
|
|
"""Test that vote counts are returned."""
|
|
from local_deep_research.news.api import submit_feedback
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
|
mock_db.query.return_value.filter_by.return_value.count.return_value = 5
|
|
mock_session.return_value = mock_db
|
|
|
|
result = submit_feedback("card-1", "testuser", "up")
|
|
|
|
assert "upvotes" in result
|
|
assert "downvotes" in result
|
|
|
|
def test_submit_feedback_no_username_raises_error(self):
|
|
"""Test that missing username raises ValueError."""
|
|
from local_deep_research.news.api import submit_feedback
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with pytest.raises(ValueError):
|
|
submit_feedback("card-1", None, "up")
|
|
|
|
|
|
class TestGetVotesForCards:
|
|
"""Tests for get_votes_for_cards function."""
|
|
|
|
def test_get_votes_for_cards_returns_votes(self):
|
|
"""Test that votes are returned for each card."""
|
|
from local_deep_research.news.api import get_votes_for_cards
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
|
mock_db.query.return_value.filter_by.return_value.count.return_value = 0
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_votes_for_cards(
|
|
["card-1", "card-2"], user_id="testuser"
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert "votes" in result
|
|
assert "card-1" in result["votes"]
|
|
assert "card-2" in result["votes"]
|
|
|
|
def test_get_votes_for_cards_includes_user_vote(self):
|
|
"""Test that user's vote is included."""
|
|
from local_deep_research.news.api import get_votes_for_cards
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
user_vote = MagicMock()
|
|
user_vote.rating_value = "up"
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = user_vote
|
|
mock_db.query.return_value.filter_by.return_value.count.return_value = 1
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_votes_for_cards(["card-1"], user_id="testuser")
|
|
|
|
assert result["votes"]["card-1"]["user_vote"] == "up"
|
|
|
|
def test_get_votes_for_cards_no_username_raises_error(self):
|
|
"""Test that missing username raises error."""
|
|
from local_deep_research.news.api import get_votes_for_cards
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with pytest.raises(ValueError):
|
|
get_votes_for_cards(["card-1"], user_id=None)
|
|
|
|
def test_get_votes_for_cards_empty_list(self):
|
|
"""Test with empty card list."""
|
|
from local_deep_research.news.api import get_votes_for_cards
|
|
|
|
with patch("flask.has_request_context") as mock_ctx:
|
|
mock_ctx.return_value = False
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_votes_for_cards([], user_id="testuser")
|
|
|
|
assert result["success"] is True
|
|
assert result["votes"] == {}
|
|
|
|
|
|
class TestFormatTimeAgo:
|
|
"""Tests for _format_time_ago function."""
|
|
|
|
def test_format_time_ago_just_now(self):
|
|
"""Test that recent timestamps return 'Just now'."""
|
|
from local_deep_research.news.api import _format_time_ago
|
|
|
|
now = datetime.now(timezone.utc)
|
|
result = _format_time_ago(now.isoformat())
|
|
|
|
assert result == "Just now"
|
|
|
|
def test_format_time_ago_minutes(self):
|
|
"""Test formatting for minutes ago."""
|
|
from local_deep_research.news.api import _format_time_ago
|
|
|
|
past = datetime.now(timezone.utc) - timedelta(minutes=30)
|
|
result = _format_time_ago(past.isoformat())
|
|
|
|
assert "minute" in result
|
|
|
|
def test_format_time_ago_hours(self):
|
|
"""Test formatting for hours ago."""
|
|
from local_deep_research.news.api import _format_time_ago
|
|
|
|
past = datetime.now(timezone.utc) - timedelta(hours=5)
|
|
result = _format_time_ago(past.isoformat())
|
|
|
|
assert "hour" in result
|
|
|
|
def test_format_time_ago_days(self):
|
|
"""Test formatting for days ago."""
|
|
from local_deep_research.news.api import _format_time_ago
|
|
|
|
past = datetime.now(timezone.utc) - timedelta(days=3)
|
|
result = _format_time_ago(past.isoformat())
|
|
|
|
assert "day" in result
|
|
|
|
def test_format_time_ago_invalid_timestamp(self):
|
|
"""Invalid timestamps raise (caller logs + skips the row)."""
|
|
from local_deep_research.news.api import _format_time_ago
|
|
|
|
with pytest.raises(ValueError):
|
|
_format_time_ago("not a valid timestamp")
|
|
|
|
def test_format_time_ago_timezone_handling(self):
|
|
"""Naive timestamps are interpreted as UTC by the formatter.
|
|
|
|
Build the naive string from UTC explicitly — using datetime.now()
|
|
(local) made this test fail on any non-UTC machine, since the
|
|
formatter then computed diff = (utc_now - local_now_minus_2h),
|
|
which collapses to ~0 on a UTC+2 box and returns "Just now".
|
|
"""
|
|
from local_deep_research.news.api import _format_time_ago
|
|
|
|
naive_utc = (
|
|
datetime.now(timezone.utc) - timedelta(hours=2, seconds=1)
|
|
).replace(tzinfo=None)
|
|
result = _format_time_ago(naive_utc.isoformat())
|
|
|
|
# The function may return "2 hours ago" or "120 minutes ago"
|
|
assert "hour" in result or "minute" in result
|
|
|
|
|
|
class TestGetSubscription:
|
|
"""Tests for get_subscription function."""
|
|
|
|
def test_get_subscription_not_found_raises_exception(self):
|
|
"""Test that missing subscription raises SubscriptionNotFoundException."""
|
|
from local_deep_research.news.api import get_subscription
|
|
from local_deep_research.news.exceptions import (
|
|
SubscriptionNotFoundException,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
|
mock_session.return_value = mock_db
|
|
|
|
with pytest.raises(SubscriptionNotFoundException):
|
|
get_subscription("nonexistent-id")
|
|
|
|
def test_get_subscription_returns_formatted_data(self):
|
|
"""Test that subscription data is properly formatted."""
|
|
from local_deep_research.news.api import get_subscription
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
mock_sub = MagicMock()
|
|
mock_sub.id = "sub-123"
|
|
mock_sub.name = "Test Sub"
|
|
mock_sub.query_or_topic = "test query"
|
|
mock_sub.subscription_type = "search"
|
|
mock_sub.refresh_interval_minutes = 240
|
|
mock_sub.status = "active"
|
|
mock_sub.folder_id = None
|
|
mock_sub.model_provider = "openai"
|
|
mock_sub.model = "gpt-4"
|
|
mock_sub.search_strategy = "default"
|
|
mock_sub.custom_endpoint = None
|
|
mock_sub.search_engine = "searxng"
|
|
mock_sub.search_iterations = 3
|
|
mock_sub.questions_per_iteration = 5
|
|
mock_sub.created_at = datetime.now(timezone.utc)
|
|
mock_sub.updated_at = datetime.now(timezone.utc)
|
|
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = mock_sub
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_subscription("sub-123")
|
|
|
|
assert result["id"] == "sub-123"
|
|
assert result["name"] == "Test Sub"
|
|
assert result["is_active"] is True
|
|
|
|
|
|
class TestCreateSubscription:
|
|
"""Tests for create_subscription function."""
|
|
|
|
def test_create_subscription_success(self):
|
|
"""Test successful subscription creation."""
|
|
from local_deep_research.news.api import create_subscription
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_session.return_value = mock_db
|
|
|
|
with patch(
|
|
"local_deep_research.news.api._notify_scheduler_about_subscription_change"
|
|
):
|
|
result = create_subscription(
|
|
user_id="testuser",
|
|
query="test query",
|
|
subscription_type="search",
|
|
refresh_minutes=120,
|
|
)
|
|
|
|
assert result["status"] == "success"
|
|
assert "subscription_id" in result
|
|
assert mock_db.add.called
|
|
assert mock_db.commit.called
|
|
|
|
def test_create_subscription_uses_default_refresh_minutes(self):
|
|
"""Test that default refresh minutes is used when not provided."""
|
|
from local_deep_research.news.api import create_subscription
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_session.return_value = mock_db
|
|
|
|
with patch(
|
|
"local_deep_research.news.api._notify_scheduler_about_subscription_change"
|
|
):
|
|
with patch(
|
|
"local_deep_research.utilities.db_utils.get_settings_manager"
|
|
) as mock_settings:
|
|
mock_mgr = MagicMock()
|
|
mock_mgr.get_setting.return_value = 240
|
|
mock_settings.return_value = mock_mgr
|
|
|
|
result = create_subscription(
|
|
user_id="testuser",
|
|
query="test query",
|
|
)
|
|
|
|
assert result["status"] == "success"
|
|
|
|
|
|
class TestDeleteSubscription:
|
|
"""Tests for delete_subscription function."""
|
|
|
|
def test_delete_subscription_not_found_raises_exception(self):
|
|
"""Test that missing subscription raises exception."""
|
|
from local_deep_research.news.api import delete_subscription
|
|
from local_deep_research.news.exceptions import (
|
|
SubscriptionNotFoundException,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
|
mock_session.return_value = mock_db
|
|
|
|
with pytest.raises(SubscriptionNotFoundException):
|
|
delete_subscription("nonexistent-id")
|
|
|
|
def test_delete_subscription_success(self):
|
|
"""Test successful subscription deletion."""
|
|
from local_deep_research.news.api import delete_subscription
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
mock_sub = MagicMock()
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = mock_sub
|
|
mock_session.return_value = mock_db
|
|
|
|
with patch(
|
|
"local_deep_research.news.api._notify_scheduler_about_subscription_change"
|
|
):
|
|
result = delete_subscription("sub-123")
|
|
|
|
assert result["status"] == "success"
|
|
assert mock_db.delete.called
|
|
assert mock_db.commit.called
|
|
|
|
|
|
class TestUpdateSubscription:
|
|
"""Tests for update_subscription function."""
|
|
|
|
def test_update_subscription_not_found_raises_exception(self):
|
|
"""Test that missing subscription raises exception."""
|
|
from local_deep_research.news.api import update_subscription
|
|
from local_deep_research.news.exceptions import (
|
|
SubscriptionNotFoundException,
|
|
)
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = None
|
|
mock_session.return_value = mock_db
|
|
|
|
with pytest.raises(SubscriptionNotFoundException):
|
|
update_subscription("nonexistent-id", {"name": "New Name"})
|
|
|
|
def test_update_subscription_updates_fields(self):
|
|
"""Test that subscription fields are updated."""
|
|
from local_deep_research.news.api import update_subscription
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
mock_sub = MagicMock()
|
|
mock_sub.id = "sub-123"
|
|
mock_sub.name = "Old Name"
|
|
mock_sub.query_or_topic = "old query"
|
|
mock_sub.subscription_type = "search"
|
|
mock_sub.refresh_interval_minutes = 240
|
|
mock_sub.status = "active"
|
|
mock_sub.folder_id = None
|
|
mock_sub.model_provider = None
|
|
mock_sub.model = None
|
|
mock_sub.search_strategy = None
|
|
mock_sub.custom_endpoint = None
|
|
mock_sub.search_engine = None
|
|
mock_sub.search_iterations = 3
|
|
mock_sub.questions_per_iteration = 5
|
|
|
|
mock_db.query.return_value.filter_by.return_value.first.return_value = mock_sub
|
|
mock_session.return_value = mock_db
|
|
|
|
with patch(
|
|
"local_deep_research.news.api._notify_scheduler_about_subscription_change"
|
|
):
|
|
result = update_subscription(
|
|
"sub-123", {"name": "New Name", "is_active": False}
|
|
)
|
|
|
|
assert mock_sub.name == "New Name"
|
|
assert mock_sub.status == "paused"
|
|
assert result["status"] == "success"
|
|
|
|
|
|
class TestGetSubscriptions:
|
|
"""Tests for get_subscriptions function."""
|
|
|
|
def test_get_subscriptions_returns_list(self):
|
|
"""Test that subscription list is returned."""
|
|
from local_deep_research.news.api import get_subscriptions
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
mock_db.query.return_value.all.return_value = []
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_subscriptions(user_id="testuser")
|
|
|
|
assert "subscriptions" in result
|
|
assert "total" in result
|
|
|
|
def test_get_subscriptions_counts_runs(self):
|
|
"""Test that total runs are counted."""
|
|
from local_deep_research.news.api import get_subscriptions
|
|
|
|
with patch(
|
|
"local_deep_research.database.session_context.get_user_db_session"
|
|
) as mock_session:
|
|
mock_db = MagicMock()
|
|
mock_db.__enter__ = Mock(return_value=mock_db)
|
|
mock_db.__exit__ = Mock(return_value=False)
|
|
|
|
mock_sub = MagicMock()
|
|
mock_sub.id = "sub-123"
|
|
mock_sub.query_or_topic = "test"
|
|
mock_sub.subscription_type = "search"
|
|
mock_sub.refresh_interval_minutes = 240
|
|
mock_sub.status = "active"
|
|
mock_sub.created_at = datetime.now(timezone.utc)
|
|
mock_sub.next_refresh = datetime.now(timezone.utc)
|
|
mock_sub.last_refresh = None
|
|
mock_sub.name = "Test"
|
|
mock_sub.folder_id = None
|
|
mock_sub.model_provider = "openai"
|
|
mock_sub.model = "gpt-4o"
|
|
mock_sub.search_strategy = "focused-iteration"
|
|
mock_sub.search_engine = "searxng"
|
|
mock_sub.custom_endpoint = None
|
|
mock_sub.search_iterations = 3
|
|
mock_sub.questions_per_iteration = 5
|
|
|
|
mock_db.query.return_value.all.return_value = [mock_sub]
|
|
mock_db.query.return_value.filter.return_value.scalar.return_value = 5
|
|
mock_session.return_value = mock_db
|
|
|
|
result = get_subscriptions(user_id="testuser")
|
|
|
|
assert len(result["subscriptions"]) == 1
|
|
# The subscription's saved model/search config must be included
|
|
# (it was previously dropped, so callers silently got None).
|
|
sub = result["subscriptions"][0]
|
|
assert sub["model_provider"] == "openai"
|
|
assert sub["model"] == "gpt-4o"
|
|
assert sub["search_strategy"] == "focused-iteration"
|
|
assert sub["search_engine"] == "searxng"
|
|
assert sub["status"] == "active"
|