From 17a4a19b7f9a6ad94c14d2dcf570046d59d35452 Mon Sep 17 00:00:00 2001 From: minsoo0506 Date: Tue, 7 Jul 2026 17:25:53 +0900 Subject: [PATCH] =?UTF-8?q?[Feat]=20=ED=8E=98=EC=9D=B4=EC=A7=80=20?= =?UTF-8?q?=EC=8A=A4=EB=83=85=EC=83=B7=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/core/config.py | 9 ++ app/schemas/__init__.py | 3 + app/schemas/analysis.py | 3 + app/schemas/db_independent_pipeline.py | 2 + app/schemas/page_snapshot.py | 20 ++++ app/schemas/pipeline.py | 2 + app/services/page_snapshot.py | 130 +++++++++++++++++++++ app/services/pipeline.py | 56 +++++++++ tests/services/test_page_snapshot.py | 65 +++++++++++ tests/services/test_pipeline.py | 150 ++++++++++++++++++++++++- 10 files changed, 437 insertions(+), 3 deletions(-) create mode 100644 app/schemas/page_snapshot.py create mode 100644 app/services/page_snapshot.py create mode 100644 tests/services/test_page_snapshot.py diff --git a/app/core/config.py b/app/core/config.py index 878eca9..0f71a04 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -194,6 +194,15 @@ def alembic_database_url(self) -> str: content_render_settle_ms: int = 500 content_render_concurrency: int = 2 + # 사용자 표시용 페이지 스냅샷. 파이프라인 응답 SLA 보호를 위해 동기 경로에서는 + # 기본 1초까지만 기다리고, 지연/실패는 verdict 와 분리해 상태로만 내려준다. + page_snapshot_enabled: bool = True + page_snapshot_timeout_seconds: float = 1.0 + page_snapshot_storage_dir: str = "/private/tmp/linclean-page-snapshots" + page_snapshot_viewport_width: int = 1365 + page_snapshot_viewport_height: int = 768 + page_snapshot_full_page: bool = False + # 콘텐츠 분석 점수 score_weight_brand_impersonation: int = 50 score_weight_logo_alt_impersonation: int = 10 diff --git a/app/schemas/__init__.py b/app/schemas/__init__.py index 3f490a0..ed62638 100644 --- a/app/schemas/__init__.py +++ b/app/schemas/__init__.py @@ -4,10 +4,13 @@ DbIndependentPipelineStages, DbIndependentPipelineSuccess, ) +from app.schemas.page_snapshot import PageSnapshotResult, PageSnapshotStatus __all__ = [ "DbIndependentPipelineFailure", "DbIndependentPipelineResult", "DbIndependentPipelineStages", "DbIndependentPipelineSuccess", + "PageSnapshotResult", + "PageSnapshotStatus", ] diff --git a/app/schemas/analysis.py b/app/schemas/analysis.py index abf8554..6da5975 100644 --- a/app/schemas/analysis.py +++ b/app/schemas/analysis.py @@ -16,6 +16,7 @@ ) from app.schemas.domain_heuristic import DomainHeuristicResult, DomainHeuristicSignal, RdapInfo from app.schemas.normalize import NormalizeResult +from app.schemas.page_snapshot import PageSnapshotResult, PageSnapshotStatus from app.schemas.pipeline import ( PipelineFailure, PipelineResult, @@ -47,6 +48,8 @@ "GSBResult", "HopRecord", "NormalizeResult", + "PageSnapshotResult", + "PageSnapshotStatus", "PipelineFailure", "PipelineResult", "PipelineStage", diff --git a/app/schemas/db_independent_pipeline.py b/app/schemas/db_independent_pipeline.py index 62da8f0..fa8ea5d 100644 --- a/app/schemas/db_independent_pipeline.py +++ b/app/schemas/db_independent_pipeline.py @@ -5,6 +5,7 @@ from app.schemas.content_analysis import ContentAnalysisResult from app.schemas.domain_heuristic import DomainHeuristicResult from app.schemas.normalize import NormalizeResult +from app.schemas.page_snapshot import PageSnapshotResult from app.schemas.pipeline import PipelineStage, PipelineTimings, Verdict from app.schemas.unchain import UnchainResult @@ -27,6 +28,7 @@ class DbIndependentPipelineSuccess(BaseModel): final_url: str verdict: Verdict score: int = Field(ge=0, le=100) + snapshot: PageSnapshotResult | None = None timings: PipelineTimings | None = None stages: DbIndependentPipelineStages diff --git a/app/schemas/page_snapshot.py b/app/schemas/page_snapshot.py new file mode 100644 index 0000000..1e236c5 --- /dev/null +++ b/app/schemas/page_snapshot.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from enum import StrEnum + +from pydantic import BaseModel, Field + + +class PageSnapshotStatus(StrEnum): + AVAILABLE = "available" + SKIPPED = "skipped" + TIMEOUT = "timeout" + FAILED = "failed" + + +class PageSnapshotResult(BaseModel): + status: PageSnapshotStatus + final_url: str + storage_key: str | None = None + elapsed_seconds: float | None = Field(default=None, ge=0) + error: str | None = None diff --git a/app/schemas/pipeline.py b/app/schemas/pipeline.py index 6e5d9c4..ad2c061 100644 --- a/app/schemas/pipeline.py +++ b/app/schemas/pipeline.py @@ -6,6 +6,7 @@ from app.schemas.content_analysis import ContentAnalysisResult from app.schemas.domain_heuristic import DomainHeuristicResult from app.schemas.normalize import NormalizeResult +from app.schemas.page_snapshot import PageSnapshotResult from app.schemas.threat_db import ThreatDbResult from app.schemas.unchain import UnchainResult @@ -69,6 +70,7 @@ class PipelineSuccess(BaseModel): verdict: Verdict score: int = Field(ge=0, le=100) summary: str | None = None + snapshot: PageSnapshotResult | None = None timings: PipelineTimings | None = None stages: PipelineStages diff --git a/app/services/page_snapshot.py b/app/services/page_snapshot.py new file mode 100644 index 0000000..7270c51 --- /dev/null +++ b/app/services/page_snapshot.py @@ -0,0 +1,130 @@ +"""사용자 표시용 페이지 스냅샷 생성.""" + +from __future__ import annotations + +import asyncio +import re +import time +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from app.core.config import settings +from app.core.logging import get_logger +from app.schemas.page_snapshot import PageSnapshotResult, PageSnapshotStatus +from app.services.content_analyzer.fetch import _pick_user_agent +from app.services.content_analyzer.render import _target_blocked + +logger = get_logger(__name__) + +_SAFE_KEY = re.compile(r"[^A-Za-z0-9_.-]+") + + +def _elapsed_seconds(started: float) -> float: + return round(time.perf_counter() - started, 6) + + +def skipped_page_snapshot(final_url: str, reason: str) -> PageSnapshotResult: + return PageSnapshotResult( + status=PageSnapshotStatus.SKIPPED, + final_url=final_url, + error=reason, + ) + + +def timed_out_page_snapshot(final_url: str, started: float | None = None) -> PageSnapshotResult: + return PageSnapshotResult( + status=PageSnapshotStatus.TIMEOUT, + final_url=final_url, + elapsed_seconds=_elapsed_seconds(started) if started is not None else None, + error="timeout", + ) + + +def failed_page_snapshot( + final_url: str, + *, + error: str, + started: float | None = None, +) -> PageSnapshotResult: + return PageSnapshotResult( + status=PageSnapshotStatus.FAILED, + final_url=final_url, + elapsed_seconds=_elapsed_seconds(started) if started is not None else None, + error=error, + ) + + +def _storage_path(analysis_id: str) -> tuple[Path, str]: + safe_id = _SAFE_KEY.sub("-", analysis_id).strip("-") or "snapshot" + storage_dir = Path(settings.page_snapshot_storage_dir) + filename = f"{safe_id}.png" + return storage_dir / filename, filename + + +def _load_playwright() -> tuple[type[BaseException], Callable[[], Any]]: + from playwright.async_api import TimeoutError as PlaywrightTimeoutError + from playwright.async_api import async_playwright + + return PlaywrightTimeoutError, async_playwright + + +async def capture_page_snapshot(analysis_id: str, final_url: str) -> PageSnapshotResult: + """Playwright 로 최종 URL 첫 화면 PNG 를 저장한다. + + 보안 차단은 콘텐츠 렌더링과 같은 대상 host 검사를 재사용한다. Playwright 가 없거나 + 브라우저 실행이 실패하면 파이프라인을 깨지 않고 failed 상태를 반환한다. + """ + started = time.perf_counter() + + if not settings.page_snapshot_enabled: + return skipped_page_snapshot(final_url, "snapshot_disabled") + + if await _target_blocked(final_url): + logger.info("page_snapshot.blocked_host", url=final_url) + return failed_page_snapshot(final_url, error="blocked_host", started=started) + + try: + playwright_timeout_error, async_playwright = _load_playwright() + except ImportError: + return failed_page_snapshot(final_url, error="playwright_unavailable", started=started) + + path, storage_key = _storage_path(analysis_id) + path.parent.mkdir(parents=True, exist_ok=True) + timeout_ms = int(settings.page_snapshot_timeout_seconds * 1000) + + try: + async with async_playwright() as p: + browser = await p.chromium.launch(headless=True) + try: + page = await browser.new_page( + user_agent=_pick_user_agent(), + locale="ko-KR", + viewport={ + "width": settings.page_snapshot_viewport_width, + "height": settings.page_snapshot_viewport_height, + }, + ) + await page.goto(final_url, wait_until="domcontentloaded", timeout=timeout_ms) + await page.screenshot(path=str(path), full_page=settings.page_snapshot_full_page) + finally: + await browser.close() + except playwright_timeout_error: + return timed_out_page_snapshot(final_url, started) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning( + "page_snapshot.failed", + url=final_url, + error=str(exc), + error_type=type(exc).__name__, + ) + return failed_page_snapshot(final_url, error="snapshot_failed", started=started) + + return PageSnapshotResult( + status=PageSnapshotStatus.AVAILABLE, + final_url=final_url, + storage_key=storage_key, + elapsed_seconds=_elapsed_seconds(started), + ) diff --git a/app/services/pipeline.py b/app/services/pipeline.py index 2e4b10f..7a0664b 100644 --- a/app/services/pipeline.py +++ b/app/services/pipeline.py @@ -21,6 +21,7 @@ DomainHeuristicSkippedReason, ) from app.schemas.normalize import NormalizeResult +from app.schemas.page_snapshot import PageSnapshotResult from app.schemas.pipeline import ( PipelineFailure, PipelineStage, @@ -36,6 +37,12 @@ from app.services.content_analyzer import analyze_content, skipped_already_danger from app.services.domain_heuristic import check_domain_heuristic from app.services.normalizer import normalize_url +from app.services.page_snapshot import ( + capture_page_snapshot, + failed_page_snapshot, + skipped_page_snapshot, + timed_out_page_snapshot, +) from app.services.page_unavailability import ( PAGE_UNAVAILABLE_CODE, content_page_unavailable, @@ -213,6 +220,40 @@ async def _stage_content_analysis( return result +async def _capture_snapshot_for_pipeline( + log: structlog.stdlib.BoundLogger, + analysis_id: str, + final_url: str, +) -> PageSnapshotResult: + started = time.perf_counter() + try: + result = await asyncio.wait_for( + capture_page_snapshot(analysis_id, final_url), + timeout=settings.page_snapshot_timeout_seconds, + ) + except TimeoutError: + result = timed_out_page_snapshot(final_url, started) + except asyncio.CancelledError: + raise + except Exception as exc: + log.warning( + "pipeline.page_snapshot.failed", + final_url=final_url, + error=str(exc), + error_type=type(exc).__name__, + ) + result = failed_page_snapshot(final_url, error="snapshot_failed", started=started) + + log.info( + "pipeline.page_snapshot.done", + status=result.status.value, + storage_key=result.storage_key, + elapsed_seconds=result.elapsed_seconds, + error=result.error, + ) + return result + + def _preceding_score(threat: ThreatDbResult, heuristic: DomainHeuristicResult) -> int: """2~3단계 누적 점수. 4단계 건너뛸지 판단하는 기준.""" score = heuristic.score @@ -307,6 +348,7 @@ def _pipeline_success( threat: ThreatDbResult, heuristic: DomainHeuristicResult, content: ContentAnalysisResult, + snapshot: PageSnapshotResult | None = None, ) -> PipelineSuccess: return PipelineSuccess( analysis_id=analysis_id, @@ -320,6 +362,7 @@ def _pipeline_success( heuristic=heuristic, content=content, ), + snapshot=snapshot, timings=_build_timings(started, stage_timings), stages=PipelineStages( normalize=normalize, @@ -499,6 +542,7 @@ async def run_pipeline( score = _total_score(threat, heuristic, content) if threat.is_malicious or score >= settings.score_caution_threshold: verdict = _decide_verdict(score, threat) + page_unavailable_snapshot = skipped_page_snapshot(unchain.final_url, "page_unavailable") return _pipeline_success( analysis_id=analysis_id, original_url=original_url, @@ -512,6 +556,7 @@ async def run_pipeline( threat=threat, heuristic=heuristic, content=content, + snapshot=page_unavailable_snapshot, ) return _page_unavailable_failure( analysis_id=analysis_id, @@ -557,6 +602,8 @@ async def run_pipeline( short_circuited = False heuristic = _augment_heuristic_with_redirect_signals(heuristic, unchain) + snapshot: PageSnapshotResult | None = None + snapshot_task: asyncio.Task[PageSnapshotResult] | None = None if short_circuited: log.info( @@ -568,6 +615,7 @@ async def run_pipeline( stage_started = time.perf_counter() content = skipped_already_danger(unchain.final_url) _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started) + snapshot = skipped_page_snapshot(unchain.final_url, "skipped_already_danger") else: # known malicious 는 verdict 가 이미 외부 DB 로 확정됐으므로 페이지를 받아보지 않는다. # 휴리스틱 danger 는 페이지가 존재하지 않을 수 있으므로 content fetch 로 가용성을 확인한다. @@ -581,8 +629,12 @@ async def run_pipeline( stage_started = time.perf_counter() content = skipped_already_danger(unchain.final_url) _set_stage_timing(stage_timings, PipelineStage.CONTENT_ANALYSIS, stage_started) + snapshot = skipped_page_snapshot(unchain.final_url, "skipped_already_danger") else: upstream = _collect_upstream_signals(threat, heuristic) + snapshot_task = asyncio.create_task( + _capture_snapshot_for_pipeline(log, analysis_id, unchain.final_url) + ) try: content = await deadline.run( PipelineStage.CONTENT_ANALYSIS.value, @@ -610,6 +662,8 @@ async def run_pipeline( ) content = timed_out_content_result(unchain.final_url) + snapshot = await snapshot_task + if unavailable := content_page_unavailable(content): message, status_code = unavailable log.info( @@ -635,6 +689,7 @@ async def run_pipeline( threat=threat, heuristic=heuristic, content=content, + snapshot=snapshot, ) return _page_unavailable_failure( analysis_id=analysis_id, @@ -668,4 +723,5 @@ async def run_pipeline( threat=threat, heuristic=heuristic, content=content, + snapshot=snapshot, ) diff --git a/tests/services/test_page_snapshot.py b/tests/services/test_page_snapshot.py new file mode 100644 index 0000000..705b085 --- /dev/null +++ b/tests/services/test_page_snapshot.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from app.core.config import settings +from app.schemas.page_snapshot import PageSnapshotStatus +from app.services.page_snapshot import capture_page_snapshot, skipped_page_snapshot + + +def test_skipped_page_snapshot_marks_reason() -> None: + result = skipped_page_snapshot("https://example.com/", "skipped_already_danger") + + assert result.status == PageSnapshotStatus.SKIPPED + assert result.final_url == "https://example.com/" + assert result.error == "skipped_already_danger" + assert result.storage_key is None + + +@pytest.mark.asyncio +async def test_capture_page_snapshot_returns_failed_when_target_is_blocked() -> None: + with patch("app.services.page_snapshot._target_blocked", new_callable=AsyncMock) as blocked: + blocked.return_value = True + + result = await capture_page_snapshot("aid-blocked", "http://127.0.0.1/") + + assert result.status == PageSnapshotStatus.FAILED + assert result.error == "blocked_host" + assert result.elapsed_seconds is not None + + +@pytest.mark.asyncio +async def test_capture_page_snapshot_sanitizes_storage_key( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(settings, "page_snapshot_storage_dir", str(tmp_path)) + + with ( + patch("app.services.page_snapshot._target_blocked", new_callable=AsyncMock) as blocked, + patch("app.services.page_snapshot._load_playwright") as load_playwright, + ): + blocked.return_value = False + page = AsyncMock() + browser = AsyncMock() + browser.new_page.return_value = page + chromium = AsyncMock() + chromium.launch.return_value = browser + playwright_context = MagicMock() + playwright_context.__aenter__ = AsyncMock( + return_value=MagicMock(chromium=chromium) + ) + playwright_context.__aexit__ = AsyncMock(return_value=None) + async_playwright = MagicMock(return_value=playwright_context) + load_playwright.return_value = (TimeoutError, async_playwright) + + result = await capture_page_snapshot("aid/#45 snapshot", "https://example.com/") + + assert result.status == PageSnapshotStatus.AVAILABLE + assert result.storage_key == "aid-45-snapshot.png" + page.screenshot.assert_awaited_once_with( + path=str(tmp_path / "aid-45-snapshot.png"), + full_page=settings.page_snapshot_full_page, + ) diff --git a/tests/services/test_pipeline.py b/tests/services/test_pipeline.py index 3958ab3..65e267d 100644 --- a/tests/services/test_pipeline.py +++ b/tests/services/test_pipeline.py @@ -4,6 +4,7 @@ import asyncio import inspect +import time from typing import TYPE_CHECKING from unittest.mock import AsyncMock, patch @@ -12,6 +13,7 @@ from app.schemas.content_analysis import ContentAnalysisResult, ContentSignal from app.schemas.domain_heuristic import DomainHeuristicResult, DomainHeuristicSignal from app.schemas.normalize import NormalizeResult +from app.schemas.page_snapshot import PageSnapshotResult, PageSnapshotStatus from app.schemas.pipeline import PipelineFailure, PipelineStage, PipelineSuccess, Verdict from app.schemas.threat_db import GSBMatch, GSBResult, ThreatDbResult, URLhausResult from app.schemas.unchain import HopRecord, UnchainResult @@ -147,6 +149,150 @@ async def test_run_pipeline_includes_domain_heuristic_stage(async_session: Async assert await _resolve_upstream(kwargs["upstream_signals"]) == ("HOSTING_PLATFORM",) +@pytest.mark.asyncio +async def test_run_pipeline_no_ai_passes_flag_to_content_analysis( + async_session: AsyncSession, +) -> None: + final_url = "https://example.com/" + + with ( + patch("app.services.pipeline.normalize_url") as mock_norm, + patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain, + patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat, + patch( + "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock + ) as mock_heuristic, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content, + ): + mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) + mock_unchain.return_value = _make_unchain(final_url) + mock_threat.return_value = _make_threat(final_url) + mock_heuristic.return_value = _make_heuristic("example.com") + mock_content.return_value = _make_content(final_url) + + result = await run_pipeline("aid-no-ai", final_url, async_session, use_ai=False) + + assert isinstance(result, PipelineSuccess) + mock_content.assert_awaited_once() + assert mock_content.await_args.kwargs["use_ai"] is False + + +@pytest.mark.asyncio +async def test_run_pipeline_includes_available_page_snapshot( + async_session: AsyncSession, +) -> None: + final_url = "https://example.com/" + + with ( + patch("app.services.pipeline.normalize_url") as mock_norm, + patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain, + patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat, + patch( + "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock + ) as mock_heuristic, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content, + patch( + "app.services.pipeline.capture_page_snapshot", new_callable=AsyncMock + ) as mock_snapshot, + ): + mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) + mock_unchain.return_value = _make_unchain(final_url) + mock_threat.return_value = _make_threat(final_url) + mock_heuristic.return_value = _make_heuristic("example.com") + mock_content.return_value = _make_content(final_url) + mock_snapshot.return_value = PageSnapshotResult( + status=PageSnapshotStatus.AVAILABLE, + final_url=final_url, + storage_key="page-snapshots/aid-snapshot.png", + elapsed_seconds=0.25, + ) + + result = await run_pipeline("aid-snapshot", final_url, async_session) + + assert isinstance(result, PipelineSuccess) + assert result.snapshot is not None + assert result.snapshot.status == PageSnapshotStatus.AVAILABLE + assert result.snapshot.storage_key == "page-snapshots/aid-snapshot.png" + assert result.snapshot.elapsed_seconds == 0.25 + mock_snapshot.assert_awaited_once_with("aid-snapshot", final_url) + + +@pytest.mark.asyncio +async def test_run_pipeline_skips_page_snapshot_when_threat_db_short_circuits( + async_session: AsyncSession, +) -> None: + final_url = "https://evil.test/" + + with ( + patch("app.services.pipeline.normalize_url") as mock_norm, + patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain, + patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat, + patch( + "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock + ) as mock_heuristic, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), + patch( + "app.services.pipeline.capture_page_snapshot", new_callable=AsyncMock + ) as mock_snapshot, + ): + mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) + mock_unchain.return_value = _make_unchain(final_url) + mock_threat.return_value = _malicious_threat(final_url) + mock_heuristic.return_value = _heuristic_with_score(20) + + result = await run_pipeline("aid-skip-snapshot", final_url, async_session) + + assert isinstance(result, PipelineSuccess) + assert result.snapshot is not None + assert result.snapshot.status == PageSnapshotStatus.SKIPPED + assert result.snapshot.error == "skipped_already_danger" + mock_snapshot.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_run_pipeline_times_out_slow_page_snapshot_without_delaying_verdict( + async_session: AsyncSession, + monkeypatch: pytest.MonkeyPatch, +) -> None: + final_url = "https://slow-snapshot.test/" + monkeypatch.setattr(settings, "page_snapshot_timeout_seconds", 0.01) + + async def slow_snapshot(_analysis_id: str, _url: str) -> PageSnapshotResult: + await asyncio.sleep(5.0) + return PageSnapshotResult( + status=PageSnapshotStatus.AVAILABLE, + final_url=final_url, + storage_key="page-snapshots/too-late.png", + ) + + with ( + patch("app.services.pipeline.normalize_url") as mock_norm, + patch("app.services.pipeline.unchain_url", new_callable=AsyncMock) as mock_unchain, + patch("app.services.pipeline.check_threat_db", new_callable=AsyncMock) as mock_threat, + patch( + "app.services.pipeline.check_domain_heuristic", new_callable=AsyncMock + ) as mock_heuristic, + patch("app.services.pipeline.analyze_content", new_callable=AsyncMock) as mock_content, + patch("app.services.pipeline.capture_page_snapshot", side_effect=slow_snapshot), + ): + mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) + mock_unchain.return_value = _make_unchain(final_url) + mock_threat.return_value = _make_threat(final_url) + mock_heuristic.return_value = _make_heuristic("slow-snapshot.test") + mock_content.return_value = _make_content(final_url) + + started = time.perf_counter() + result = await run_pipeline("aid-slow-snapshot", final_url, async_session) + elapsed = time.perf_counter() - started + + assert isinstance(result, PipelineSuccess) + assert result.verdict == Verdict.SAFE + assert result.snapshot is not None + assert result.snapshot.status == PageSnapshotStatus.TIMEOUT + assert result.snapshot.elapsed_seconds is not None + assert elapsed < 1.0 + + @pytest.mark.asyncio async def test_run_pipeline_normalize_failure_skips_heuristic( async_session: AsyncSession, @@ -676,9 +822,7 @@ async def fast_heuristic(_url: str) -> DomainHeuristicResult: patch("app.services.pipeline.check_domain_heuristic", side_effect=fast_heuristic), patch("app.services.pipeline.analyze_content", new_callable=AsyncMock), ): - mock_norm.return_value = NormalizeResult( - original_url=final_url, normalized_url=final_url - ) + mock_norm.return_value = NormalizeResult(original_url=final_url, normalized_url=final_url) mock_unchain.return_value = _make_unchain(final_url) result = await run_pipeline("aid-race", final_url, async_session)