"""新闻格式限制、内容稳定性及历史哈希兼容。"""
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from scrapy import Request
from scrapy.exceptions import IgnoreRequest, StopDownload
from scrapy.http import Headers
from alphapulse.crawler.content_policy import clean_html, ignored_retry
@pytest.fixture
def guard(trading):
from alphapulse.crawler.middlewares import NewsContentPolicyMiddleware
return NewsContentPolicyMiddleware()
def test_body_images_tables_and_real_changes_survive_normalization():
html = '
政策正文\x00
3%
图表说明'
value = clean_html(html, 'https://official.example/news/1')
assert '政策正文' in value and '3%' in value and '图表说明' in value
assert 'https://official.example/chart.png' in value
assert 'https://official.example/a.png 1x' in value
assert 'iframe' not in value and 'script' not in value and '\x00' not in value
assert clean_html(value, 'https://official.example/news/1') == value
assert clean_html(html.replace('3%', '4%')) != clean_html(html)
def test_email_share_encoding_is_not_news_content():
html = '
Monetary policy.
Email'
assert clean_html(html) == clean_html(html.replace('abc', 'def'))
def test_picture_sources_and_lazy_candidates_resolve_to_publisher():
html = (''
''
'')
base = 'https://official.example/news/article'
value = clean_html(html, base)
from parsel import Selector
sel = Selector(value)
assert sel.css('picture source::attr(srcset)').getall() == [
'https://official.example/wide.webp 1x, https://official.example/large.webp 2x',
'https://cdn.example/chart.webp 800w',
]
assert sel.css('img::attr(src)').get() == 'https://official.example/fallback.png'
assert sel.css('img::attr(alt)').get() == '政策图表'
assert clean_html(value, base) == value
@pytest.mark.parametrize('url', ['https://official.example/a.PDF?download=1', 'https://official.example/a.docx', 'https://official.example/exit?https://other.example/a.pdf'])
def test_file_urls_are_rejected_before_download(url, guard):
request = Request(url)
with pytest.raises(IgnoreRequest):
guard.process_request(request)
assert request.meta['policy_skipped']
def test_unknown_file_and_split_magic_stop_without_a_complete_body(guard):
req = Request('https://official.example/download')
with pytest.raises(StopDownload):
guard.headers_received(Headers({'Content-Type': 'application/pdf'}), req)
req = Request('https://official.example/mislabeled')
guard.bytes_received(b'%P', req)
with pytest.raises(StopDownload):
guard.bytes_received(b'DF-1.7', req)
assert req.meta['policy_skipped']
def test_legacy_attachment_request_is_rejected_even_without_file_extension(guard):
request = Request('https://official.example/media/1/download', meta={'parsed': {'title': 'old'}})
with pytest.raises(IgnoreRequest):
guard.process_request(request)
assert ignored_retry('attachment_retries', {'attempts': 3})
assert not ignored_retry('document_retries', {'document': {'url': 'https://official.example/news/1'}})
def test_normalization_preserves_existing_hash_and_sends_only_real_updates(trading, monkeypatch):
from alphapulse.crawler import pipelines
html = '
