chore: complete test file migration (delete old, add new paths)
This commit is contained in:
chiguyong
parent
c253ccd794
commit
bdf351977b
|
|
@ -1,54 +0,0 @@
|
|||
# test_html_generator.py
|
||||
import pytest
|
||||
|
||||
# 使用实际实现的 HTMLGenerator
|
||||
from app.services.content.html_generator import HTMLGenerator
|
||||
|
||||
def test_filter_banned_tags_zhihu():
|
||||
"""知乎HTML标签过滤"""
|
||||
generator = HTMLGenerator()
|
||||
html = generator.generate(
|
||||
content="<script>alert(1)</script><p>这是内容</p>",
|
||||
platform="zhihu"
|
||||
)
|
||||
assert "<script>" not in html
|
||||
assert "<p>这是内容</p>" in html
|
||||
|
||||
def test_filter_banned_tags_wechat():
|
||||
"""微信公众号HTML过滤"""
|
||||
generator = HTMLGenerator()
|
||||
html = generator.generate(
|
||||
content="<a href='http://baidu.com'>外部链接</a><p>内容</p>",
|
||||
platform="wechat"
|
||||
)
|
||||
# 微信公众号禁止外部链接
|
||||
assert "http://baidu.com" not in html
|
||||
|
||||
def test_convert_to_markdown():
|
||||
"""Markdown转换"""
|
||||
generator = HTMLGenerator()
|
||||
md = generator.to_markdown("<h1>标题</h1><p>段落</p>")
|
||||
assert "# 标题" in md
|
||||
assert "段落" in md
|
||||
|
||||
def test_convert_to_plain():
|
||||
"""纯文本转换"""
|
||||
generator = HTMLGenerator()
|
||||
plain = generator.to_plain("<h1>标题</h1><p>段落<b>加粗</b></p>")
|
||||
assert "标题" in plain
|
||||
assert "段落" in plain
|
||||
assert "<" not in plain # 不应包含HTML标签
|
||||
|
||||
def test_multi_format_output():
|
||||
"""多格式同时输出"""
|
||||
generator = HTMLGenerator()
|
||||
html = generator.generate("<p>内容</p>", "zhihu", "html")
|
||||
md = generator.to_markdown("<p>内容</p>")
|
||||
plain = generator.to_plain("<p>内容</p>")
|
||||
|
||||
assert html is not None
|
||||
assert md is not None
|
||||
assert plain is not None
|
||||
assert len(html) > 0
|
||||
assert len(md) > 0
|
||||
assert len(plain) > 0
|
||||
|
|
@ -0,0 +1,181 @@
|
|||
"""限流中间件测试 — 覆盖内存后端和 Redis 后端降级场景。"""
|
||||
import asyncio
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from app.middleware.rate_limit import (
|
||||
MemoryRateLimitBackend,
|
||||
RateLimitBackend,
|
||||
RedisRateLimitBackend,
|
||||
_create_backend,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MemoryRateLimitBackend 测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestMemoryRateLimitBackend:
|
||||
"""内存后端核心逻辑测试。"""
|
||||
|
||||
def setup_method(self):
|
||||
self.backend = MemoryRateLimitBackend()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_under_limit_not_blocked(self):
|
||||
"""未达限流阈值时不被限流。"""
|
||||
now = time.time()
|
||||
for i in range(5):
|
||||
result = await self.backend.is_rate_limited("test:key", now + i * 0.001, 5, 60)
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_at_limit_blocked(self):
|
||||
"""达到限流阈值后被限流。"""
|
||||
now = time.time()
|
||||
# 先发送 max_requests 个请求
|
||||
for i in range(5):
|
||||
await self.backend.is_rate_limited("test:key", now + i * 0.001, 5, 60)
|
||||
# 第 6 个请求应被限流
|
||||
result = await self.backend.is_rate_limited("test:key", now + 0.006, 5, 60)
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_window_expiry_allows_new_requests(self):
|
||||
"""窗口过期后允许新请求。"""
|
||||
now = time.time()
|
||||
# 在窗口开始时发送 5 个请求
|
||||
for i in range(5):
|
||||
await self.backend.is_rate_limited("test:key", now + i * 0.001, 5, 60)
|
||||
# 窗口过期后发送新请求
|
||||
new_now = now + 61 # 超过 60 秒窗口
|
||||
result = await self.backend.is_rate_limited("test:key", new_now, 5, 60)
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_different_keys_independent(self):
|
||||
"""不同 key 的限流状态独立。"""
|
||||
now = time.time()
|
||||
# key1 达到限流
|
||||
for i in range(5):
|
||||
await self.backend.is_rate_limited("key1", now + i * 0.001, 5, 60)
|
||||
result1 = await self.backend.is_rate_limited("key1", now + 0.006, 5, 60)
|
||||
assert result1 is True
|
||||
# key2 不受限流影响
|
||||
result2 = await self.backend.is_rate_limited("key2", now, 5, 60)
|
||||
assert result2 is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reset_clears_key(self):
|
||||
"""reset 清除指定 key 的限流状态。"""
|
||||
now = time.time()
|
||||
for i in range(5):
|
||||
await self.backend.is_rate_limited("test:key", now + i * 0.001, 5, 60)
|
||||
# 限流中
|
||||
result = await self.backend.is_rate_limited("test:key", now + 0.006, 5, 60)
|
||||
assert result is True
|
||||
# 重置后不再限流
|
||||
await self.backend.reset("test:key")
|
||||
result = await self.backend.is_rate_limited("test:key", now + 0.007, 5, 60)
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cleanup_removes_expired_entries(self):
|
||||
"""后台清理任务移除过期记录。"""
|
||||
now = time.time()
|
||||
# 添加一些记录
|
||||
await self.backend.is_rate_limited("old_key", now - 3700, 5, 60)
|
||||
await self.backend.is_rate_limited("recent_key", now, 5, 60)
|
||||
# 模拟清理
|
||||
self.backend._requests["old_key"] = [now - 3700]
|
||||
self.backend._requests["recent_key"] = [now]
|
||||
# 手动触发清理逻辑
|
||||
expired_keys = []
|
||||
for key in list(self.backend._requests.keys()):
|
||||
self.backend._requests[key] = [t for t in self.backend._requests[key] if now - t < 3600]
|
||||
if not self.backend._requests[key]:
|
||||
expired_keys.append(key)
|
||||
for key in expired_keys:
|
||||
del self.backend._requests[key]
|
||||
assert "old_key" not in self.backend._requests
|
||||
assert "recent_key" in self.backend._requests
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RedisRateLimitBackend 测试（无 Redis 连接时的降级行为）
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRedisRateLimitBackendFallback:
|
||||
"""Redis 后端在连接失败时的降级行为。"""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_redis_unavailable_allows_request(self):
|
||||
"""Redis 不可用时放行请求（不阻塞服务）。"""
|
||||
backend = RedisRateLimitBackend("redis://nonexistent:6379/0")
|
||||
now = time.time()
|
||||
# Redis 连接失败时应返回 False（不被限流）
|
||||
result = await backend.is_rate_limited("test:key", now, 5, 60)
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_redis_reset_silently_fails(self):
|
||||
"""Redis reset 失败时静默处理。"""
|
||||
backend = RedisRateLimitBackend("redis://nonexistent:6379/0")
|
||||
# 不应抛出异常
|
||||
await backend.reset("test:key")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Backend 工厂函数测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestCreateBackend:
|
||||
"""_create_backend 工厂函数测试。"""
|
||||
|
||||
def test_default_creates_memory_backend(self, monkeypatch):
|
||||
"""默认配置创建内存后端。"""
|
||||
monkeypatch.setattr("app.middleware.rate_limit.settings.RATE_LIMIT_BACKEND", "memory")
|
||||
backend = _create_backend()
|
||||
assert isinstance(backend, MemoryRateLimitBackend)
|
||||
|
||||
def test_redis_with_empty_url_falls_back_to_memory(self, monkeypatch):
|
||||
"""RATE_LIMIT_BACKEND=redis 但 REDIS_URL 为空时降级到内存。"""
|
||||
monkeypatch.setattr("app.middleware.rate_limit.settings.RATE_LIMIT_BACKEND", "redis")
|
||||
monkeypatch.setattr("app.middleware.rate_limit.settings.REDIS_URL", "")
|
||||
backend = _create_backend()
|
||||
assert isinstance(backend, MemoryRateLimitBackend)
|
||||
|
||||
def test_redis_with_url_creates_redis_backend(self, monkeypatch):
|
||||
"""RATE_LIMIT_BACKEND=redis 且 REDIS_URL 非空时创建 Redis 后端。"""
|
||||
monkeypatch.setattr("app.middleware.rate_limit.settings.RATE_LIMIT_BACKEND", "redis")
|
||||
monkeypatch.setattr("app.middleware.rate_limit.settings.REDIS_URL", "redis://localhost:6379/0")
|
||||
backend = _create_backend()
|
||||
assert isinstance(backend, RedisRateLimitBackend)
|
||||
|
||||
def test_unknown_backend_defaults_to_memory(self, monkeypatch):
|
||||
"""未知后端类型默认使用内存。"""
|
||||
monkeypatch.setattr("app.middleware.rate_limit.settings.RATE_LIMIT_BACKEND", "unknown")
|
||||
backend = _create_backend()
|
||||
assert isinstance(backend, MemoryRateLimitBackend)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RateLimitBackend 接口测试
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestRateLimitBackendInterface:
|
||||
"""验证 RateLimitBackend 抽象接口。"""
|
||||
|
||||
def test_cannot_instantiate_abstract_class(self):
|
||||
"""不能直接实例化抽象类。"""
|
||||
with pytest.raises(TypeError):
|
||||
RateLimitBackend()
|
||||
|
||||
def test_memory_backend_is_subclass(self):
|
||||
"""MemoryRateLimitBackend 是 RateLimitBackend 的子类。"""
|
||||
assert issubclass(MemoryRateLimitBackend, RateLimitBackend)
|
||||
|
||||
def test_redis_backend_is_subclass(self):
|
||||
"""RedisRateLimitBackend 是 RateLimitBackend 的子类。"""
|
||||
assert issubclass(RedisRateLimitBackend, RateLimitBackend)
|
||||
|
|
@ -1,105 +0,0 @@
|
|||
# test_rule_validator.py
|
||||
import pytest
|
||||
from app.services.distribution.platform_rules import PLATFORM_RULES
|
||||
from app.services.content.rule_validator import RuleValidator, ValidationIssue, ValidationResult, AI_Pattern
|
||||
|
||||
def test_validate_title_length_pass():
|
||||
"""标题长度符合规则时返回passed"""
|
||||
validator = RuleValidator()
|
||||
result = validator.validate(
|
||||
content="这是一篇关于AI医疗的深度分析文章...",
|
||||
title="AI医疗的发展趋势与未来展望", # 符合知乎10-30要求
|
||||
platform="zhihu"
|
||||
)
|
||||
assert result.is_valid == True
|
||||
assert any("标题长度合规" in p or "合规" in p for p in result.passed)
|
||||
|
||||
def test_validate_title_length_fail():
|
||||
"""标题长度超出限制时返回issue"""
|
||||
validator = RuleValidator()
|
||||
result = validator.validate(
|
||||
content="内容",
|
||||
title="这个标题太长了超过了三十个字符的限制了哈哈哈哈哈哈", # 超过微信公众号22字限制
|
||||
platform="wechat" # 微信公众号限制22字
|
||||
)
|
||||
assert result.is_valid == False
|
||||
assert any("超过" in i.message for i in result.issues if i.severity == "high")
|
||||
|
||||
def test_validate_content_length_pass():
|
||||
"""内容长度符合规则时返回passed"""
|
||||
validator = RuleValidator()
|
||||
result = validator.validate(
|
||||
content="A" * 1500, # 1500字，符合知乎500-50000要求
|
||||
title="测试标题",
|
||||
platform="zhihu"
|
||||
)
|
||||
assert result.score >= 80
|
||||
|
||||
def test_validate_content_length_fail():
|
||||
"""内容超长返回issue"""
|
||||
validator = RuleValidator()
|
||||
result = validator.validate(
|
||||
content="A" * 30000, # 30000字，微信公众号限制20000
|
||||
title="测试标题",
|
||||
platform="wechat"
|
||||
)
|
||||
assert any("超过" in i.message for i in result.issues if i.severity == "high")
|
||||
|
||||
def test_detect_ai_patterns_banned_words():
|
||||
"""检测禁用词"""
|
||||
validator = RuleValidator()
|
||||
result = validator.detect_ai_patterns(
|
||||
content="首先，其次，最后，总而言之，总之，总之",
|
||||
platform="zhihu"
|
||||
)
|
||||
assert len(result) > 0
|
||||
assert any("首先" in r.pattern or "总之" in r.pattern for r in result)
|
||||
|
||||
def test_detect_ai_patterns_banned_structures():
|
||||
"""检测禁用结构"""
|
||||
validator = RuleValidator()
|
||||
result = validator.detect_ai_patterns(
|
||||
content="第一，观点一。第二，观点二。第三，观点三。",
|
||||
platform="zhihu"
|
||||
)
|
||||
assert len(result) > 0
|
||||
|
||||
def test_validate_zhihu_specific_rules():
|
||||
"""知乎特定规则"""
|
||||
validator = RuleValidator()
|
||||
result = validator.validate(
|
||||
content="这是一个专业回答",
|
||||
title="专业回答",
|
||||
platform="zhihu"
|
||||
)
|
||||
# 知乎应检查营销用语
|
||||
assert result.score > 0
|
||||
|
||||
def test_validate_wechat_specific_rules():
|
||||
"""微信公众号特定规则"""
|
||||
validator = RuleValidator()
|
||||
result = validator.validate(
|
||||
content="点击购买，限时优惠",
|
||||
title="限时优惠",
|
||||
platform="wechat"
|
||||
)
|
||||
# 微信公众号应检测诱导分享
|
||||
assert any("诱导" in i.message or "营销" in i.message for i in result.issues)
|
||||
|
||||
def test_validate_xiaohongshu_rules():
|
||||
"""小红书特定规则"""
|
||||
validator = RuleValidator()
|
||||
result = validator.validate(
|
||||
content="微信公众号搜索xxx获取更多内容",
|
||||
title="种草笔记",
|
||||
platform="xiaohongshu"
|
||||
)
|
||||
# 小红书应检测跨平台引流
|
||||
assert any("引流" in i.message or "平台" in i.message for i in result.issues)
|
||||
|
||||
def test_get_optimization_tips():
|
||||
"""获取优化建议"""
|
||||
validator = RuleValidator()
|
||||
tips = validator.get_optimization_tips("zhihu")
|
||||
assert len(tips) > 0
|
||||
assert any(isinstance(tip, str) for tip in tips)
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
import pytest
|
||||
|
||||
# 导入实际的 SensitiveFilter 实现
|
||||
from app.services.content.sensitive_filter import SensitiveFilter
|
||||
|
||||
|
||||
def test_filter_politics_words():
|
||||
"""政治敏感词被替换为占位符"""
|
||||
filter = SensitiveFilter()
|
||||
result = filter.filter(
|
||||
content="这是一个关于台湾问题的分析",
|
||||
platform="zhihu"
|
||||
)
|
||||
assert "**" in result.filtered_content
|
||||
assert len(result.found_words) > 0
|
||||
assert result.found_words[0].category == "politics"
|
||||
|
||||
def test_filter_medical_words():
|
||||
"""医疗敏感词处理"""
|
||||
filter = SensitiveFilter()
|
||||
result = filter.filter(
|
||||
content="这个药品效果很好",
|
||||
platform="wechat"
|
||||
)
|
||||
# 医疗类敏感词应被检测
|
||||
assert result.found_words is not None
|
||||
|
||||
def test_filter_finance_words():
|
||||
"""金融敏感词处理"""
|
||||
filter = SensitiveFilter()
|
||||
result = filter.filter(
|
||||
content="年化收益率10%",
|
||||
platform="zhihu"
|
||||
)
|
||||
# 金融敏感词检测
|
||||
assert result.found_words is not None
|
||||
|
||||
def test_filter_multiple_categories():
|
||||
"""多分类同时过滤"""
|
||||
filter = SensitiveFilter()
|
||||
result = filter.filter(
|
||||
content="这是内容包含政治和医疗敏感词的内容",
|
||||
platform="wechat"
|
||||
)
|
||||
categories = [w.category for w in result.found_words]
|
||||
assert len(set(categories)) >= 1 # 至少检测到一个分类
|
||||
|
||||
def test_add_custom_words():
|
||||
"""自定义敏感词添加"""
|
||||
filter = SensitiveFilter()
|
||||
filter.add_custom_words("custom", ["敏感词1", "敏感词2"])
|
||||
|
||||
result = filter.filter(
|
||||
content="这是一段包含敏感词1的内容",
|
||||
platform="zhihu"
|
||||
)
|
||||
assert "敏感词1" not in result.filtered_content
|
||||
|
|
@ -1,61 +0,0 @@
|
|||
# test_seo_optimizer.py
|
||||
import pytest
|
||||
|
||||
# 导入实际实现的 SEOOptimizer
|
||||
from app.services.content.seo_optimizer import SEOOptimizer
|
||||
|
||||
def test_get_keyword_density():
|
||||
"""关键词密度计算"""
|
||||
optimizer = SEOOptimizer()
|
||||
content = "AI医疗AI医疗AI医疗" # 5个字，AI医疗出现3次
|
||||
density = optimizer.get_keyword_density(content, "AI医疗")
|
||||
# 密度计算：(3 * 4) / 15 ≈ 0.8 (约80%)
|
||||
assert density > 0
|
||||
|
||||
def test_adjust_keyword_density():
|
||||
"""密度调整到推荐范围"""
|
||||
optimizer = SEOOptimizer()
|
||||
result = optimizer.optimize(
|
||||
content="AI医疗是未来发展趋势。随着人工智能技术的不断进步，医疗领域正在经历智能化变革。智能诊断系统能够分析海量医学数据，为医生提供辅助决策支持，提高诊疗效率和准确性，改善患者就医体验，推动医疗资源的优化配置和行业升级，促进整个医疗生态的可持续发展，提升医疗服务质量与管理水平。",
|
||||
title="AI医疗",
|
||||
platform="zhihu", # 推荐密度 1-3%
|
||||
keyword="AI医疗"
|
||||
)
|
||||
# 优化后密度应在推荐范围内
|
||||
assert result.density >= 1.0
|
||||
assert result.density <= 3.0
|
||||
|
||||
def test_optimize_keyword_position():
|
||||
"""关键词位置优化"""
|
||||
optimizer = SEOOptimizer()
|
||||
result = optimizer.optimize(
|
||||
content="这是一篇关于人工智能医疗的文章",
|
||||
title="文章标题",
|
||||
platform="zhihu",
|
||||
keyword="AI医疗"
|
||||
)
|
||||
# 应建议在标题中添加关键词
|
||||
assert result.suggestions is not None
|
||||
assert len(result.suggestions) > 0
|
||||
|
||||
def test_optimize_multiple_keywords():
|
||||
"""多关键词处理"""
|
||||
optimizer = SEOOptimizer()
|
||||
result = optimizer.optimize(
|
||||
content="人工智能和机器学习是热门技术",
|
||||
title="技术文章",
|
||||
platform="zhihu",
|
||||
keyword="人工智能"
|
||||
)
|
||||
assert result.optimized_content is not None
|
||||
|
||||
def test_seo_tips_generation():
|
||||
"""SEO建议生成"""
|
||||
optimizer = SEOOptimizer()
|
||||
result = optimizer.optimize(
|
||||
content="内容",
|
||||
title="标题",
|
||||
platform="zhihu"
|
||||
)
|
||||
assert result.tips is not None
|
||||
assert len(result.tips) > 0
|
||||
|
|
@ -91,6 +91,61 @@ class TestPlatformRuleEngine:
|
|||
tips = engine.get_optimization_tips("unknown_xyz")
|
||||
assert tips == []
|
||||
|
||||
# --- 标签验证测试 ---
|
||||
|
||||
def test_validate_tags_within_range(self, engine):
|
||||
"""标签数量在范围内验证通过"""
|
||||
title = "合规标题测试"
|
||||
content = "这是符合要求的内容。" * 50
|
||||
result = engine.validate_content(content, title, "wechat", tags=["标签1", "标签2", "标签3"])
|
||||
|
||||
tag_issues = [i for i in result["issues"] if i.get("category") == "tag_count"]
|
||||
assert len(tag_issues) == 0
|
||||
|
||||
def test_validate_tags_below_minimum(self, engine):
|
||||
"""标签数量低于最低要求返回 medium issue"""
|
||||
title = "合规标题测试"
|
||||
content = "这是符合要求的内容。" * 50
|
||||
result = engine.validate_content(content, title, "wechat", tags=[])
|
||||
|
||||
tag_issues = [i for i in result["issues"] if i.get("category") == "tag_count"]
|
||||
assert len(tag_issues) > 0
|
||||
assert tag_issues[0]["severity"] == "medium"
|
||||
assert "低于最低要求" in tag_issues[0]["message"]
|
||||
|
||||
def test_validate_tags_exceed_maximum(self, engine):
|
||||
"""标签数量超过限制返回 high issue"""
|
||||
title = "合规标题测试"
|
||||
content = "这是符合要求的内容。" * 50
|
||||
# 微信 max_tags=10，传入 15 个
|
||||
many_tags = [f"标签{i}" for i in range(15)]
|
||||
result = engine.validate_content(content, title, "wechat", tags=many_tags)
|
||||
|
||||
tag_issues = [i for i in result["issues"] if i.get("category") == "tag_count"]
|
||||
assert len(tag_issues) > 0
|
||||
assert tag_issues[0]["severity"] == "high"
|
||||
assert "超过限制" in tag_issues[0]["message"]
|
||||
|
||||
def test_validate_tags_as_comma_separated_string(self, engine):
|
||||
"""标签为逗号分隔字符串时正确解析"""
|
||||
title = "合规标题测试"
|
||||
content = "这是符合要求的内容。" * 50
|
||||
result = engine.validate_content(content, title, "wechat", tags="标签1,标签2,标签3")
|
||||
|
||||
tag_issues = [i for i in result["issues"] if i.get("category") == "tag_count"]
|
||||
assert len(tag_issues) == 0
|
||||
|
||||
def test_validate_no_tags_treated_as_zero(self, engine):
|
||||
"""无标签时视为 0 个标签，低于 min_tags 时触发 issue"""
|
||||
title = "合规标题测试"
|
||||
content = "这是符合要求的内容。" * 50
|
||||
result = engine.validate_content(content, title, "wechat", tags=None)
|
||||
|
||||
tag_issues = [i for i in result["issues"] if i.get("category") == "tag_count"]
|
||||
# 微信 min_tags=3，tags=None 时 tag_count=0 < 3，触发 medium issue
|
||||
assert len(tag_issues) > 0
|
||||
assert tag_issues[0]["severity"] == "medium"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ContentFormatter 测试
|
||||
|
|
|
|||
Loading…
Reference in New Issue