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
895 lines
30 KiB
Python
895 lines
30 KiB
Python
"""
|
|
Comprehensive coverage tests for news/flask_api.py
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from flask import Flask
|
|
|
|
MODULE = "local_deep_research.news.flask_api"
|
|
AUTH_MOD = "local_deep_research.web.auth.decorators"
|
|
|
|
|
|
@pytest.fixture()
|
|
def app():
|
|
from local_deep_research.news.flask_api import news_api_bp
|
|
|
|
app = Flask(__name__)
|
|
app.config["SECRET_KEY"] = "test-secret"
|
|
app.config["TESTING"] = True
|
|
app.register_blueprint(news_api_bp, url_prefix="/news/api")
|
|
return app
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(app):
|
|
return app.test_client()
|
|
|
|
|
|
def _auth(client, username="testuser"):
|
|
with client.session_transaction() as sess:
|
|
sess["username"] = username
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _bypass_auth():
|
|
mock_db = MagicMock()
|
|
mock_db.is_user_connected.return_value = True
|
|
with patch(f"{AUTH_MOD}.db_manager", mock_db):
|
|
yield
|
|
|
|
|
|
class TestSubscribe:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.create_subscription", return_value={"id": "s1"}
|
|
) as m,
|
|
):
|
|
resp = client.post("/news/api/subscribe", json={"query": "AI news"})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["id"] == "s1"
|
|
assert m.call_args.kwargs["query"] == "AI news"
|
|
|
|
def test_folder_name_and_active_forwarded(self, client):
|
|
"""folder_id, name and is_active from the request body must reach
|
|
api.create_subscription (regression test for #4489, where the
|
|
frontend dropped these fields)."""
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.create_subscription", return_value={"id": "s1"}
|
|
) as m,
|
|
):
|
|
resp = client.post(
|
|
"/news/api/subscribe",
|
|
json={
|
|
"query": "AI news",
|
|
"name": "My subscription",
|
|
"folder_id": "7f3d2a10-0000-0000-0000-000000000000",
|
|
"is_active": False,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
kwargs = m.call_args.kwargs
|
|
assert kwargs["folder_id"] == "7f3d2a10-0000-0000-0000-000000000000"
|
|
assert kwargs["name"] == "My subscription"
|
|
assert kwargs["is_active"] is False
|
|
|
|
def test_missing_query_returns_400(self, client):
|
|
_auth(client)
|
|
with patch(f"{MODULE}.get_user_id", return_value="testuser"):
|
|
resp = client.post(
|
|
"/news/api/subscribe", json={"subscription_type": "search"}
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "required" in resp.get_json()["error"]
|
|
|
|
def test_invalid_json_returns_400(self, client):
|
|
_auth(client)
|
|
resp = client.post(
|
|
"/news/api/subscribe",
|
|
data="not json",
|
|
content_type="application/json",
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_value_error_returns_400(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.create_subscription",
|
|
side_effect=ValueError("bad"),
|
|
),
|
|
):
|
|
resp = client.post("/news/api/subscribe", json={"query": "test"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_generic_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.create_subscription",
|
|
side_effect=RuntimeError("boom"),
|
|
),
|
|
):
|
|
resp = client.post("/news/api/subscribe", json={"query": "test"})
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestVoteOnNews:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback", return_value={"status": "ok"}
|
|
),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/vote", json={"card_id": "c1", "vote": "up"}
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_missing_card_id_returns_400(self, client):
|
|
_auth(client)
|
|
with patch(f"{MODULE}.get_user_id", return_value="testuser"):
|
|
resp = client.post("/news/api/vote", json={"vote": "up"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_missing_vote_returns_400(self, client):
|
|
_auth(client)
|
|
with patch(f"{MODULE}.get_user_id", return_value="testuser"):
|
|
resp = client.post("/news/api/vote", json={"card_id": "c1"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_not_found_returns_404(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback",
|
|
side_effect=ValueError("Card not found"),
|
|
),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/vote", json={"card_id": "c1", "vote": "up"}
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_other_value_error_returns_400(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback",
|
|
side_effect=ValueError("invalid direction"),
|
|
),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/vote", json={"card_id": "c1", "vote": "bad"}
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_generic_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback",
|
|
side_effect=RuntimeError("boom"),
|
|
),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/vote", json={"card_id": "c1", "vote": "up"}
|
|
)
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestGetBatchFeedback:
|
|
def test_returns_votes(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.get_votes_for_cards",
|
|
return_value={"votes": {"c1": "up"}},
|
|
),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/feedback/batch", json={"card_ids": ["c1"]}
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["votes"]["c1"] == "up"
|
|
|
|
def test_empty_card_ids_returns_empty_votes(self, client):
|
|
_auth(client)
|
|
resp = client.post("/news/api/feedback/batch", json={"card_ids": []})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["votes"] == {}
|
|
|
|
def test_not_found_returns_404(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.get_votes_for_cards",
|
|
side_effect=ValueError("not found"),
|
|
),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/feedback/batch", json={"card_ids": ["c1"]}
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_generic_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.get_votes_for_cards",
|
|
side_effect=RuntimeError("db"),
|
|
),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/feedback/batch", json={"card_ids": ["c1"]}
|
|
)
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestSubmitFeedback:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback",
|
|
return_value={"status": "recorded"},
|
|
),
|
|
):
|
|
resp = client.post("/news/api/feedback/card1", json={"vote": "up"})
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "recorded"
|
|
|
|
def test_missing_vote_returns_400(self, client):
|
|
_auth(client)
|
|
with patch(f"{MODULE}.get_user_id", return_value="testuser"):
|
|
resp = client.post("/news/api/feedback/card1", json={})
|
|
assert resp.status_code == 400
|
|
assert "vote" in resp.get_json()["error"].lower()
|
|
|
|
def test_not_found_returns_404(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback",
|
|
side_effect=ValueError("Card not found"),
|
|
),
|
|
):
|
|
resp = client.post("/news/api/feedback/card1", json={"vote": "up"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_must_be_returns_400(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback",
|
|
side_effect=ValueError("vote must be up or down"),
|
|
),
|
|
):
|
|
resp = client.post("/news/api/feedback/card1", json={"vote": "x"})
|
|
assert resp.status_code == 400
|
|
assert resp.get_json()["error"] == "Invalid input value"
|
|
|
|
def test_other_value_error_returns_400(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback",
|
|
side_effect=ValueError("something else"),
|
|
),
|
|
):
|
|
resp = client.post("/news/api/feedback/card1", json={"vote": "up"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_generic_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.submit_feedback",
|
|
side_effect=RuntimeError("crash"),
|
|
),
|
|
):
|
|
resp = client.post("/news/api/feedback/card1", json={"vote": "up"})
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestResearchNewsItem:
|
|
def test_with_depth(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.research_news_item", return_value={"summary": "ok"}
|
|
) as m:
|
|
resp = client.post(
|
|
"/news/api/research/card42", json={"depth": "detailed"}
|
|
)
|
|
assert resp.status_code == 200
|
|
m.assert_called_once_with("card42", "detailed")
|
|
|
|
def test_default_depth(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.research_news_item", return_value={"summary": "ok"}
|
|
) as m:
|
|
resp = client.post("/news/api/research/card42", json={})
|
|
assert resp.status_code == 200
|
|
m.assert_called_once_with("card42", "quick")
|
|
|
|
def test_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.research_news_item", side_effect=RuntimeError("fail")
|
|
):
|
|
resp = client.post(
|
|
"/news/api/research/card42", json={"depth": "quick"}
|
|
)
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestGetNewsFeed:
|
|
def _ms(self):
|
|
m = MagicMock()
|
|
m.get_setting.return_value = 20
|
|
return m
|
|
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(f"{MODULE}.get_settings_manager", return_value=self._ms()),
|
|
patch(
|
|
f"{MODULE}.api.get_news_feed",
|
|
return_value={"news_items": [{"id": "1"}]},
|
|
),
|
|
):
|
|
resp = client.get("/news/api/feed")
|
|
assert resp.status_code == 200
|
|
assert len(resp.get_json()["news_items"]) == 1
|
|
|
|
def test_error_in_result_returns_500(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(f"{MODULE}.get_settings_manager", return_value=self._ms()),
|
|
patch(
|
|
f"{MODULE}.api.get_news_feed",
|
|
return_value={"error": "db error", "news_items": []},
|
|
),
|
|
):
|
|
resp = client.get("/news/api/feed")
|
|
assert resp.status_code == 500
|
|
|
|
def test_must_be_between_returns_400(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(f"{MODULE}.get_settings_manager", return_value=self._ms()),
|
|
patch(
|
|
f"{MODULE}.api.get_news_feed",
|
|
return_value={
|
|
"error": "limit must be between 1 and 100",
|
|
"news_items": [],
|
|
},
|
|
),
|
|
):
|
|
resp = client.get("/news/api/feed")
|
|
assert resp.status_code == 400
|
|
|
|
def test_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with patch(f"{MODULE}.get_user_id", side_effect=RuntimeError("boom")):
|
|
resp = client.get("/news/api/feed")
|
|
assert resp.status_code == 500
|
|
|
|
|
|
def _mock_db_session():
|
|
ms = MagicMock()
|
|
ctx = MagicMock()
|
|
ctx.__enter__ = MagicMock(return_value=ms)
|
|
ctx.__exit__ = MagicMock(return_value=False)
|
|
return ctx, ms
|
|
|
|
|
|
class TestCreateFolder:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
ctx, ms = _mock_db_session()
|
|
ms.query.return_value.filter_by.return_value.first.return_value = None
|
|
mf = MagicMock()
|
|
mf.to_dict.return_value = {"id": "f1", "name": "Tech"}
|
|
mm = MagicMock()
|
|
mm.create_folder.return_value = mf
|
|
with (
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/subscription/folders", json={"name": "Tech"}
|
|
)
|
|
assert resp.status_code == 201
|
|
|
|
def test_missing_name_returns_400(self, client):
|
|
_auth(client)
|
|
resp = client.post("/news/api/subscription/folders", json={})
|
|
assert resp.status_code == 400
|
|
|
|
def test_duplicate_returns_409(self, client):
|
|
_auth(client)
|
|
ctx, ms = _mock_db_session()
|
|
ms.query.return_value.filter_by.return_value.first.return_value = (
|
|
MagicMock()
|
|
)
|
|
with (
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager"),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/subscription/folders", json={"name": "Dup"}
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
class TestUpdateFolder:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
ctx, ms = _mock_db_session()
|
|
mf = MagicMock()
|
|
mf.to_dict.return_value = {"id": "f1", "name": "Updated"}
|
|
mm = MagicMock()
|
|
mm.update_folder.return_value = mf
|
|
with (
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.put(
|
|
"/news/api/subscription/folders/f1", json={"name": "Updated"}
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_not_found_returns_404(self, client):
|
|
_auth(client)
|
|
ctx, ms = _mock_db_session()
|
|
mm = MagicMock()
|
|
mm.update_folder.return_value = None
|
|
with (
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.put(
|
|
"/news/api/subscription/folders/f1", json={"name": "X"}
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestDeleteFolder:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
ctx, _ = _mock_db_session()
|
|
mm = MagicMock()
|
|
mm.delete_folder.return_value = True
|
|
with (
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.delete("/news/api/subscription/folders/f1")
|
|
assert resp.status_code == 200
|
|
|
|
def test_not_found_returns_404(self, client):
|
|
_auth(client)
|
|
ctx, _ = _mock_db_session()
|
|
mm = MagicMock()
|
|
mm.delete_folder.return_value = False
|
|
with (
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.delete("/news/api/subscription/folders/f1")
|
|
assert resp.status_code == 404
|
|
|
|
def test_with_move_to(self, client):
|
|
_auth(client)
|
|
ctx, _ = _mock_db_session()
|
|
mm = MagicMock()
|
|
mm.delete_folder.return_value = True
|
|
with (
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.delete("/news/api/subscription/folders/f1?move_to=f2")
|
|
assert resp.status_code == 200
|
|
mm.delete_folder.assert_called_once_with("f1", "f2")
|
|
|
|
|
|
class TestGetFolders:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
ctx, _ = _mock_db_session()
|
|
fo = MagicMock()
|
|
fo.to_dict.return_value = {"id": "f1", "name": "Tech"}
|
|
mm = MagicMock()
|
|
mm.get_user_folders.return_value = [fo]
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.get("/news/api/subscription/folders")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()[0]["name"] == "Tech"
|
|
|
|
|
|
class TestDeleteSubscription:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with patch(f"{MODULE}.api.delete_subscription", return_value=True):
|
|
resp = client.delete("/news/api/subscriptions/s1")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["status"] == "success"
|
|
|
|
def test_not_found(self, client):
|
|
_auth(client)
|
|
with patch(f"{MODULE}.api.delete_subscription", return_value=False):
|
|
resp = client.delete("/news/api/subscriptions/s1")
|
|
assert resp.status_code == 404
|
|
|
|
def test_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.delete_subscription",
|
|
side_effect=RuntimeError("boom"),
|
|
):
|
|
resp = client.delete("/news/api/subscriptions/s1")
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestGetSubscription:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.get_subscription",
|
|
return_value={"id": "s1", "query": "AI"},
|
|
):
|
|
resp = client.get("/news/api/subscriptions/s1")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["id"] == "s1"
|
|
|
|
def test_null_id_returns_400(self, client):
|
|
_auth(client)
|
|
resp = client.get("/news/api/subscriptions/null")
|
|
assert resp.status_code == 400
|
|
|
|
def test_undefined_id_returns_400(self, client):
|
|
_auth(client)
|
|
resp = client.get("/news/api/subscriptions/undefined")
|
|
assert resp.status_code == 400
|
|
|
|
def test_not_found(self, client):
|
|
_auth(client)
|
|
with patch(f"{MODULE}.api.get_subscription", return_value=None):
|
|
resp = client.get("/news/api/subscriptions/s1")
|
|
assert resp.status_code == 404
|
|
|
|
def test_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.get_subscription", side_effect=RuntimeError("db")
|
|
):
|
|
resp = client.get("/news/api/subscriptions/s1")
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestUpdateSubscription:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.update_subscription",
|
|
return_value={"id": "s1", "name": "New"},
|
|
):
|
|
resp = client.put(
|
|
"/news/api/subscriptions/s1", json={"name": "New"}
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_not_found_error(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.update_subscription",
|
|
return_value={"error": "Subscription not found"},
|
|
):
|
|
resp = client.put("/news/api/subscriptions/s1", json={"name": "X"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_other_error(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.update_subscription",
|
|
return_value={"error": "invalid field"},
|
|
):
|
|
resp = client.put("/news/api/subscriptions/s1", json={"name": "X"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_exception_returns_500(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.update_subscription",
|
|
side_effect=RuntimeError("crash"),
|
|
):
|
|
resp = client.put("/news/api/subscriptions/s1", json={"name": "X"})
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestGetCurrentUserSubscriptions:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.get_subscriptions",
|
|
return_value={"subscriptions": []},
|
|
),
|
|
):
|
|
resp = client.get("/news/api/subscriptions/current")
|
|
assert resp.status_code == 200
|
|
|
|
def test_error_in_result(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.get_subscriptions",
|
|
return_value={"error": "db issue"},
|
|
),
|
|
):
|
|
resp = client.get("/news/api/subscriptions/current")
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestSubscriptionHistory:
|
|
def _ms(self):
|
|
m = MagicMock()
|
|
m.get_setting.return_value = 20
|
|
return m
|
|
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_settings_manager", return_value=self._ms()),
|
|
patch(
|
|
f"{MODULE}.api.get_subscription_history",
|
|
return_value={"history": []},
|
|
),
|
|
):
|
|
resp = client.get("/news/api/subscriptions/s1/history")
|
|
assert resp.status_code == 200
|
|
|
|
def test_error_in_result(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_settings_manager", return_value=self._ms()),
|
|
patch(
|
|
f"{MODULE}.api.get_subscription_history",
|
|
return_value={"error": "fail"},
|
|
),
|
|
):
|
|
resp = client.get("/news/api/subscriptions/s1/history")
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestSavePreferences:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(
|
|
f"{MODULE}.api.save_news_preferences",
|
|
return_value={"status": "ok"},
|
|
),
|
|
):
|
|
resp = client.post(
|
|
"/news/api/preferences", json={"preferences": {"theme": "dark"}}
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
|
|
class TestGetCategories:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.get_news_categories",
|
|
return_value={"categories": ["tech"]},
|
|
):
|
|
resp = client.get("/news/api/categories")
|
|
assert resp.status_code == 200
|
|
|
|
def test_exception(self, client):
|
|
_auth(client)
|
|
with patch(
|
|
f"{MODULE}.api.get_news_categories",
|
|
side_effect=RuntimeError("fail"),
|
|
):
|
|
resp = client.get("/news/api/categories")
|
|
assert resp.status_code == 500
|
|
|
|
|
|
class TestSubscriptionStats:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
ctx, _ = _mock_db_session()
|
|
mm = MagicMock()
|
|
mm.get_subscription_stats.return_value = {"total": 5}
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.get("/news/api/subscription/stats")
|
|
assert resp.status_code == 200
|
|
assert resp.get_json()["total"] == 5
|
|
|
|
|
|
class TestGetSubscriptionsOrganized:
|
|
def test_success(self, client):
|
|
"""Returns the flattened {folder_name: [subscription, ...]} map.
|
|
|
|
get_subscriptions_by_folder returns plain dicts in a
|
|
{"folders": [...], "uncategorized": [...]} shape; the route must
|
|
flatten them. (The previous version called .to_dict() on those plain
|
|
dicts and 500'd; this also pins that regression.)
|
|
"""
|
|
_auth(client)
|
|
ctx, _ = _mock_db_session()
|
|
mm = MagicMock()
|
|
mm.get_subscriptions_by_folder.return_value = {
|
|
"folders": [
|
|
{
|
|
"folder": {"id": "f1", "name": "General"},
|
|
"subscriptions": [{"id": "s1", "query_or_topic": "q1"}],
|
|
},
|
|
],
|
|
"uncategorized": [{"id": "s2", "query_or_topic": "q2"}],
|
|
}
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.get("/news/api/subscription/subscriptions/organized")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["General"] == [{"id": "s1", "query_or_topic": "q1"}]
|
|
assert data["uncategorized"] == [
|
|
{"id": "s2", "query_or_topic": "q2"}
|
|
]
|
|
|
|
def test_folder_named_uncategorized_does_not_drop_subscriptions(
|
|
self, client
|
|
):
|
|
"""A user folder literally named "uncategorized" must not have its
|
|
subscriptions clobbered by the ungrouped bucket — both are merged."""
|
|
_auth(client)
|
|
ctx, _ = _mock_db_session()
|
|
mm = MagicMock()
|
|
mm.get_subscriptions_by_folder.return_value = {
|
|
"folders": [
|
|
{
|
|
"folder": {"id": "f1", "name": "uncategorized"},
|
|
"subscriptions": [{"id": "folder_sub"}],
|
|
},
|
|
],
|
|
"uncategorized": [{"id": "ungrouped_sub"}],
|
|
}
|
|
with (
|
|
patch(f"{MODULE}.get_user_id", return_value="testuser"),
|
|
patch(f"{MODULE}.get_user_db_session", return_value=ctx),
|
|
patch(f"{MODULE}.FolderManager", return_value=mm),
|
|
):
|
|
resp = client.get("/news/api/subscription/subscriptions/organized")
|
|
assert resp.status_code == 200
|
|
ids = {s["id"] for s in resp.get_json()["uncategorized"]}
|
|
# Neither the folder's sub nor the ungrouped sub is dropped.
|
|
assert ids == {"folder_sub", "ungrouped_sub"}
|
|
|
|
|
|
SCHED_MOD = "local_deep_research.scheduler.background"
|
|
|
|
|
|
class TestSchedulerStatus:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
ms = MagicMock()
|
|
ms.is_running = True
|
|
ms.config = {"check_interval": 300}
|
|
ms.user_sessions = {"testuser": {"scheduled_jobs": {"j1"}}}
|
|
job = MagicMock()
|
|
job.id = "j1"
|
|
job.name = "Test Job"
|
|
job.args = ("testuser", 1)
|
|
job.next_run_time = None
|
|
ms.scheduler.get_jobs.return_value = [job]
|
|
with (
|
|
patch(f"{MODULE}.get_env_setting", return_value=False),
|
|
patch(f"{SCHED_MOD}.get_background_job_scheduler", return_value=ms),
|
|
):
|
|
resp = client.get("/news/api/scheduler/status")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["is_running"] is True
|
|
assert data["active_users"] == 1
|
|
|
|
|
|
class TestSchedulerStats:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
ms = MagicMock()
|
|
ms.is_running = True
|
|
ms.user_sessions = {}
|
|
ms.scheduler.get_jobs.return_value = []
|
|
with (
|
|
patch(f"{MODULE}.get_env_setting", return_value=True),
|
|
patch(f"{SCHED_MOD}.get_background_job_scheduler", return_value=ms),
|
|
):
|
|
resp = client.get("/news/api/scheduler/stats")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["scheduler_running"] is True
|
|
|
|
|
|
class TestGetActiveUsers:
|
|
def test_success(self, client):
|
|
_auth(client)
|
|
ms = MagicMock()
|
|
ms.get_user_sessions_summary.return_value = [
|
|
{"user_id": "testuser", "last_activity": "2026-01-01"},
|
|
]
|
|
with (
|
|
patch(f"{MODULE}.get_env_setting", return_value=False),
|
|
patch(f"{SCHED_MOD}.get_background_job_scheduler", return_value=ms),
|
|
):
|
|
resp = client.get("/news/api/scheduler/users")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["active_users"] == 1
|
|
assert data["users"][0]["user_id"] == "testuser"
|
|
|
|
|
|
class TestErrorHandlers:
|
|
def test_bad_request_handler(self, app):
|
|
from local_deep_research.news.flask_api import bad_request
|
|
|
|
with app.app_context():
|
|
resp, status = bad_request(None)
|
|
assert status == 400
|
|
|
|
def test_not_found_handler(self, app):
|
|
from local_deep_research.news.flask_api import not_found
|
|
|
|
with app.app_context():
|
|
resp, status = not_found(None)
|
|
assert status == 404
|
|
|
|
def test_internal_error_handler(self, app):
|
|
from local_deep_research.news.flask_api import internal_error
|
|
|
|
with app.app_context():
|
|
resp, status = internal_error(None)
|
|
assert status == 500
|