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
540 lines
17 KiB
Python
540 lines
17 KiB
Python
"""
|
|
Deep behavioral tests for all news exception classes.
|
|
Tests status codes, error codes, message formatting, to_dict(), and inheritance.
|
|
"""
|
|
|
|
import pytest
|
|
|
|
from local_deep_research.news.exceptions import (
|
|
NewsAPIException,
|
|
NewsFeatureDisabledException,
|
|
InvalidLimitException,
|
|
SubscriptionNotFoundException,
|
|
SubscriptionCreationException,
|
|
SubscriptionUpdateException,
|
|
SubscriptionDeletionException,
|
|
DatabaseAccessException,
|
|
NewsFeedGenerationException,
|
|
ResearchProcessingException,
|
|
NotImplementedException,
|
|
InvalidParameterException,
|
|
SchedulerNotificationException,
|
|
)
|
|
|
|
|
|
# --- NewsAPIException base ---
|
|
|
|
|
|
class TestNewsAPIExceptionBase:
|
|
"""Tests for the base NewsAPIException."""
|
|
|
|
def test_message_stored(self):
|
|
e = NewsAPIException("test message")
|
|
assert e.message == "test message"
|
|
|
|
def test_default_status_code(self):
|
|
e = NewsAPIException("msg")
|
|
assert e.status_code == 500
|
|
|
|
def test_custom_status_code(self):
|
|
e = NewsAPIException("msg", status_code=404)
|
|
assert e.status_code == 404
|
|
|
|
def test_default_error_code_is_class_name(self):
|
|
e = NewsAPIException("msg")
|
|
assert e.error_code == "NewsAPIException"
|
|
|
|
def test_custom_error_code(self):
|
|
e = NewsAPIException("msg", error_code="CUSTOM_ERROR")
|
|
assert e.error_code == "CUSTOM_ERROR"
|
|
|
|
def test_default_details_empty(self):
|
|
e = NewsAPIException("msg")
|
|
assert e.details == {}
|
|
|
|
def test_custom_details(self):
|
|
e = NewsAPIException("msg", details={"key": "val"})
|
|
assert e.details == {"key": "val"}
|
|
|
|
def test_is_exception(self):
|
|
e = NewsAPIException("msg")
|
|
assert isinstance(e, Exception)
|
|
|
|
def test_str_is_message(self):
|
|
e = NewsAPIException("test message")
|
|
assert str(e) == "test message"
|
|
|
|
def test_to_dict_has_error(self):
|
|
e = NewsAPIException("msg")
|
|
d = e.to_dict()
|
|
assert d["error"] == "msg"
|
|
|
|
def test_to_dict_has_error_code(self):
|
|
e = NewsAPIException("msg")
|
|
d = e.to_dict()
|
|
assert d["error_code"] == "NewsAPIException"
|
|
|
|
def test_to_dict_has_status_code(self):
|
|
e = NewsAPIException("msg", status_code=400)
|
|
d = e.to_dict()
|
|
assert d["status_code"] == 400
|
|
|
|
def test_to_dict_no_details_when_empty(self):
|
|
e = NewsAPIException("msg")
|
|
d = e.to_dict()
|
|
assert "details" not in d
|
|
|
|
def test_to_dict_includes_details_when_present(self):
|
|
e = NewsAPIException("msg", details={"foo": "bar"})
|
|
d = e.to_dict()
|
|
assert d["details"] == {"foo": "bar"}
|
|
|
|
|
|
# --- NewsFeatureDisabledException ---
|
|
|
|
|
|
class TestNewsFeatureDisabledException:
|
|
"""Tests for NewsFeatureDisabledException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = NewsFeatureDisabledException()
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_default_message(self):
|
|
e = NewsFeatureDisabledException()
|
|
assert e.message == "News system is disabled"
|
|
|
|
def test_custom_message(self):
|
|
e = NewsFeatureDisabledException("custom msg")
|
|
assert e.message == "custom msg"
|
|
|
|
def test_status_code_503(self):
|
|
e = NewsFeatureDisabledException()
|
|
assert e.status_code == 503
|
|
|
|
def test_error_code(self):
|
|
e = NewsFeatureDisabledException()
|
|
assert e.error_code == "NEWS_DISABLED"
|
|
|
|
|
|
# --- InvalidLimitException ---
|
|
|
|
|
|
class TestInvalidLimitException:
|
|
"""Tests for InvalidLimitException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = InvalidLimitException(0)
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_400(self):
|
|
e = InvalidLimitException(-1)
|
|
assert e.status_code == 400
|
|
|
|
def test_error_code(self):
|
|
e = InvalidLimitException(0)
|
|
assert e.error_code == "INVALID_LIMIT"
|
|
|
|
def test_message_contains_limit(self):
|
|
e = InvalidLimitException(-5)
|
|
assert "-5" in e.message
|
|
|
|
def test_details_has_provided_limit(self):
|
|
e = InvalidLimitException(0)
|
|
assert e.details["provided_limit"] == 0
|
|
|
|
def test_details_has_min_limit(self):
|
|
e = InvalidLimitException(0)
|
|
assert e.details["min_limit"] == 1
|
|
|
|
def test_negative_limit(self):
|
|
e = InvalidLimitException(-100)
|
|
assert e.details["provided_limit"] == -100
|
|
|
|
def test_zero_limit(self):
|
|
e = InvalidLimitException(0)
|
|
assert "0" in e.message
|
|
|
|
|
|
# --- SubscriptionNotFoundException ---
|
|
|
|
|
|
class TestSubscriptionNotFoundException:
|
|
"""Tests for SubscriptionNotFoundException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = SubscriptionNotFoundException("sub-123")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_404(self):
|
|
e = SubscriptionNotFoundException("sub-123")
|
|
assert e.status_code == 404
|
|
|
|
def test_error_code(self):
|
|
e = SubscriptionNotFoundException("sub-123")
|
|
assert e.error_code == "SUBSCRIPTION_NOT_FOUND"
|
|
|
|
def test_message_contains_id(self):
|
|
e = SubscriptionNotFoundException("sub-abc-xyz")
|
|
assert "sub-abc-xyz" in e.message
|
|
|
|
def test_details_has_subscription_id(self):
|
|
e = SubscriptionNotFoundException("sub-123")
|
|
assert e.details["subscription_id"] == "sub-123"
|
|
|
|
def test_to_dict_round_trip(self):
|
|
e = SubscriptionNotFoundException("sub-123")
|
|
d = e.to_dict()
|
|
assert d["status_code"] == 404
|
|
assert d["details"]["subscription_id"] == "sub-123"
|
|
|
|
|
|
# --- SubscriptionCreationException ---
|
|
|
|
|
|
class TestSubscriptionCreationException:
|
|
"""Tests for SubscriptionCreationException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = SubscriptionCreationException("error")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_500(self):
|
|
e = SubscriptionCreationException("error")
|
|
assert e.status_code == 500
|
|
|
|
def test_error_code(self):
|
|
e = SubscriptionCreationException("error")
|
|
assert e.error_code == "SUBSCRIPTION_CREATE_FAILED"
|
|
|
|
def test_message_contains_reason(self):
|
|
e = SubscriptionCreationException("DB timeout")
|
|
assert "DB timeout" in e.message
|
|
|
|
def test_with_details(self):
|
|
e = SubscriptionCreationException(
|
|
"fail", details={"query": "test", "type": "search"}
|
|
)
|
|
assert e.details["query"] == "test"
|
|
assert e.details["type"] == "search"
|
|
|
|
def test_without_details(self):
|
|
e = SubscriptionCreationException("fail")
|
|
assert e.details == {} or e.details is None
|
|
|
|
|
|
# --- SubscriptionUpdateException ---
|
|
|
|
|
|
class TestSubscriptionUpdateException:
|
|
"""Tests for SubscriptionUpdateException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = SubscriptionUpdateException("sub-1", "error")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_500(self):
|
|
e = SubscriptionUpdateException("sub-1", "error")
|
|
assert e.status_code == 500
|
|
|
|
def test_error_code(self):
|
|
e = SubscriptionUpdateException("sub-1", "error")
|
|
assert e.error_code == "SUBSCRIPTION_UPDATE_FAILED"
|
|
|
|
def test_message_contains_id_and_reason(self):
|
|
e = SubscriptionUpdateException("sub-99", "DB error")
|
|
assert "sub-99" in e.message
|
|
assert "DB error" in e.message
|
|
|
|
def test_details_has_subscription_id(self):
|
|
e = SubscriptionUpdateException("sub-1", "msg")
|
|
assert e.details["subscription_id"] == "sub-1"
|
|
|
|
|
|
# --- SubscriptionDeletionException ---
|
|
|
|
|
|
class TestSubscriptionDeletionException:
|
|
"""Tests for SubscriptionDeletionException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = SubscriptionDeletionException("sub-1", "error")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_500(self):
|
|
e = SubscriptionDeletionException("sub-1", "error")
|
|
assert e.status_code == 500
|
|
|
|
def test_error_code(self):
|
|
e = SubscriptionDeletionException("sub-1", "error")
|
|
assert e.error_code == "SUBSCRIPTION_DELETE_FAILED"
|
|
|
|
def test_message_contains_id(self):
|
|
e = SubscriptionDeletionException("sub-42", "timeout")
|
|
assert "sub-42" in e.message
|
|
|
|
def test_details_has_subscription_id(self):
|
|
e = SubscriptionDeletionException("sub-1", "msg")
|
|
assert e.details["subscription_id"] == "sub-1"
|
|
|
|
|
|
# --- DatabaseAccessException ---
|
|
|
|
|
|
class TestDatabaseAccessException:
|
|
"""Tests for DatabaseAccessException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = DatabaseAccessException("query", "timeout")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_500(self):
|
|
e = DatabaseAccessException("query", "timeout")
|
|
assert e.status_code == 500
|
|
|
|
def test_error_code(self):
|
|
e = DatabaseAccessException("query", "timeout")
|
|
assert e.error_code == "DATABASE_ERROR"
|
|
|
|
def test_message_contains_operation(self):
|
|
e = DatabaseAccessException("get_news_feed", "connection lost")
|
|
assert "get_news_feed" in e.message
|
|
|
|
def test_message_contains_error(self):
|
|
e = DatabaseAccessException("op", "connection lost")
|
|
assert "connection lost" in e.message
|
|
|
|
def test_details_has_operation(self):
|
|
e = DatabaseAccessException("get_feed", "error")
|
|
assert e.details["operation"] == "get_feed"
|
|
|
|
|
|
# --- NewsFeedGenerationException ---
|
|
|
|
|
|
class TestNewsFeedGenerationException:
|
|
"""Tests for NewsFeedGenerationException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = NewsFeedGenerationException("error")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_500(self):
|
|
e = NewsFeedGenerationException("error")
|
|
assert e.status_code == 500
|
|
|
|
def test_error_code(self):
|
|
e = NewsFeedGenerationException("error")
|
|
assert e.error_code == "FEED_GENERATION_FAILED"
|
|
|
|
def test_message_contains_reason(self):
|
|
e = NewsFeedGenerationException("no data")
|
|
assert "no data" in e.message
|
|
|
|
def test_with_user_id(self):
|
|
e = NewsFeedGenerationException("error", user_id="user1")
|
|
assert e.details["user_id"] == "user1"
|
|
|
|
def test_without_user_id(self):
|
|
e = NewsFeedGenerationException("error")
|
|
assert e.details == {}
|
|
|
|
|
|
# --- ResearchProcessingException ---
|
|
|
|
|
|
class TestResearchProcessingException:
|
|
"""Tests for ResearchProcessingException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = ResearchProcessingException("error")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_500(self):
|
|
e = ResearchProcessingException("error")
|
|
assert e.status_code == 500
|
|
|
|
def test_error_code(self):
|
|
e = ResearchProcessingException("error")
|
|
assert e.error_code == "RESEARCH_PROCESSING_FAILED"
|
|
|
|
def test_with_research_id(self):
|
|
e = ResearchProcessingException("fail", research_id="r-123")
|
|
assert e.details["research_id"] == "r-123"
|
|
|
|
def test_without_research_id(self):
|
|
e = ResearchProcessingException("fail")
|
|
assert e.details == {}
|
|
|
|
|
|
# --- NotImplementedException ---
|
|
|
|
|
|
class TestNotImplementedException:
|
|
"""Tests for NotImplementedException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = NotImplementedException("feature_x")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_501(self):
|
|
e = NotImplementedException("feature_x")
|
|
assert e.status_code == 501
|
|
|
|
def test_error_code(self):
|
|
e = NotImplementedException("feature_x")
|
|
assert e.error_code == "NOT_IMPLEMENTED"
|
|
|
|
def test_message_contains_feature(self):
|
|
e = NotImplementedException("dark_mode")
|
|
assert "dark_mode" in e.message
|
|
|
|
def test_details_has_feature(self):
|
|
e = NotImplementedException("feature_x")
|
|
assert e.details["feature"] == "feature_x"
|
|
|
|
|
|
# --- InvalidParameterException ---
|
|
|
|
|
|
class TestInvalidParameterException:
|
|
"""Tests for InvalidParameterException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = InvalidParameterException("limit", -1, "must be positive")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_400(self):
|
|
e = InvalidParameterException("limit", -1, "must be positive")
|
|
assert e.status_code == 400
|
|
|
|
def test_error_code(self):
|
|
e = InvalidParameterException("limit", -1, "msg")
|
|
assert e.error_code == "INVALID_PARAMETER"
|
|
|
|
def test_message_contains_parameter(self):
|
|
e = InvalidParameterException("limit", -1, "msg")
|
|
assert "limit" in e.message
|
|
|
|
def test_details_has_parameter(self):
|
|
e = InvalidParameterException("limit", -1, "msg")
|
|
assert e.details["parameter"] == "limit"
|
|
|
|
def test_details_has_value(self):
|
|
e = InvalidParameterException("limit", -1, "msg")
|
|
assert e.details["value"] == -1
|
|
|
|
|
|
# --- SchedulerNotificationException ---
|
|
|
|
|
|
class TestSchedulerNotificationException:
|
|
"""Tests for SchedulerNotificationException."""
|
|
|
|
def test_inherits_from_base(self):
|
|
e = SchedulerNotificationException("created", "msg")
|
|
assert isinstance(e, NewsAPIException)
|
|
|
|
def test_status_code_500(self):
|
|
e = SchedulerNotificationException("created", "msg")
|
|
assert e.status_code == 500
|
|
|
|
def test_error_code(self):
|
|
e = SchedulerNotificationException("created", "msg")
|
|
assert e.error_code == "SCHEDULER_NOTIFICATION_FAILED"
|
|
|
|
def test_message_contains_action(self):
|
|
e = SchedulerNotificationException("deleted", "error")
|
|
assert "deleted" in e.message
|
|
|
|
def test_details_has_action(self):
|
|
e = SchedulerNotificationException("updated", "msg")
|
|
assert e.details["action"] == "updated"
|
|
|
|
|
|
# --- Exception hierarchy catch patterns ---
|
|
|
|
|
|
class TestExceptionHierarchyCatching:
|
|
"""Tests that exception hierarchy works correctly for catch blocks."""
|
|
|
|
def test_catch_invalid_limit_as_base(self):
|
|
with pytest.raises(NewsAPIException):
|
|
raise InvalidLimitException(0)
|
|
|
|
def test_catch_not_found_as_base(self):
|
|
with pytest.raises(NewsAPIException):
|
|
raise SubscriptionNotFoundException("id")
|
|
|
|
def test_catch_disabled_as_base(self):
|
|
with pytest.raises(NewsAPIException):
|
|
raise NewsFeatureDisabledException()
|
|
|
|
def test_catch_db_error_as_base(self):
|
|
with pytest.raises(NewsAPIException):
|
|
raise DatabaseAccessException("op", "msg")
|
|
|
|
def test_catch_all_as_exception(self):
|
|
with pytest.raises(Exception):
|
|
raise NewsFeedGenerationException("msg")
|
|
|
|
def test_not_found_not_caught_as_value_error(self):
|
|
"""NewsAPIException should NOT be caught as ValueError."""
|
|
with pytest.raises(NewsAPIException):
|
|
raise SubscriptionNotFoundException("id")
|
|
# If it were a ValueError, this test would fail differently
|
|
|
|
def test_each_exception_has_unique_error_code(self):
|
|
"""All concrete exceptions should have unique error codes."""
|
|
exceptions = [
|
|
NewsFeatureDisabledException(),
|
|
InvalidLimitException(0),
|
|
SubscriptionNotFoundException("id"),
|
|
SubscriptionCreationException("msg"),
|
|
SubscriptionUpdateException("id", "msg"),
|
|
SubscriptionDeletionException("id", "msg"),
|
|
DatabaseAccessException("op", "msg"),
|
|
NewsFeedGenerationException("msg"),
|
|
ResearchProcessingException("msg"),
|
|
NotImplementedException("feature"),
|
|
InvalidParameterException("p", "v", "msg"),
|
|
SchedulerNotificationException("action", "msg"),
|
|
]
|
|
error_codes = [e.error_code for e in exceptions]
|
|
assert len(error_codes) == len(set(error_codes))
|
|
|
|
def test_all_subclasses_are_catchable_as_base(self):
|
|
"""All subclasses should be catchable as NewsAPIException."""
|
|
exceptions = [
|
|
NewsFeatureDisabledException(),
|
|
InvalidLimitException(0),
|
|
SubscriptionNotFoundException("id"),
|
|
SubscriptionCreationException("msg"),
|
|
SubscriptionUpdateException("id", "msg"),
|
|
SubscriptionDeletionException("id", "msg"),
|
|
DatabaseAccessException("op", "msg"),
|
|
NewsFeedGenerationException("msg"),
|
|
ResearchProcessingException("msg"),
|
|
NotImplementedException("feature"),
|
|
InvalidParameterException("p", "v", "msg"),
|
|
SchedulerNotificationException("action", "msg"),
|
|
]
|
|
for exc in exceptions:
|
|
assert isinstance(exc, NewsAPIException)
|
|
|
|
def test_to_dict_always_has_required_fields(self):
|
|
"""All exceptions' to_dict should have error, error_code, status_code."""
|
|
exceptions = [
|
|
NewsAPIException("msg"),
|
|
NewsFeatureDisabledException(),
|
|
InvalidLimitException(0),
|
|
SubscriptionNotFoundException("id"),
|
|
DatabaseAccessException("op", "msg"),
|
|
NotImplementedException("feature"),
|
|
]
|
|
for exc in exceptions:
|
|
d = exc.to_dict()
|
|
assert "error" in d
|
|
assert "error_code" in d
|
|
assert "status_code" in d
|