From ef9810792c8bc28bb83d6a4fdc551c16bb7dc043 Mon Sep 17 00:00:00 2001 From: Curry Date: Sun, 19 Jul 2026 04:19:06 +0800 Subject: [PATCH 1/2] fix(control+notify): emit error_type so SCHEMA_DRIFT actually fires + constrain trigger_event [W1,W2] W1: the control layer's SCHEMA_DRIFT chain (error_kinds -> evaluator -> policies -> actuator) runs every 60s but control/recorder.py only records an error kind when error_type is not None. Of 4 parse-failure sites, only crawl4ai_channel.py passed error_type correctly. Wire the other 3: - rss_channel.py: both collect() and fetch()'s feedparser bozo-flag branch now pass error_type via a new _bozo_error_type() helper, using feedparser's real bozo_exception class name (verified empirically: SAXParseException across malformed markup, truncated declarations, and encoding mismatches), falling back to "ParseError" when none is attached. - cli_channel.py: the json.JSONDecodeError branch now passes error_type=type(exc).__name__ ("JSONDecodeError", already mapped). - error_kinds.py: added "SAXParseException" -> ErrorKind.SCHEMA_DRIFT since it wasn't in the map yet (minimal extension of the existing taxonomy). W2: NotificationRule.trigger_event was a free-text str but dispatch_notifications() only ever queries/fires "on_new_record" (its default and only caller-supplied value) -- a rule saved with any other value became permanently, silently inert with no producer. Constrained trigger_event to Literal["on_new_record"] in NotificationRuleCreate/Update (backend/schemas/notification.py) so the API rejects an unsupported value loudly instead of persisting a dead rule. DB column stays String (no migration) -- this is a validation-layer guard only. --- backend/channels/cli_channel.py | 7 ++- backend/channels/rss_channel.py | 26 ++++++++- backend/control/error_kinds.py | 5 ++ backend/schemas/notification.py | 11 +++- tests/unit/channels/test_cli_channel.py | 9 ++- tests/unit/channels/test_rss_channel.py | 74 +++++++++++++++++++++++++ tests/unit/channels/test_rss_fetch.py | 27 +++++++++ tests/unit/control/test_error_kinds.py | 7 ++- tests/unit/test_schemas_notification.py | 60 ++++++++++++++++++++ 9 files changed, 219 insertions(+), 7 deletions(-) create mode 100644 tests/unit/test_schemas_notification.py diff --git a/backend/channels/cli_channel.py b/backend/channels/cli_channel.py index 35c4c1b..624c716 100644 --- a/backend/channels/cli_channel.py +++ b/backend/channels/cli_channel.py @@ -104,7 +104,12 @@ async def collect( data = json.loads(output) items = data if isinstance(data, list) else [data] except json.JSONDecodeError as exc: - return ChannelResult.fail(f"Failed to parse CLI JSON output: {exc}") + # WIRING_GAP_LEDGER W1: error_type must be set so the SCHEMA_DRIFT + # chain (error_kinds -> control.recorder) actually fires instead of + # being dropped by recorder's `elif error_type is not None` guard. + return ChannelResult.fail( + f"Failed to parse CLI JSON output: {exc}", error_type=type(exc).__name__ + ) else: # Plain text: each line is a record items = [{"line": line} for line in output.splitlines() if line.strip()] diff --git a/backend/channels/rss_channel.py b/backend/channels/rss_channel.py index a254f2b..3c8fbe9 100644 --- a/backend/channels/rss_channel.py +++ b/backend/channels/rss_channel.py @@ -22,6 +22,26 @@ ) +def _bozo_error_type(parsed: Any) -> str: + """error_type for a bozo (unparseable) feed — WIRING_GAP_LEDGER W1. + + feedparser sets ``bozo_exception`` to the underlying parse failure (e.g. + ``xml.sax._exceptions.SAXParseException`` for malformed markup, truncated + declarations, or an encoding mismatch — verified empirically across all + of those shapes; see backend/control/error_kinds.py's SCHEMA_DRIFT set). + Its class name is a real structured ``error_type`` like every other + channel already produces, so it flows through ChannelResult.fail() / + ChannelFetchError into control.recorder instead of being dropped by the + ``elif error_type is not None`` guard there. Falls back to a generic + "ParseError" (also mapped to SCHEMA_DRIFT) on the defensive case where + feedparser didn't attach one. + """ + bozo_exception = getattr(parsed, "bozo_exception", None) + if bozo_exception is not None: + return type(bozo_exception).__name__ + return "ParseError" + + @register_channel class RSSChannel(AbstractChannel): """Collect entries from RSS/Atom feeds.""" @@ -84,7 +104,8 @@ async def collect( parsed = await asyncio.to_thread(feedparser.parse, content) if parsed.bozo and not parsed.entries: return ChannelResult.fail( - f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}" + f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}", + error_type=_bozo_error_type(parsed), ) entries = parsed.entries[:max_entries] @@ -179,7 +200,8 @@ async def fetch(self, ctx: FetchContext) -> FetchResult: parsed = await asyncio.to_thread(feedparser.parse, response.text) if parsed.bozo and not parsed.entries: raise ChannelFetchError( - f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}" + f"Failed to parse feed: {getattr(parsed, 'bozo_exception', 'unknown error')}", + error_type=_bozo_error_type(parsed), ) items = [self._entry_to_dict(entry) for entry in parsed.entries[:max_entries]] diff --git a/backend/control/error_kinds.py b/backend/control/error_kinds.py index 4e5243c..98b58c9 100644 --- a/backend/control/error_kinds.py +++ b/backend/control/error_kinds.py @@ -72,6 +72,11 @@ class ErrorKind(str, Enum): "JSONDecodeError": ErrorKind.SCHEMA_DRIFT, "SchemaDriftError": ErrorKind.SCHEMA_DRIFT, "ParseError": ErrorKind.SCHEMA_DRIFT, + # feedparser's bozo_exception class for malformed/truncated XML or an + # encoding mismatch (rss_channel.py's bozo branch) — verified empirically + # across markup corruption, truncated declarations, and encoding + # mismatches; feedparser's strict parser wraps all of them the same way. + "SAXParseException": ErrorKind.SCHEMA_DRIFT, # Validation / permanent bad input (including SSRF rejections — a # config/validation problem with the source, not a transient network fault) "ValueError": ErrorKind.VALIDATION, diff --git a/backend/schemas/notification.py b/backend/schemas/notification.py index 1746b63..1e78ed7 100644 --- a/backend/schemas/notification.py +++ b/backend/schemas/notification.py @@ -9,7 +9,14 @@ class NotificationRuleCreate(BaseModel): name: str = Field(..., min_length=1, max_length=255) source_id: str | None = None - trigger_event: str + # WIRING_GAP_LEDGER W2: dispatch_notifications() only ever queries/fires + # rules with trigger_event == "on_new_record" (backend/pipeline/ + # notifier_dispatch.py) — there is no producer for any other value. + # Before this Literal, a rule saved with any other string (the free-text + # str field previously accepted anything) became permanently, silently + # inert. Constrained here at the schema layer so the API rejects an + # unsupported value loudly instead of persisting a dead rule. + trigger_event: Literal["on_new_record"] notifier_type: str notifier_config: dict[str, Any] = Field(default_factory=dict) filter_conditions: dict[str, Any] | None = None @@ -18,7 +25,7 @@ class NotificationRuleCreate(BaseModel): class NotificationRuleUpdate(BaseModel): name: str | None = None - trigger_event: str | None = None + trigger_event: Literal["on_new_record"] | None = None notifier_type: str | None = None notifier_config: dict[str, Any] | None = None filter_conditions: dict[str, Any] | None = None diff --git a/tests/unit/channels/test_cli_channel.py b/tests/unit/channels/test_cli_channel.py index 45d71b8..f144246 100644 --- a/tests/unit/channels/test_cli_channel.py +++ b/tests/unit/channels/test_cli_channel.py @@ -254,7 +254,12 @@ async def test_collect_nonzero_exit_code(channel): @pytest.mark.asyncio async def test_collect_invalid_json_output(channel): - """Invalid JSON output returns failed ChannelResult.""" + """Invalid JSON output returns failed ChannelResult with error_type set so + the SCHEMA_DRIFT chain (error_kinds -> control.recorder) actually fires, + instead of being dropped by recorder's `elif error_type is not None` + guard (WIRING_GAP_LEDGER W1).""" + from backend.control.error_kinds import ErrorKind, map_error_type + with _allow(sys.executable): result = await channel.collect( { @@ -266,6 +271,8 @@ async def test_collect_invalid_json_output(channel): ) assert result.success is False assert "parse" in result.error.lower() + assert result.error_type == "JSONDecodeError" + assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT @pytest.mark.asyncio diff --git a/tests/unit/channels/test_rss_channel.py b/tests/unit/channels/test_rss_channel.py index 6f8fd19..2a9fc3d 100644 --- a/tests/unit/channels/test_rss_channel.py +++ b/tests/unit/channels/test_rss_channel.py @@ -253,6 +253,80 @@ async def test_collect_bozo_feed_no_entries_returns_fail(channel): assert result.error is not None +@pytest.mark.asyncio +async def test_collect_bozo_feed_error_type_maps_to_schema_drift(channel): + """WIRING_GAP_LEDGER W1: the bozo failure must carry error_type set from + feedparser's real bozo_exception class (e.g. SAXParseException for + malformed XML) so error_kinds.map_error_type resolves it to SCHEMA_DRIFT + -- previously this branch passed no error_type at all, so control's + recorder (`elif error_type is not None`) silently dropped it and the + SCHEMA_DRIFT chain never fired for a broken RSS feed.""" + from xml.sax import SAXParseException + + from backend.control.error_kinds import ErrorKind, map_error_type + + mock_response = MagicMock() + mock_response.text = "NOT VALID XML AT ALL !!!" + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client_ctx = AsyncMock() + mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_ctx.__aexit__ = AsyncMock(return_value=False) + + # Force feedparser to report bozo with no entries, with the same + # exception class real feedparser attaches for malformed markup. + fake_parsed = MagicMock() + fake_parsed.bozo = True + fake_parsed.entries = [] + fake_parsed.bozo_exception = SAXParseException("syntax error", None, MagicMock()) + + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + with patch("feedparser.parse", return_value=fake_parsed): + result = await channel.collect( + {"feed_url": "https://example.com/rss"}, {} + ) + + assert result.success is False + assert result.error_type == "SAXParseException" + assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT + + +@pytest.mark.asyncio +async def test_collect_bozo_feed_without_bozo_exception_falls_back_to_parse_error(channel): + """Defensive fallback: if feedparser reports bozo with no bozo_exception + attached at all, error_type still lands on a SCHEMA_DRIFT-mapped constant + ("ParseError") instead of None, which the recorder's guard would drop.""" + from types import SimpleNamespace + + from backend.control.error_kinds import ErrorKind, map_error_type + + mock_response = MagicMock() + mock_response.text = "NOT VALID XML AT ALL !!!" + mock_response.raise_for_status = MagicMock() + + mock_client = AsyncMock() + mock_client.get = AsyncMock(return_value=mock_response) + mock_client_ctx = AsyncMock() + mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_client_ctx.__aexit__ = AsyncMock(return_value=False) + + # SimpleNamespace genuinely has no bozo_exception attribute (unlike + # MagicMock, which would auto-vivify one on access). + fake_parsed = SimpleNamespace(bozo=True, entries=[]) + + with patch("httpx.AsyncClient", return_value=mock_client_ctx): + with patch("feedparser.parse", return_value=fake_parsed): + result = await channel.collect( + {"feed_url": "https://example.com/rss"}, {} + ) + + assert result.success is False + assert result.error_type == "ParseError" + assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT + + @pytest.mark.asyncio async def test_collect_bozo_feed_with_entries_succeeds(channel): """A bozo feed that still has entries should succeed (feedparser partial parse).""" diff --git a/tests/unit/channels/test_rss_fetch.py b/tests/unit/channels/test_rss_fetch.py index d0d53e1..c3eb4eb 100644 --- a/tests/unit/channels/test_rss_fetch.py +++ b/tests/unit/channels/test_rss_fetch.py @@ -164,6 +164,33 @@ async def test_fetch_client_404_classified_permanent(): assert exc_info.value.error_type == "PermanentHTTPStatus" +@pytest.mark.asyncio +async def test_fetch_bozo_feed_raises_error_type_mapped_to_schema_drift(): + """WIRING_GAP_LEDGER W1: fetch()'s bozo branch must carry error_type (from + feedparser's real bozo_exception class) so error_kinds.map_error_type + resolves it to SCHEMA_DRIFT -- previously this raise passed no error_type, + so control.recorder's `elif error_type is not None` guard silently + dropped it and the SCHEMA_DRIFT chain never fired.""" + from xml.sax import SAXParseException + + from backend.control.error_kinds import ErrorKind, map_error_type + + http = _Http(_Resp(200, text="NOT VALID XML AT ALL !!!", headers={})) + ctx = FetchContext(config={"feed_url": "https://x/feed"}, params={}, cursor=None, http=http) + + fake_parsed = MagicMock() + fake_parsed.bozo = True + fake_parsed.entries = [] + fake_parsed.bozo_exception = SAXParseException("syntax error", None, MagicMock()) + + with patch("feedparser.parse", return_value=fake_parsed): + with pytest.raises(ChannelFetchError) as exc_info: + await RSSChannel().fetch(ctx) + + assert exc_info.value.error_type == "SAXParseException" + assert map_error_type(exc_info.value.error_type) is ErrorKind.SCHEMA_DRIFT + + @pytest.mark.asyncio async def test_run_channel_drives_rss_and_persists_cursor(): from backend.pipeline.channel_runner import run_channel diff --git a/tests/unit/control/test_error_kinds.py b/tests/unit/control/test_error_kinds.py index 097c871..458c66c 100644 --- a/tests/unit/control/test_error_kinds.py +++ b/tests/unit/control/test_error_kinds.py @@ -35,7 +35,12 @@ def test_validation_types_including_ssrf(self): assert map_error_type(t) is ErrorKind.VALIDATION def test_schema_drift(self): - assert map_error_type("JSONDecodeError") is ErrorKind.SCHEMA_DRIFT + # WIRING_GAP_LEDGER W1: SAXParseException is feedparser's real + # bozo_exception class for a malformed RSS/Atom feed (rss_channel.py's + # bozo branch) -- verified empirically across markup corruption, + # truncated declarations, and encoding mismatches. + for t in ("JSONDecodeError", "SchemaDriftError", "ParseError", "SAXParseException"): + assert map_error_type(t) is ErrorKind.SCHEMA_DRIFT def test_store_failed(self): assert map_error_type("IntegrityError") is ErrorKind.STORE_FAILED diff --git a/tests/unit/test_schemas_notification.py b/tests/unit/test_schemas_notification.py new file mode 100644 index 0000000..e5cd2f7 --- /dev/null +++ b/tests/unit/test_schemas_notification.py @@ -0,0 +1,60 @@ +"""Tests for backend/schemas/notification.py -- WIRING_GAP_LEDGER W2. + +NotificationRule.trigger_event used to be an unconstrained str field. +dispatch_notifications() only ever queries/fires rules with +trigger_event == "on_new_record" (backend/pipeline/notifier_dispatch.py) -- +there is no producer for any other value, so a rule saved with anything else +became permanently, silently inert. Constrained to Literal["on_new_record"] +at the schema layer (Create/Update, the write paths) so the API rejects an +unsupported value loudly instead of persisting a dead rule. +""" + +import pytest +from pydantic import ValidationError + +from backend.schemas.notification import NotificationRuleCreate, NotificationRuleUpdate + + +class TestNotificationRuleCreateTriggerEvent: + def test_on_new_record_accepted(self): + rule = NotificationRuleCreate( + name="r1", trigger_event="on_new_record", notifier_type="webhook" + ) + assert rule.trigger_event == "on_new_record" + + def test_unsupported_value_rejected(self): + """on_task_failed reads like a plausible trigger (it's even named in + the model's stale comment) but has no dispatch producer -- must be + rejected, not silently accepted into a dead rule.""" + with pytest.raises(ValidationError): + NotificationRuleCreate( + name="r1", trigger_event="on_task_failed", notifier_type="webhook" + ) + + def test_arbitrary_string_rejected(self): + with pytest.raises(ValidationError): + NotificationRuleCreate( + name="r1", trigger_event="on_ai_processed", notifier_type="webhook" + ) + + def test_missing_trigger_event_still_required(self): + """trigger_event stays required on create -- only the type narrowed + from str to Literal, not the field's optionality.""" + with pytest.raises(ValidationError): + NotificationRuleCreate(name="r1", notifier_type="webhook") + + +class TestNotificationRuleUpdateTriggerEvent: + def test_on_new_record_accepted(self): + update = NotificationRuleUpdate(trigger_event="on_new_record") + assert update.trigger_event == "on_new_record" + + def test_unsupported_value_rejected(self): + with pytest.raises(ValidationError): + NotificationRuleUpdate(trigger_event="on_task_failed") + + def test_omitted_trigger_event_stays_none(self): + """Update is a partial patch -- omitting trigger_event entirely must + keep working (only a non-None value is constrained to the Literal).""" + update = NotificationRuleUpdate(name="renamed") + assert update.trigger_event is None From 9f1c5ce8d559d1d0a42b02d863ff427db2099e91 Mon Sep 17 00:00:00 2001 From: 2233admin <2233admin@users.noreply.github.com> Date: Sun, 19 Jul 2026 05:27:46 +0800 Subject: [PATCH 2/2] test: split channel suites by concern --- tests/unit/channels/test_cli_channel.py | 255 +----------- .../channels/test_cli_channel_allowlist.py | 108 ++++++ .../channels/test_cli_channel_execution.py | 145 +++++++ tests/unit/channels/test_rss_channel.py | 367 ++---------------- .../unit/channels/test_rss_channel_entries.py | 57 +++ .../unit/channels/test_rss_channel_errors.py | 93 +++++ .../channels/test_rss_channel_schema_drift.py | 88 +++++ tests/unit/channels/test_rss_fetch.py | 71 +--- .../unit/channels/test_rss_fetch_statuses.py | 73 ++++ 9 files changed, 631 insertions(+), 626 deletions(-) create mode 100644 tests/unit/channels/test_cli_channel_allowlist.py create mode 100644 tests/unit/channels/test_cli_channel_execution.py create mode 100644 tests/unit/channels/test_rss_channel_entries.py create mode 100644 tests/unit/channels/test_rss_channel_errors.py create mode 100644 tests/unit/channels/test_rss_channel_schema_drift.py create mode 100644 tests/unit/channels/test_rss_fetch_statuses.py diff --git a/tests/unit/channels/test_cli_channel.py b/tests/unit/channels/test_cli_channel.py index f144246..382f04d 100644 --- a/tests/unit/channels/test_cli_channel.py +++ b/tests/unit/channels/test_cli_channel.py @@ -1,25 +1,13 @@ -"""Unit tests for the CLI channel. - -The cli channel executes arbitrary binaries (ADR-0005, audit P0-4), so every -execution test must explicitly allowlist its binary via ``_allow`` — the -default (empty allowlist) denies everything. -""" - -import sys -from unittest.mock import AsyncMock, Mock, patch +"""Core unit tests for the CLI channel.""" import pytest -from backend.channels.cli_channel import CLIChannel, _binary_allowed, _render_template -from backend.config import Settings +from backend.channels.cli_channel import CLIChannel, _render_template -def _allow(*binaries: str): - """Patch settings with the given binary allowlist for the test's duration.""" - return patch( - "backend.config.get_settings", - return_value=Settings(cli_channel_allowed_binaries=",".join(binaries)), - ) +@pytest.fixture +def channel(): + return CLIChannel() def test_render_template_basic(): @@ -35,246 +23,29 @@ def test_render_template_multiple_keys(): assert result == "foo and bar" -@pytest.fixture -def channel(): - return CLIChannel() - - @pytest.mark.asyncio async def test_validate_config_missing_binary(channel): errors = await channel.validate_config({"command": ["search"]}) - assert any("binary" in e for e in errors) + assert any("binary" in error for error in errors) @pytest.mark.asyncio async def test_validate_config_missing_command(channel): errors = await channel.validate_config({"binary": "mycli"}) - assert any("command" in e for e in errors) + assert any("command" in error for error in errors) @pytest.mark.asyncio async def test_validate_config_valid(channel): - errors = await channel.validate_config({ - "binary": "mycli", - "command": ["search", "--keyword", "test"], - }) + errors = await channel.validate_config( + { + "binary": "mycli", + "command": ["search", "--keyword", "test"], + } + ) assert errors == [] -# ── Binary allowlist (ADR-0005, issue 05) ──────────────────────────────────── - - -def test_binary_allowed_normalizes_paths(): - assert _binary_allowed("./mycli", ["mycli"]) is True - assert _binary_allowed("mycli", ["mycli"]) is True - assert _binary_allowed("mycli", ["othercli"]) is False - assert _binary_allowed("mycli", []) is False - - -@pytest.mark.asyncio -async def test_collect_empty_allowlist_rejects_all(channel): - """Default deny: with no allowlist configured, nothing may run.""" - with _allow(): - result = await channel.collect( - {"binary": sys.executable, "command": ["-c", "print('hi')"]}, - {}, - ) - assert result.success is False - assert "allowlist" in result.error - assert result.error_type == "BinaryNotAllowedError" - - -@pytest.mark.asyncio -async def test_collect_unlisted_binary_rejected(channel): - """A non-empty allowlist still rejects any binary not on it.""" - with _allow("/usr/bin/some-other-tool"): - result = await channel.collect( - {"binary": sys.executable, "command": ["-c", "print('hi')"]}, - {}, - ) - assert result.success is False - assert result.error_type == "BinaryNotAllowedError" - - -@pytest.mark.asyncio -async def test_collect_allowlisted_binary_executes(channel): - with _allow(sys.executable): - result = await channel.collect( - { - "binary": sys.executable, - "command": ["-c", "print('[{\"ok\": true}]')"], - "output_format": "json", - }, - {}, - ) - assert result.success is True - assert result.items == [{"ok": True}] - - -@pytest.mark.asyncio -async def test_allowlist_rejection_spawns_no_subprocess(channel): - """Enforcement happens BEFORE execution — no process is ever created.""" - with _allow(), patch("asyncio.create_subprocess_exec") as spawn: - result = await channel.collect( - {"binary": sys.executable, "command": ["-c", "print('hi')"]}, - {}, - ) - assert result.success is False - spawn.assert_not_called() - - -def test_allowlist_rejection_is_permanent(): - """The taxonomy classifies the rejection non-retryable.""" - from backend.pipeline.error_taxonomy import is_retryable - - assert is_retryable("BinaryNotAllowedError") is False - - -@pytest.mark.asyncio -async def test_allowlist_rejection_permanent_through_fetch_seam(channel): - """End-to-end at the runner seam: fetch() wraps the rejection in - ChannelFetchError carrying error_type, and the taxonomy classifies it - non-retryable — no parallel error path.""" - from backend.channels.base import ChannelFetchError, FetchContext - from backend.pipeline.error_taxonomy import effective_error_type, is_retryable - - with _allow(): - with pytest.raises(ChannelFetchError) as excinfo: - await channel.fetch( - FetchContext( - config={"binary": sys.executable, "command": ["-c", "print(1)"]}, - params={}, - ) - ) - assert is_retryable(effective_error_type(excinfo.value)) is False - - -# ── Execution behaviour (binaries explicitly allowlisted) ──────────────────── - - -@pytest.mark.asyncio -async def test_collect_binary_not_found(channel): - with _allow("nonexistent_binary_xyz"): - result = await channel.collect( - {"binary": "nonexistent_binary_xyz", "command": ["run"]}, - {}, - ) - assert result.success is False - assert "not found" in result.error.lower() - - -@pytest.mark.asyncio -async def test_collect_json_output(channel): - import json - data = [{"title": "Test"}, {"title": "Other"}] - json_str = json.dumps(data) - - with _allow(sys.executable): - result = await channel.collect( - { - "binary": sys.executable, - "command": ["-c", f"print({json_str!r})"], - "output_format": "json", - }, - {}, - ) - assert result.success is True - assert len(result.items) == 2 - - -@pytest.mark.asyncio -async def test_collect_text_output(channel): - with _allow(sys.executable): - result = await channel.collect( - { - "binary": sys.executable, - "command": ["-c", "print('line1'); print('line2'); print('line3')"], - "output_format": "text", - }, - {}, - ) - assert result.success is True - assert len(result.items) == 3 - - -@pytest.mark.asyncio -async def test_collect_timeout(channel): - """asyncio.TimeoutError returns failed ChannelResult and kills the child - so a timed-out subprocess is never orphaned (issue 05).""" - import asyncio - - mock_proc = AsyncMock() - mock_proc.kill = Mock() - with ( - _allow(sys.executable), - patch("asyncio.create_subprocess_exec", return_value=mock_proc), - patch("asyncio.wait_for", side_effect=asyncio.TimeoutError()), - ): - result = await channel.collect( - { - "binary": sys.executable, - "command": ["-c", "import time; time.sleep(10)"], - "timeout": 1, - }, - {}, - ) - - assert result.success is False - assert "timed out" in result.error.lower() - mock_proc.kill.assert_called_once() - - -@pytest.mark.asyncio -async def test_collect_generic_exception(channel): - """Generic exception during subprocess exec returns failed ChannelResult.""" - with ( - _allow(sys.executable), - patch("asyncio.create_subprocess_exec", side_effect=OSError("unexpected error")), - ): - result = await channel.collect( - {"binary": sys.executable, "command": ["-c", "print('hi')"]}, - {}, - ) - - assert result.success is False - assert "CLI execution failed" in result.error - - -@pytest.mark.asyncio -async def test_collect_nonzero_exit_code(channel): - """Non-zero exit code from subprocess returns failed ChannelResult.""" - with _allow(sys.executable): - result = await channel.collect( - {"binary": sys.executable, "command": ["-c", "import sys; sys.exit(1)"]}, - {}, - ) - assert result.success is False - assert "exited with code" in result.error.lower() - - -@pytest.mark.asyncio -async def test_collect_invalid_json_output(channel): - """Invalid JSON output returns failed ChannelResult with error_type set so - the SCHEMA_DRIFT chain (error_kinds -> control.recorder) actually fires, - instead of being dropped by recorder's `elif error_type is not None` - guard (WIRING_GAP_LEDGER W1).""" - from backend.control.error_kinds import ErrorKind, map_error_type - - with _allow(sys.executable): - result = await channel.collect( - { - "binary": sys.executable, - "command": ["-c", "print('not valid json')"], - "output_format": "json", - }, - {}, - ) - assert result.success is False - assert "parse" in result.error.lower() - assert result.error_type == "JSONDecodeError" - assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT - - @pytest.mark.asyncio async def test_health_check(channel): """health_check always returns True (binary checked per collect).""" diff --git a/tests/unit/channels/test_cli_channel_allowlist.py b/tests/unit/channels/test_cli_channel_allowlist.py new file mode 100644 index 0000000..59fddf9 --- /dev/null +++ b/tests/unit/channels/test_cli_channel_allowlist.py @@ -0,0 +1,108 @@ +"""Binary allowlist tests for the CLI channel (ADR-0005, issue 05).""" + +import sys +from unittest.mock import patch + +import pytest + +from backend.channels.cli_channel import CLIChannel, _binary_allowed +from backend.config import Settings + + +@pytest.fixture +def channel(): + return CLIChannel() + + +def _allow(*binaries: str): + """Patch settings with the given binary allowlist for a test.""" + return patch( + "backend.config.get_settings", + return_value=Settings(cli_channel_allowed_binaries=",".join(binaries)), + ) + + +def test_binary_allowed_normalizes_paths(): + assert _binary_allowed("./mycli", ["mycli"]) is True + assert _binary_allowed("mycli", ["mycli"]) is True + assert _binary_allowed("mycli", ["othercli"]) is False + assert _binary_allowed("mycli", []) is False + + +@pytest.mark.asyncio +async def test_collect_empty_allowlist_rejects_all(channel): + """Default deny: with no allowlist configured, nothing may run.""" + with _allow(): + result = await channel.collect( + {"binary": sys.executable, "command": ["-c", "print('hi')"]}, + {}, + ) + assert result.success is False + assert "allowlist" in result.error + assert result.error_type == "BinaryNotAllowedError" + + +@pytest.mark.asyncio +async def test_collect_unlisted_binary_rejected(channel): + """A non-empty allowlist still rejects any binary not on it.""" + with _allow("/usr/bin/some-other-tool"): + result = await channel.collect( + {"binary": sys.executable, "command": ["-c", "print('hi')"]}, + {}, + ) + assert result.success is False + assert result.error_type == "BinaryNotAllowedError" + + +@pytest.mark.asyncio +async def test_collect_allowlisted_binary_executes(channel): + with _allow(sys.executable): + result = await channel.collect( + { + "binary": sys.executable, + "command": ["-c", "print('[{\"ok\": true}]')"], + "output_format": "json", + }, + {}, + ) + assert result.success is True + assert result.items == [{"ok": True}] + + +@pytest.mark.asyncio +async def test_allowlist_rejection_spawns_no_subprocess(channel): + """Enforcement happens before execution, so no process is created.""" + with _allow(), patch("asyncio.create_subprocess_exec") as spawn: + result = await channel.collect( + {"binary": sys.executable, "command": ["-c", "print('hi')"]}, + {}, + ) + assert result.success is False + spawn.assert_not_called() + + +def test_allowlist_rejection_is_permanent(): + """The taxonomy classifies the rejection non-retryable.""" + from backend.pipeline.error_taxonomy import is_retryable + + assert is_retryable("BinaryNotAllowedError") is False + + +@pytest.mark.asyncio +async def test_allowlist_rejection_permanent_through_fetch_seam(channel): + """The fetch seam preserves the permanent allowlist rejection.""" + from backend.channels.base import ChannelFetchError, FetchContext + from backend.pipeline.error_taxonomy import effective_error_type, is_retryable + + with _allow(): + with pytest.raises(ChannelFetchError) as excinfo: + await channel.fetch( + FetchContext( + config={ + "binary": sys.executable, + "command": ["-c", "print(1)"], + }, + params={}, + ) + ) + assert is_retryable(effective_error_type(excinfo.value)) is False diff --git a/tests/unit/channels/test_cli_channel_execution.py b/tests/unit/channels/test_cli_channel_execution.py new file mode 100644 index 0000000..2eae234 --- /dev/null +++ b/tests/unit/channels/test_cli_channel_execution.py @@ -0,0 +1,145 @@ +"""Execution behaviour tests for explicitly allowlisted CLI binaries.""" + +import json +import sys +from unittest.mock import AsyncMock, Mock, patch + +import pytest + +from backend.channels.cli_channel import CLIChannel +from backend.config import Settings + + +@pytest.fixture +def channel(): + return CLIChannel() + + +def _allow(*binaries: str): + """Patch settings with the given binary allowlist for a test.""" + return patch( + "backend.config.get_settings", + return_value=Settings(cli_channel_allowed_binaries=",".join(binaries)), + ) + + +@pytest.mark.asyncio +async def test_collect_binary_not_found(channel): + with _allow("nonexistent_binary_xyz"): + result = await channel.collect( + {"binary": "nonexistent_binary_xyz", "command": ["run"]}, + {}, + ) + assert result.success is False + assert "not found" in result.error.lower() + + +@pytest.mark.asyncio +async def test_collect_json_output(channel): + data = [{"title": "Test"}, {"title": "Other"}] + json_str = json.dumps(data) + + with _allow(sys.executable): + result = await channel.collect( + { + "binary": sys.executable, + "command": ["-c", f"print({json_str!r})"], + "output_format": "json", + }, + {}, + ) + assert result.success is True + assert len(result.items) == 2 + + +@pytest.mark.asyncio +async def test_collect_text_output(channel): + with _allow(sys.executable): + result = await channel.collect( + { + "binary": sys.executable, + "command": ["-c", "print('line1'); print('line2'); print('line3')"], + "output_format": "text", + }, + {}, + ) + assert result.success is True + assert len(result.items) == 3 + + +@pytest.mark.asyncio +async def test_collect_timeout(channel): + """Timeout kills the child so a subprocess is never orphaned.""" + mock_proc = AsyncMock() + mock_proc.kill = Mock() + + async def timeout(awaitable, *, timeout): + del timeout + awaitable.close() + raise TimeoutError + + with ( + _allow(sys.executable), + patch("asyncio.create_subprocess_exec", return_value=mock_proc), + patch("asyncio.wait_for", side_effect=timeout), + ): + result = await channel.collect( + { + "binary": sys.executable, + "command": ["-c", "import time; time.sleep(10)"], + "timeout": 1, + }, + {}, + ) + + assert result.success is False + assert "timed out" in result.error.lower() + mock_proc.kill.assert_called_once() + + +@pytest.mark.asyncio +async def test_collect_generic_exception(channel): + """Generic subprocess exceptions return a failed ChannelResult.""" + with ( + _allow(sys.executable), + patch("asyncio.create_subprocess_exec", side_effect=OSError("unexpected error")), + ): + result = await channel.collect( + {"binary": sys.executable, "command": ["-c", "print('hi')"]}, + {}, + ) + + assert result.success is False + assert "CLI execution failed" in result.error + + +@pytest.mark.asyncio +async def test_collect_nonzero_exit_code(channel): + """A non-zero exit code returns a failed ChannelResult.""" + with _allow(sys.executable): + result = await channel.collect( + {"binary": sys.executable, "command": ["-c", "import sys; sys.exit(1)"]}, + {}, + ) + assert result.success is False + assert "exited with code" in result.error.lower() + + +@pytest.mark.asyncio +async def test_collect_invalid_json_output(channel): + """Invalid JSON carries the error type needed by the SCHEMA_DRIFT chain.""" + from backend.control.error_kinds import ErrorKind, map_error_type + + with _allow(sys.executable): + result = await channel.collect( + { + "binary": sys.executable, + "command": ["-c", "print('not valid json')"], + "output_format": "json", + }, + {}, + ) + assert result.success is False + assert "parse" in result.error.lower() + assert result.error_type == "JSONDecodeError" + assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT diff --git a/tests/unit/channels/test_rss_channel.py b/tests/unit/channels/test_rss_channel.py index 2a9fc3d..47ba7bc 100644 --- a/tests/unit/channels/test_rss_channel.py +++ b/tests/unit/channels/test_rss_channel.py @@ -1,18 +1,12 @@ -"""Unit tests for the RSS channel.""" +"""Core unit tests for the RSS channel.""" import threading -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import MagicMock, patch import pytest from backend.channels.rss_channel import RSSChannel - -@pytest.fixture -def channel(): - return RSSChannel() - - VALID_RSS_XML = """ @@ -30,7 +24,25 @@ def channel(): """ -# ── validate_config ──────────────────────────────────────────────────────────── +@pytest.fixture +def channel(): + return RSSChannel() + + +class _HttpClient: + def __init__(self, text: str): + self.response = MagicMock(text=text) + self.response.raise_for_status = MagicMock() + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + async def get(self, *_args, **_kwargs): + return self.response + @pytest.mark.asyncio async def test_channel_type(channel): @@ -40,7 +52,7 @@ async def test_channel_type(channel): @pytest.mark.asyncio async def test_validate_config_missing_feed_url(channel): errors = await channel.validate_config({}) - assert any("feed_url" in e for e in errors) + assert any("feed_url" in error for error in errors) @pytest.mark.asyncio @@ -49,25 +61,11 @@ async def test_validate_config_valid(channel): assert errors == [] -# ── collect: success ─────────────────────────────────────────────────────────── - @pytest.mark.asyncio async def test_collect_success_returns_items(channel): - """Successful RSS fetch returns ChannelResult with parsed items.""" - mock_response = MagicMock() - mock_response.text = VALID_RSS_XML - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) + """Successful RSS fetch returns parsed items.""" + with patch("httpx.AsyncClient", return_value=_HttpClient(VALID_RSS_XML)): + result = await channel.collect({"feed_url": "https://example.com/rss"}, {}) assert result.success is True assert len(result.items) == 2 @@ -77,20 +75,11 @@ async def test_collect_success_returns_items(channel): @pytest.mark.asyncio async def test_collect_max_entries_limits_results(channel): - """max_entries config option trims entries.""" - mock_response = MagicMock() - mock_response.text = VALID_RSS_XML - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): + """max_entries trims parsed entries.""" + with patch("httpx.AsyncClient", return_value=_HttpClient(VALID_RSS_XML)): result = await channel.collect( - {"feed_url": "https://example.com/rss", "max_entries": 1}, {} + {"feed_url": "https://example.com/rss", "max_entries": 1}, + {}, ) assert result.success is True @@ -100,42 +89,16 @@ async def test_collect_max_entries_limits_results(channel): @pytest.mark.asyncio async def test_collect_metadata_includes_feed_title(channel): """ChannelResult metadata contains the feed title.""" - mock_response = MagicMock() - mock_response.text = VALID_RSS_XML - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) + with patch("httpx.AsyncClient", return_value=_HttpClient(VALID_RSS_XML)): + result = await channel.collect({"feed_url": "https://example.com/rss"}, {}) assert result.success is True assert result.metadata.get("feed_title") == "Test Feed" -# ── AUDIT C22: feedparser.parse() off-loaded via asyncio.to_thread ───────────── - @pytest.mark.asyncio async def test_collect_parses_feed_off_event_loop_thread(channel): - """feedparser.parse() must run via asyncio.to_thread, not inline on the - event loop — a multi-MB feed would otherwise freeze every other - request/task on this process for the duration of the parse.""" - mock_response = MagicMock() - mock_response.text = VALID_RSS_XML - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - + """feedparser.parse runs via asyncio.to_thread instead of on the event loop.""" import feedparser seen_threads: list[int] = [] @@ -145,268 +108,12 @@ def spy_parse(content): seen_threads.append(threading.get_ident()) return real_parse(content) - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - with patch("feedparser.parse", side_effect=spy_parse): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) + with ( + patch("httpx.AsyncClient", return_value=_HttpClient(VALID_RSS_XML)), + patch("feedparser.parse", side_effect=spy_parse), + ): + result = await channel.collect({"feed_url": "https://example.com/rss"}, {}) assert result.success is True assert len(seen_threads) == 1 assert seen_threads[0] != threading.get_ident() - - -# ── collect: error cases ─────────────────────────────────────────────────────── - -@pytest.mark.asyncio -async def test_collect_timeout_returns_fail(channel): - """TimeoutException should produce a failed ChannelResult.""" - import httpx - - mock_client = AsyncMock() - mock_client.get = AsyncMock(side_effect=httpx.TimeoutException("timeout")) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) - - assert result.success is False - assert "timed out" in result.error.lower() - - -@pytest.mark.asyncio -async def test_collect_http_404_returns_fail(channel): - """HTTP 404 status should produce a failed ChannelResult.""" - import httpx - - mock_response = MagicMock() - mock_response.status_code = 404 - mock_response.raise_for_status = MagicMock( - side_effect=httpx.HTTPStatusError( - message="Not Found", - request=MagicMock(), - response=MagicMock(status_code=404), - ) - ) - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) - - assert result.success is False - assert "404" in result.error - - -@pytest.mark.asyncio -async def test_collect_generic_exception_returns_fail(channel): - """Any other request exception should produce a failed ChannelResult.""" - mock_client = AsyncMock() - mock_client.get = AsyncMock(side_effect=ConnectionError("network down")) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) - - assert result.success is False - assert "Failed to fetch" in result.error - - -@pytest.mark.asyncio -async def test_collect_bozo_feed_no_entries_returns_fail(channel): - """A bozo (broken) feed with no entries should return a failed ChannelResult.""" - mock_response = MagicMock() - mock_response.text = "NOT VALID XML AT ALL !!!" - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - # Force feedparser to report bozo with no entries - fake_parsed = MagicMock() - fake_parsed.bozo = True - fake_parsed.entries = [] - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - with patch("feedparser.parse", return_value=fake_parsed): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) - - assert result.success is False - assert result.error is not None - - -@pytest.mark.asyncio -async def test_collect_bozo_feed_error_type_maps_to_schema_drift(channel): - """WIRING_GAP_LEDGER W1: the bozo failure must carry error_type set from - feedparser's real bozo_exception class (e.g. SAXParseException for - malformed XML) so error_kinds.map_error_type resolves it to SCHEMA_DRIFT - -- previously this branch passed no error_type at all, so control's - recorder (`elif error_type is not None`) silently dropped it and the - SCHEMA_DRIFT chain never fired for a broken RSS feed.""" - from xml.sax import SAXParseException - - from backend.control.error_kinds import ErrorKind, map_error_type - - mock_response = MagicMock() - mock_response.text = "NOT VALID XML AT ALL !!!" - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - # Force feedparser to report bozo with no entries, with the same - # exception class real feedparser attaches for malformed markup. - fake_parsed = MagicMock() - fake_parsed.bozo = True - fake_parsed.entries = [] - fake_parsed.bozo_exception = SAXParseException("syntax error", None, MagicMock()) - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - with patch("feedparser.parse", return_value=fake_parsed): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) - - assert result.success is False - assert result.error_type == "SAXParseException" - assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT - - -@pytest.mark.asyncio -async def test_collect_bozo_feed_without_bozo_exception_falls_back_to_parse_error(channel): - """Defensive fallback: if feedparser reports bozo with no bozo_exception - attached at all, error_type still lands on a SCHEMA_DRIFT-mapped constant - ("ParseError") instead of None, which the recorder's guard would drop.""" - from types import SimpleNamespace - - from backend.control.error_kinds import ErrorKind, map_error_type - - mock_response = MagicMock() - mock_response.text = "NOT VALID XML AT ALL !!!" - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - # SimpleNamespace genuinely has no bozo_exception attribute (unlike - # MagicMock, which would auto-vivify one on access). - fake_parsed = SimpleNamespace(bozo=True, entries=[]) - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - with patch("feedparser.parse", return_value=fake_parsed): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) - - assert result.success is False - assert result.error_type == "ParseError" - assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT - - -@pytest.mark.asyncio -async def test_collect_bozo_feed_with_entries_succeeds(channel): - """A bozo feed that still has entries should succeed (feedparser partial parse).""" - mock_response = MagicMock() - mock_response.text = VALID_RSS_XML - mock_response.raise_for_status = MagicMock() - - mock_client = AsyncMock() - mock_client.get = AsyncMock(return_value=mock_response) - mock_client_ctx = AsyncMock() - mock_client_ctx.__aenter__ = AsyncMock(return_value=mock_client) - mock_client_ctx.__aexit__ = AsyncMock(return_value=False) - - fake_entry = MagicMock() - fake_entry.get = lambda k, default="": { - "title": "Partial Item", - "link": "https://ex.com/p", - "summary": "", - "author": "", - "published": "", - "id": "pid1", - "tags": [], - }.get(k, default) - - fake_parsed = MagicMock() - fake_parsed.bozo = True - fake_parsed.entries = [fake_entry] - fake_parsed.feed = MagicMock() - fake_parsed.feed.get = lambda k, default="": default - - with patch("httpx.AsyncClient", return_value=mock_client_ctx): - with patch("feedparser.parse", return_value=fake_parsed): - result = await channel.collect( - {"feed_url": "https://example.com/rss"}, {} - ) - - assert result.success is True - assert len(result.items) == 1 - - -# ── _entry_to_dict ───────────────────────────────────────────────────────────── - -@pytest.mark.asyncio -async def test_entry_to_dict(channel): - class FakeEntry: - def get(self, key, default=""): - data = { - "title": "Test Title", - "link": "https://example.com/post", - "summary": "A summary", - "author": "Alice", - "published": "2024-01-01", - "tags": [{"term": "python"}], - "id": "abc123", - } - return data.get(key, default) - - result = channel._entry_to_dict(FakeEntry()) - assert result["title"] == "Test Title" - assert result["link"] == "https://example.com/post" - assert "python" in result["tags"] - - -def test_entry_to_dict_missing_optional_fields(channel): - """Entry with only link should use link as id fallback.""" - class MinimalEntry: - def get(self, key, default=""): - return {"link": "https://ex.com/x"}.get(key, default) - - result = channel._entry_to_dict(MinimalEntry()) - assert result["link"] == "https://ex.com/x" - assert result["id"] == "https://ex.com/x" - assert result["tags"] == [] - - -def test_entry_to_dict_all_tags(channel): - """Multiple tags are all extracted.""" - class TaggedEntry: - def get(self, key, default=""): - return {"tags": [{"term": "a"}, {"term": "b"}, {"term": "c"}]}.get(key, default) - - result = channel._entry_to_dict(TaggedEntry()) - assert result["tags"] == ["a", "b", "c"] diff --git a/tests/unit/channels/test_rss_channel_entries.py b/tests/unit/channels/test_rss_channel_entries.py new file mode 100644 index 0000000..8abb45b --- /dev/null +++ b/tests/unit/channels/test_rss_channel_entries.py @@ -0,0 +1,57 @@ +"""RSS entry mapping tests.""" + +import pytest + +from backend.channels.rss_channel import RSSChannel + + +@pytest.fixture +def channel(): + return RSSChannel() + + +@pytest.mark.asyncio +async def test_entry_to_dict(channel): + class FakeEntry: + def get(self, key, default=""): + data = { + "title": "Test Title", + "link": "https://example.com/post", + "summary": "A summary", + "author": "Alice", + "published": "2024-01-01", + "tags": [{"term": "python"}], + "id": "abc123", + } + return data.get(key, default) + + result = channel._entry_to_dict(FakeEntry()) + assert result["title"] == "Test Title" + assert result["link"] == "https://example.com/post" + assert "python" in result["tags"] + + +def test_entry_to_dict_missing_optional_fields(channel): + """An entry with only a link uses it as the id fallback.""" + + class MinimalEntry: + def get(self, key, default=""): + return {"link": "https://ex.com/x"}.get(key, default) + + result = channel._entry_to_dict(MinimalEntry()) + assert result["link"] == "https://ex.com/x" + assert result["id"] == "https://ex.com/x" + assert result["tags"] == [] + + +def test_entry_to_dict_all_tags(channel): + """All tag terms are extracted.""" + + class TaggedEntry: + def get(self, key, default=""): + return { + "tags": [{"term": "a"}, {"term": "b"}, {"term": "c"}] + }.get(key, default) + + result = channel._entry_to_dict(TaggedEntry()) + assert result["tags"] == ["a", "b", "c"] diff --git a/tests/unit/channels/test_rss_channel_errors.py b/tests/unit/channels/test_rss_channel_errors.py new file mode 100644 index 0000000..c286d87 --- /dev/null +++ b/tests/unit/channels/test_rss_channel_errors.py @@ -0,0 +1,93 @@ +"""Request and parse failure tests for the RSS channel.""" + +from unittest.mock import MagicMock, patch + +import httpx +import pytest + +from backend.channels.rss_channel import RSSChannel + + +@pytest.fixture +def channel(): + return RSSChannel() + + +class _HttpClient: + def __init__(self, *, response=None, error=None): + self.response = response + self.error = error + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + async def get(self, *_args, **_kwargs): + if self.error is not None: + raise self.error + return self.response + + +@pytest.mark.asyncio +async def test_collect_timeout_returns_fail(channel): + """TimeoutException produces a failed ChannelResult.""" + context = _HttpClient(error=httpx.TimeoutException("timeout")) + with patch("httpx.AsyncClient", return_value=context): + result = await channel.collect({"feed_url": "https://example.com/rss"}, {}) + + assert result.success is False + assert "timed out" in result.error.lower() + + +@pytest.mark.asyncio +async def test_collect_http_404_returns_fail(channel): + """HTTP 404 produces a failed ChannelResult.""" + response = MagicMock(status_code=404) + response.raise_for_status = MagicMock( + side_effect=httpx.HTTPStatusError( + message="Not Found", + request=MagicMock(), + response=MagicMock(status_code=404), + ) + ) + with patch( + "httpx.AsyncClient", + return_value=_HttpClient(response=response), + ): + result = await channel.collect({"feed_url": "https://example.com/rss"}, {}) + + assert result.success is False + assert "404" in result.error + + +@pytest.mark.asyncio +async def test_collect_generic_exception_returns_fail(channel): + """Other request exceptions produce a failed ChannelResult.""" + context = _HttpClient(error=ConnectionError("network down")) + with patch("httpx.AsyncClient", return_value=context): + result = await channel.collect({"feed_url": "https://example.com/rss"}, {}) + + assert result.success is False + assert "Failed to fetch" in result.error + + +@pytest.mark.asyncio +async def test_collect_bozo_feed_no_entries_returns_fail(channel): + """A broken feed with no entries returns a failed ChannelResult.""" + response = MagicMock(text="NOT VALID XML AT ALL !!!") + response.raise_for_status = MagicMock() + parsed = MagicMock(bozo=True, entries=[]) + + with ( + patch( + "httpx.AsyncClient", + return_value=_HttpClient(response=response), + ), + patch("feedparser.parse", return_value=parsed), + ): + result = await channel.collect({"feed_url": "https://example.com/rss"}, {}) + + assert result.success is False + assert result.error is not None diff --git a/tests/unit/channels/test_rss_channel_schema_drift.py b/tests/unit/channels/test_rss_channel_schema_drift.py new file mode 100644 index 0000000..e4b52fa --- /dev/null +++ b/tests/unit/channels/test_rss_channel_schema_drift.py @@ -0,0 +1,88 @@ +"""RSS parser behaviour and SCHEMA_DRIFT wiring tests.""" + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch +from xml.sax import SAXParseException + +import pytest + +from backend.channels.rss_channel import RSSChannel +from backend.control.error_kinds import ErrorKind, map_error_type + + +@pytest.fixture +def channel(): + return RSSChannel() + + +class _HttpClient: + def __init__(self): + self.response = MagicMock(text="NOT VALID XML AT ALL !!!") + self.response.raise_for_status = MagicMock() + + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return False + + async def get(self, *_args, **_kwargs): + return self.response + + +async def _collect_with_parsed(channel, parsed): + with ( + patch("httpx.AsyncClient", return_value=_HttpClient()), + patch("feedparser.parse", return_value=parsed), + ): + return await channel.collect({"feed_url": "https://example.com/rss"}, {}) + + +@pytest.mark.asyncio +async def test_collect_bozo_feed_error_type_maps_to_schema_drift(channel): + """A real bozo exception type reaches the SCHEMA_DRIFT mapper.""" + parsed = MagicMock(bozo=True, entries=[]) + parsed.bozo_exception = SAXParseException("syntax error", None, MagicMock()) + + result = await _collect_with_parsed(channel, parsed) + + assert result.success is False + assert result.error_type == "SAXParseException" + assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT + + +@pytest.mark.asyncio +async def test_collect_bozo_feed_without_bozo_exception_falls_back_to_parse_error( + channel, +): + """Missing bozo_exception still produces a mapped ParseError.""" + parsed = SimpleNamespace(bozo=True, entries=[]) + + result = await _collect_with_parsed(channel, parsed) + + assert result.success is False + assert result.error_type == "ParseError" + assert map_error_type(result.error_type) is ErrorKind.SCHEMA_DRIFT + + +@pytest.mark.asyncio +async def test_collect_bozo_feed_with_entries_succeeds(channel): + """A bozo feed that still has entries succeeds as a partial parse.""" + entry = MagicMock() + entry.get = lambda key, default="": { + "title": "Partial Item", + "link": "https://ex.com/p", + "summary": "", + "author": "", + "published": "", + "id": "pid1", + "tags": [], + }.get(key, default) + parsed = MagicMock(bozo=True, entries=[entry]) + parsed.feed = MagicMock() + parsed.feed.get = lambda key, default="": default + + result = await _collect_with_parsed(channel, parsed) + + assert result.success is True + assert len(result.items) == 1 diff --git a/tests/unit/channels/test_rss_fetch.py b/tests/unit/channels/test_rss_fetch.py index c3eb4eb..8b15649 100644 --- a/tests/unit/channels/test_rss_fetch.py +++ b/tests/unit/channels/test_rss_fetch.py @@ -64,13 +64,20 @@ def test_rss_identity_uses_entry_id(): @pytest.mark.asyncio async def test_fetch_200_returns_items_and_advances_cursor(): - http = _Http(_Resp(200, text=_RSS, headers={"ETag": 'W/"v2"', "Last-Modified": "Wed, 01 Jul 2026 00:00:00 GMT"})) + headers = { + "ETag": 'W/"v2"', + "Last-Modified": "Wed, 01 Jul 2026 00:00:00 GMT", + } + http = _Http(_Resp(200, text=_RSS, headers=headers)) ctx = FetchContext(config={"feed_url": "https://x/feed"}, params={}, cursor=None, http=http) result = await RSSChannel().fetch(ctx) assert [i["id"] for i in result.items] == ["id-a", "id-b"] - assert result.next_cursor == {"etag": 'W/"v2"', "last_modified": "Wed, 01 Jul 2026 00:00:00 GMT"} + assert result.next_cursor == { + "etag": 'W/"v2"', + "last_modified": "Wed, 01 Jul 2026 00:00:00 GMT", + } assert result.has_more is False @@ -113,57 +120,6 @@ async def test_fetch_304_no_new_items_keeps_cursor_and_sends_conditional(): assert http.calls[0][1]["headers"]["If-None-Match"] == 'W/"v1"' -# ── AUDIT C13: gateway statuses classify retryable, other 4xx stay permanent ── - -class _StatusErrorResp: - """A response whose raise_for_status() raises a REAL httpx.HTTPStatusError - (unlike this file's own _Resp fake, whose raise_for_status raises a bare - RuntimeError) — needed to exercise fetch()'s httpx.HTTPStatusError - classification branch, which _Resp never triggers.""" - - def __init__(self, status_code): - import httpx - - self.status_code = status_code - self.headers = {} - self._exc = httpx.HTTPStatusError( - message=f"HTTP {status_code}", - request=MagicMock(), - response=MagicMock(status_code=status_code, text=f"error {status_code}"), - ) - - def raise_for_status(self): - raise self._exc - - -@pytest.mark.asyncio -@pytest.mark.parametrize("status", [502, 503, 504, 520, 522, 524]) -async def test_fetch_gateway_status_classified_retryable(status): - """fetch()'s response.raise_for_status() used to be bare (no try/except - at all) — any status error, gateway or not, propagated as a raw - httpx.HTTPStatusError with no retry-classification hint. Now a gateway - status must raise ChannelFetchError(error_type="RetryableHTTPStatus").""" - http = _Http(_StatusErrorResp(status)) - ctx = FetchContext(config={"feed_url": "https://x/feed"}, params={}, cursor=None, http=http) - - with pytest.raises(ChannelFetchError) as exc_info: - await RSSChannel().fetch(ctx) - - assert exc_info.value.error_type == "RetryableHTTPStatus" - - -@pytest.mark.asyncio -async def test_fetch_client_404_classified_permanent(): - """A genuine 4xx (not 408/429) stays permanent.""" - http = _Http(_StatusErrorResp(404)) - ctx = FetchContext(config={"feed_url": "https://x/feed"}, params={}, cursor=None, http=http) - - with pytest.raises(ChannelFetchError) as exc_info: - await RSSChannel().fetch(ctx) - - assert exc_info.value.error_type == "PermanentHTTPStatus" - - @pytest.mark.asyncio async def test_fetch_bozo_feed_raises_error_type_mapped_to_schema_drift(): """WIRING_GAP_LEDGER W1: fetch()'s bozo branch must carry error_type (from @@ -202,7 +158,14 @@ async def test_run_channel_drives_rss_and_persists_cursor(): http = _Http(_Resp(200, text=_RSS, headers={"ETag": 'W/"v2"'})) store = InMemoryCursorStore() - items = (await run_channel(source, {}, cursor_store=store, channel=RSSChannel(), http=http)).items + result = await run_channel( + source, + {}, + cursor_store=store, + channel=RSSChannel(), + http=http, + ) + items = result.items assert [i["id"] for i in items] == ["id-a", "id-b"] # Incremental channel: the runner persisted the advanced cursor. diff --git a/tests/unit/channels/test_rss_fetch_statuses.py b/tests/unit/channels/test_rss_fetch_statuses.py new file mode 100644 index 0000000..6bf8608 --- /dev/null +++ b/tests/unit/channels/test_rss_fetch_statuses.py @@ -0,0 +1,73 @@ +"""HTTP status classification tests for the RSS fetch contract.""" + +from unittest.mock import MagicMock, patch + +import httpx +import pytest + +from backend.channels.base import ChannelFetchError, FetchContext +from backend.channels.rss_channel import RSSChannel + + +@pytest.fixture(autouse=True) +def _fake_dns(): + """Keep placeholder feed hosts independent from live DNS.""" + with patch( + "socket.getaddrinfo", + return_value=[(None, None, None, "", ("93.184.216.34", 0))], + ): + yield + + +class _Http: + """Minimal stand-in for the runner's rate-limited client.""" + + def __init__(self, response): + self._response = response + + async def get(self, _url, **_kwargs): + return self._response + + +class _StatusErrorResp: + """Response whose status check raises a real httpx status error.""" + + def __init__(self, status_code): + self.status_code = status_code + self.headers = {} + self._exception = httpx.HTTPStatusError( + message=f"HTTP {status_code}", + request=MagicMock(), + response=MagicMock(status_code=status_code, text=f"error {status_code}"), + ) + + def raise_for_status(self): + raise self._exception + + +def _fetch_context(status_code: int) -> FetchContext: + return FetchContext( + config={"feed_url": "https://x/feed"}, + params={}, + cursor=None, + http=_Http(_StatusErrorResp(status_code)), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [502, 503, 504, 520, 522, 524]) +async def test_fetch_gateway_status_classified_retryable(status): + """Gateway responses carry the retryable status classification.""" + with pytest.raises(ChannelFetchError) as exc_info: + await RSSChannel().fetch(_fetch_context(status)) + + assert exc_info.value.error_type == "RetryableHTTPStatus" + + +@pytest.mark.asyncio +async def test_fetch_client_404_classified_permanent(): + """A genuine 4xx response stays permanent.""" + with pytest.raises(ChannelFetchError) as exc_info: + await RSSChannel().fetch(_fetch_context(404)) + + assert exc_info.value.error_type == "PermanentHTTPStatus"