Compare commits

...

3 Commits

Author SHA1 Message Date
CI Bot c245b5bcd2 style: auto-format with black + isort + prettier [skip ci-format-check] 2026-07-28 08:27:06 +00:00
xiaoxia 6586a8b943 test(wave140): watermark_config domain 层 +81 单测
CI/CD Pipeline / Staging E2E Tests (pull_request) Blocked by required conditions
CI/CD Pipeline / Staging API Integration Tests (pull_request) Blocked by required conditions
CI/CD Pipeline / Deploy Production (pull_request) Blocked by required conditions
CI/CD Pipeline / Production Browser E2E (pull_request) Blocked by required conditions
CI/CD Pipeline / ACR Image Cleanup (pull_request) Blocked by required conditions
CI/CD Pipeline / Canary Release to Production (pull_request) Blocked by required conditions
Preview Cleanup / Cleanup Preview Environment (pull_request) Waiting to run
ACR Cleanup / ACR Image Cleanup (pull_request_target) Waiting to run
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 26s
CI/CD Pipeline / Unit Tests (pull_request) Waiting to run
CI/CD Pipeline / Frontend Unit Tests (pull_request) Waiting to run
CI/CD Pipeline / Build Production API Image (pull_request) Waiting to run
CI/CD Pipeline / Build Production Web Image (pull_request) Waiting to run
CI/CD Pipeline / Build Production Worker Image (pull_request) Waiting to run
CI/CD Pipeline / CI Gate (pull_request) Blocked by required conditions
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Waiting to run
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m2s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m35s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 2m36s
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 2m59s
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 3m1s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 3m41s
CI/CD Pipeline / Integration Tests (pull_request) Waiting to run
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m28s
AI Code Review / AI Code Review (pull_request) Successful in 2m6s
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 5m46s
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m48s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 46m43s
为 domain/watermark_config.py 新增 81 个单测,0 FFmpeg 依赖:

- 常量与配置:6 个
- from_dict 配置解析:10 个
- validate 配置校验:14 个
- has_effect 有效性判断:8 个
- calc_position 9宫格位置计算:13 个
- calc_scroll_x 滚动表达式:5 个
- build_image_watermark_filter 图片水印滤镜:8 个
- build_text_watermark_filter 文字水印滤镜:10 个
- 位置工具函数:5 个

覆盖:9宫格全部位置、透明度边界、缩放边界、
文字转义(冒号/单引号)、滚动水印、字体路径、
无效配置降级、空配置返回 None 等。
2026-07-28 16:06:03 +08:00
xiaoxia 834d9f219e test(wave139): url_security domain 层 +86 单测
为 domain/url_security.py 新增 86 个单测,SSRF 安全校验全覆盖:

- 常量与异常类:6 个
- check_internal_hostname 内部主机名拦截:12 个
- is_trusted_domain 可信域名匹配:7 个
- check_ssrf_ip IP SSRF 检查:9 个
- is_ip_address IP 判断:5 个
- validate_url_basic URL 基础校验:16 个
- is_url_basic_safe 便捷函数:5 个
- validate_magic_number 魔数校验:20 个

覆盖场景:
- 回环/私有/链路本地/组播/未指定/保留 IP 拦截
- .local/.internal/.localdomain 内网域名后缀拦截
- localhost/metadata 等敏感主机名拦截
- scheme 白名单 / 端口白名单 / URL 长度限制
- 直接 IP 访问开关 + 可信域名白名单(含子域名)
- 9 种文件格式魔数校验(图片/音频/视频)
- 空字节 / 过短 / 未知 MIME 等边界情况
2026-07-28 16:06:03 +08:00
6 changed files with 1269 additions and 4 deletions
@@ -6,7 +6,6 @@
from __future__ import annotations
# ── 单轨时间计算 ──────────────────────────────────────────────────────────────
@@ -13,7 +13,6 @@
from __future__ import annotations
from packages.domain.speed_config import (
DEFAULT_SPEED,
SpeedConfig,
+567
View File
@@ -0,0 +1,567 @@
"""url_security 单测.
domain 层 URL 安全校验纯逻辑模块,0 网络依赖。
覆盖 SSRF 防护、主机名校验、IP 检查、魔数校验等。
"""
from __future__ import annotations
import pytest
from packages.domain.url_security import (
ALLOWED_IMAGE_MIME_TYPES,
ALLOWED_PORTS,
ALLOWED_SCHEMES,
ALLOWED_VIDEO_MIME_TYPES,
MAGIC_NUMBERS,
MAX_URL_LENGTH,
UrlSecurityError,
check_internal_hostname,
check_ssrf_ip,
is_ip_address,
is_trusted_domain,
is_url_basic_safe,
validate_magic_number,
validate_url_basic,
)
# ═══════════════════════════════════════════════════════════════════════════════
# 常量与异常类
# ═══════════════════════════════════════════════════════════════════════════════
class TestConstants:
"""常量测试."""
def test_allowed_schemes(self):
"""允许的 scheme 包含 http 和 https."""
assert "http" in ALLOWED_SCHEMES
assert "https" in ALLOWED_SCHEMES
def test_allowed_ports(self):
"""允许的端口:80, 443."""
assert 80 in ALLOWED_PORTS
assert 443 in ALLOWED_PORTS
def test_max_url_length(self):
"""最大 URL 长度 2048."""
assert MAX_URL_LENGTH == 2048
def test_magic_numbers_has_common_formats(self):
"""魔数表包含常见格式."""
assert "image/jpeg" in MAGIC_NUMBERS
assert "image/png" in MAGIC_NUMBERS
assert "image/gif" in MAGIC_NUMBERS
assert "video/mp4" in MAGIC_NUMBERS
assert "audio/mpeg" in MAGIC_NUMBERS
class TestUrlSecurityError:
"""异常类测试."""
def test_is_value_error(self):
"""UrlSecurityError 继承 ValueError."""
assert issubclass(UrlSecurityError, ValueError)
def test_raise_with_message(self):
"""抛出时携带错误信息."""
with pytest.raises(UrlSecurityError, match="test error"):
raise UrlSecurityError("test error")
# ═══════════════════════════════════════════════════════════════════════════════
# check_internal_hostname
# ═══════════════════════════════════════════════════════════════════════════════
class TestCheckInternalHostname:
"""内部主机名检查测试."""
def test_normal_domain_passes(self):
"""普通外部域名通过."""
check_internal_hostname("example.com")
check_internal_hostname("www.google.com")
def test_localhost_blocked(self):
"""localhost 被拦截."""
with pytest.raises(UrlSecurityError, match="内部主机名"):
check_internal_hostname("localhost")
def test_localhost_case_insensitive(self):
"""大小写不敏感."""
with pytest.raises(UrlSecurityError):
check_internal_hostname("LOCALHOST")
with pytest.raises(UrlSecurityError):
check_internal_hostname("LocalHost")
def test_localhost_localdomain_blocked(self):
"""localhost.localdomain 被拦截."""
with pytest.raises(UrlSecurityError):
check_internal_hostname("localhost.localdomain")
def test_metadata_blocked(self):
"""metadata 被拦截."""
with pytest.raises(UrlSecurityError):
check_internal_hostname("metadata")
def test_metadata_google_internal_blocked(self):
"""GCP 元数据服务被拦截."""
with pytest.raises(UrlSecurityError):
check_internal_hostname("metadata.google.internal")
def test_cloud_metadata_ip_blocked(self):
"""云元数据 IP 169.254.169.254 被拦截."""
with pytest.raises(UrlSecurityError):
check_internal_hostname("169.254.169.254")
def test_local_suffix_blocked(self):
""".local 后缀域名被拦截."""
with pytest.raises(UrlSecurityError, match="内网域名"):
check_internal_hostname("myhost.local")
def test_internal_suffix_blocked(self):
""".internal 后缀被拦截."""
with pytest.raises(UrlSecurityError):
check_internal_hostname("svc.cluster.internal")
def test_localdomain_suffix_blocked(self):
""".localdomain 后缀被拦截."""
with pytest.raises(UrlSecurityError):
check_internal_hostname("host.localdomain")
def test_com_domain_not_blocked(self):
""".com 域名不被拦截."""
check_internal_hostname("example.com")
check_internal_hostname("sub.example.com")
def test_subdomain_of_public_domain_ok(self):
"""公网域名的子域名正常."""
check_internal_hostname("api.example.com")
check_internal_hostname("cdn.assets.example.org")
# ═══════════════════════════════════════════════════════════════════════════════
# is_trusted_domain
# ═══════════════════════════════════════════════════════════════════════════════
class TestIsTrustedDomain:
"""可信域名匹配测试."""
def test_empty_trusted_domains_allows_all(self):
"""空集合允许所有域名."""
assert is_trusted_domain("anything.com", set()) is True
assert is_trusted_domain("anywhere.org", set()) is True
def test_exact_match(self):
"""精确匹配."""
trusted = {"example.com", "example.org"}
assert is_trusted_domain("example.com", trusted) is True
assert is_trusted_domain("example.org", trusted) is True
def test_subdomain_match(self):
"""子域名匹配."""
trusted = {"example.com"}
assert is_trusted_domain("api.example.com", trusted) is True
assert is_trusted_domain("cdn.assets.example.com", trusted) is True
def test_no_match(self):
"""不匹配."""
trusted = {"example.com"}
assert is_trusted_domain("other.com", trusted) is False
assert is_trusted_domain("example.net", trusted) is False
def test_case_insensitive(self):
"""大小写不敏感."""
trusted = {"Example.COM"}
assert is_trusted_domain("example.com", trusted) is True
assert is_trusted_domain("API.EXAMPLE.COM", trusted) is True
def test_partial_match_no(self):
"""域名部分相同但不是子域名不匹配."""
trusted = {"example.com"}
# fakeexample.com 不是 example.com 的子域名
assert is_trusted_domain("fakeexample.com", trusted) is False
def test_none_trusted_domains(self):
"""trusted_domains 为 None 时由调用方处理,空 set 全允许."""
# 传空集合时全允许
assert is_trusted_domain("a.com", set()) is True
# ═══════════════════════════════════════════════════════════════════════════════
# check_ssrf_ip
# ═══════════════════════════════════════════════════════════════════════════════
class TestCheckSsrIp:
"""IP SSRF 检查测试."""
def test_public_ip_passes(self):
"""公网 IP 通过."""
check_ssrf_ip("8.8.8.8")
check_ssrf_ip("1.1.1.1")
check_ssrf_ip("114.114.114.114")
def test_loopback_blocked(self):
"""回环地址被拦截."""
with pytest.raises(UrlSecurityError, match="回环"):
check_ssrf_ip("127.0.0.1")
with pytest.raises(UrlSecurityError):
check_ssrf_ip("127.0.0.53")
def test_private_ip_blocked(self):
"""私有内网 IP 被拦截."""
with pytest.raises(UrlSecurityError, match="内网"):
check_ssrf_ip("192.168.1.1")
with pytest.raises(UrlSecurityError):
check_ssrf_ip("10.0.0.1")
with pytest.raises(UrlSecurityError):
check_ssrf_ip("172.16.0.1")
def test_link_local_blocked(self):
"""链路本地地址被拦截."""
with pytest.raises(UrlSecurityError, match="链路本地"):
check_ssrf_ip("169.254.169.254")
with pytest.raises(UrlSecurityError):
check_ssrf_ip("169.254.1.1")
def test_multicast_blocked(self):
"""组播地址被拦截."""
with pytest.raises(UrlSecurityError, match="组播"):
check_ssrf_ip("224.0.0.1")
with pytest.raises(UrlSecurityError):
check_ssrf_ip("239.255.255.250")
def test_unspecified_blocked(self):
"""未指定地址被拦截."""
with pytest.raises(UrlSecurityError, match="未指定"):
check_ssrf_ip("0.0.0.0")
def test_ipv6_loopback_blocked(self):
"""IPv6 回环地址被拦截."""
with pytest.raises(UrlSecurityError):
check_ssrf_ip("::1")
def test_ipv6_private_blocked(self):
"""IPv6 内网地址被拦截."""
with pytest.raises(UrlSecurityError):
check_ssrf_ip("fc00::1")
with pytest.raises(UrlSecurityError):
check_ssrf_ip("fe80::1")
def test_ipv6_public_passes(self):
"""IPv6 公网地址通过."""
check_ssrf_ip("2001:4860:4860::8888")
def test_invalid_ip_raises_value_error(self):
"""非法 IP 抛出 ValueError(不是 UrlSecurityError."""
with pytest.raises(ValueError):
check_ssrf_ip("not-an-ip")
with pytest.raises(ValueError):
check_ssrf_ip("999.999.999.999")
# ═══════════════════════════════════════════════════════════════════════════════
# is_ip_address
# ═══════════════════════════════════════════════════════════════════════════════
class TestIsIpAddress:
"""IP 地址判断测试."""
def test_ipv4_true(self):
"""IPv4 地址返回 True."""
assert is_ip_address("127.0.0.1") is True
assert is_ip_address("8.8.8.8") is True
assert is_ip_address("0.0.0.0") is True
def test_ipv6_true(self):
"""IPv6 地址返回 True."""
assert is_ip_address("::1") is True
assert is_ip_address("2001:db8::1") is True
def test_hostname_false(self):
"""主机名返回 False."""
assert is_ip_address("example.com") is False
assert is_ip_address("localhost") is False
assert is_ip_address("sub.domain.org") is False
def test_empty_string_false(self):
"""空字符串返回 False."""
assert is_ip_address("") is False
def test_invalid_ip_false(self):
"""非法 IP 返回 False."""
assert is_ip_address("999.999.999.999") is False
assert is_ip_address("1234") is False
assert is_ip_address("abc.def") is False
# ═══════════════════════════════════════════════════════════════════════════════
# validate_url_basic
# ═══════════════════════════════════════════════════════════════════════════════
class TestValidateUrlBasic:
"""URL 基础校验测试."""
def test_normal_https_url_passes(self):
"""正常 HTTPS URL 通过."""
result = validate_url_basic("https://example.com/path")
assert result == "https://example.com/path"
def test_normal_http_url_passes(self):
"""正常 HTTP URL 通过."""
result = validate_url_basic("http://example.com/path")
assert result == "http://example.com/path"
def test_empty_url_rejected(self):
"""空 URL 被拒."""
with pytest.raises(UrlSecurityError, match="为空"):
validate_url_basic("")
def test_none_url_not_passed_as_str(self):
"""None 作为 URL(这里只测空字符串)."""
# 空字符串被拒
with pytest.raises(UrlSecurityError):
validate_url_basic("")
def test_too_long_url_rejected(self):
"""超长 URL 被拒."""
long_url = "https://example.com/" + "a" * 3000
with pytest.raises(UrlSecurityError, match="过长"):
validate_url_basic(long_url)
def test_invalid_scheme_rejected(self):
"""非法 scheme 被拒."""
with pytest.raises(UrlSecurityError, match="scheme"):
validate_url_basic("ftp://example.com/file")
with pytest.raises(UrlSecurityError):
validate_url_basic("file:///etc/passwd")
with pytest.raises(UrlSecurityError):
validate_url_basic("javascript:alert(1)")
def test_missing_hostname_rejected(self):
"""缺少主机名被拒."""
with pytest.raises(UrlSecurityError, match="主机名"):
validate_url_basic("https:///path")
def test_localhost_rejected(self):
"""localhost 被拒."""
with pytest.raises(UrlSecurityError):
validate_url_basic("https://localhost/api")
def test_internal_domain_rejected(self):
"""内网域名被拒."""
with pytest.raises(UrlSecurityError):
validate_url_basic("http://server.local/api")
def test_non_standard_port_rejected(self):
"""非标准端口被拒."""
with pytest.raises(UrlSecurityError, match="端口"):
validate_url_basic("https://example.com:8080/")
with pytest.raises(UrlSecurityError):
validate_url_basic("http://example.com:3000/")
def test_port_80_ok(self):
"""80 端口允许."""
validate_url_basic("http://example.com:80/path")
def test_port_443_ok(self):
"""443 端口允许."""
validate_url_basic("https://example.com:443/path")
def test_no_port_ok(self):
"""无端口默认允许."""
validate_url_basic("https://example.com/path")
def test_direct_ip_rejected_by_default(self):
"""默认禁止直接 IP 访问."""
with pytest.raises(UrlSecurityError, match="直接 IP"):
validate_url_basic("https://8.8.8.8/path")
def test_direct_ip_allowed_when_enabled(self):
"""allow_direct_ip=True 时允许公网 IP."""
validate_url_basic("https://8.8.8.8/path", allow_direct_ip=True)
def test_direct_ip_private_still_blocked(self):
"""即使 allow_direct_ip,内网 IP 仍被拒."""
with pytest.raises(UrlSecurityError, match="内网"):
validate_url_basic("https://192.168.1.1/", allow_direct_ip=True)
def test_direct_ip_loopback_still_blocked(self):
"""回环 IP 即使开启 direct_ip 也被拒."""
with pytest.raises(UrlSecurityError):
validate_url_basic("https://127.0.0.1/", allow_direct_ip=True)
def test_trusted_domains_pass(self):
"""可信域名列表内的域名通过."""
trusted = {"example.com", "cdn.com"}
validate_url_basic("https://api.example.com/path", trusted_domains=trusted)
validate_url_basic("https://cdn.com/asset.jpg", trusted_domains=trusted)
def test_untrusted_domain_rejected(self):
"""不在可信域名列表中的域名被拒."""
trusted = {"example.com"}
with pytest.raises(UrlSecurityError, match="白名单"):
validate_url_basic("https://evil.com/malware", trusted_domains=trusted)
def test_trusted_domain_subdomain_pass(self):
"""可信域名的子域名通过."""
trusted = {"example.com"}
validate_url_basic("https://sub.example.com/a", trusted_domains=trusted)
validate_url_basic("https://a.b.example.com/b", trusted_domains=trusted)
def test_return_value_is_original_url(self):
"""返回原始 URL 字符串."""
url = "https://example.com/path?query=value#frag"
assert validate_url_basic(url) == url
def test_metadata_ip_rejected(self):
"""云元数据 IP 被内部主机名检查拦截."""
with pytest.raises(UrlSecurityError):
validate_url_basic("http://169.254.169.254/latest/meta-data/")
# ═══════════════════════════════════════════════════════════════════════════════
# is_url_basic_safe
# ═══════════════════════════════════════════════════════════════════════════════
class TestIsUrlBasicSafe:
"""便捷函数 is_url_basic_safe 测试."""
def test_safe_url_returns_true(self):
"""安全 URL 返回 True."""
assert is_url_basic_safe("https://example.com/") is True
assert is_url_basic_safe("http://example.org/path") is True
def test_unsafe_url_returns_false(self):
"""不安全 URL 返回 False."""
assert is_url_basic_safe("https://localhost/") is False
assert is_url_basic_safe("ftp://example.com/") is False
assert is_url_basic_safe("") is False
def test_trusted_domains_param(self):
"""支持 trusted_domains 参数."""
trusted = {"example.com"}
assert is_url_basic_safe("https://other.com/", trusted_domains=trusted) is False
assert is_url_basic_safe("https://example.com/", trusted_domains=trusted) is True
def test_allow_direct_ip_param(self):
"""支持 allow_direct_ip 参数."""
assert is_url_basic_safe("https://8.8.8.8/") is False
assert is_url_basic_safe("https://8.8.8.8/", allow_direct_ip=True) is True
def test_no_exceptions_raised(self):
"""不抛出异常,只返回 bool."""
# 各种边界情况都不抛异常
try:
is_url_basic_safe("")
is_url_basic_safe("not a url")
is_url_basic_safe("http://" + "a" * 3000)
except UrlSecurityError:
pytest.fail("is_url_basic_safe should not raise UrlSecurityError")
# ═══════════════════════════════════════════════════════════════════════════════
# validate_magic_number
# ═══════════════════════════════════════════════════════════════════════════════
class TestValidateMagicNumber:
"""魔数校验测试."""
def test_jpeg_valid(self):
"""JPEG 文件通过."""
# JPEG 文件头: FF D8 FF
jpeg_header = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00"
validate_magic_number(jpeg_header, {"image/jpeg"})
def test_png_valid(self):
"""PNG 文件通过."""
png_header = b"\x89PNG\r\n\x1a\n\x00\x00\x00"
validate_magic_number(png_header, {"image/png"})
def test_gif_valid(self):
"""GIF 文件通过(GIF89a 和 GIF87a."""
validate_magic_number(b"GIF89a...", {"image/gif"})
validate_magic_number(b"GIF87a...", {"image/gif"})
def test_wav_valid(self):
"""WAV 文件通过(RIFF + WAVE."""
wav_header = b"RIFF\x00\x00\x00\x00WAVEfmt "
validate_magic_number(wav_header, {"audio/wav"})
def test_mp3_id3_valid(self):
"""带 ID3 标签的 MP3 通过."""
mp3_header = b"ID3\x03\x00\x00\x00\x00\x0f\x76"
validate_magic_number(mp3_header, {"audio/mpeg"})
def test_mp3_sync_valid(self):
"""不带 ID3 的 MP3(帧同步字)通过."""
mp3_header = b"\xff\xfb\x90\x00" + b"\x00" * 32
validate_magic_number(mp3_header, {"audio/mpeg"})
def test_ogg_valid(self):
"""OGG 文件通过."""
validate_magic_number(b"OggS\x00\x00...", {"audio/ogg"})
def test_flac_valid(self):
"""FLAC 文件通过."""
validate_magic_number(b"fLaC\x00\x00...", {"audio/flac"})
def test_webp_valid(self):
"""WebP 文件通过(RIFF + WEBP."""
webp_header = b"RIFF\x00\x00\x00\x00WEBPVP8 "
validate_magic_number(webp_header, {"image/webp"})
def test_bmp_valid(self):
"""BMP 文件通过."""
validate_magic_number(b"BM\x00\x00\x00\x00...", {"image/bmp"})
def test_mp4_valid(self):
"""MP4 文件通过(ftyp 在偏移 4)."""
mp4_header = b"\x00\x00\x00\x20ftypisom\x00\x00\x02\x00"
validate_magic_number(mp4_header, {"video/mp4"})
def test_invalid_format_rejected(self):
"""不匹配的格式被拒."""
with pytest.raises(UrlSecurityError, match="魔数"):
validate_magic_number(b"hello world", {"image/jpeg"})
def test_empty_bytes_rejected(self):
"""空字节被拒."""
with pytest.raises(UrlSecurityError, match="为空"):
validate_magic_number(b"", {"image/jpeg"})
def test_too_short_bytes_rejected(self):
"""字节太短不匹配魔数时被拒."""
with pytest.raises(UrlSecurityError):
validate_magic_number(b"\xff\xd8", {"image/jpeg"}) # 只2字节,不够JPEG魔数
def test_multiple_allowed_types(self):
"""允许多种格式时任一匹配即通过."""
jpeg_header = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00"
validate_magic_number(jpeg_header, {"image/jpeg", "image/png", "image/gif"})
def test_wrong_type_rejected(self):
"""用 PNG 魔数校验 JPEG 类型失败."""
jpeg_header = b"\xff\xd8\xff\xe0\x00\x10JFIF\x00"
with pytest.raises(UrlSecurityError):
validate_magic_number(jpeg_header, {"image/png"})
def test_unknown_mime_skipped(self):
"""未知 MIME 类型(无对应魔数)不阻断."""
# application/octet-stream 没有魔数定义,直接通过
validate_magic_number(b"random bytes here", {"application/octet-stream"})
def test_allowed_image_mime_types_has_common(self):
"""图片 MIME 白名单包含常见类型."""
assert "image/jpeg" in ALLOWED_IMAGE_MIME_TYPES
assert "image/png" in ALLOWED_IMAGE_MIME_TYPES
def test_allowed_video_mime_types_has_common(self):
"""视频 MIME 白名单包含常见类型."""
assert "video/mp4" in ALLOWED_VIDEO_MIME_TYPES
@@ -6,7 +6,6 @@ domain 层纯逻辑模块,0 FFmpeg 依赖,快速轻量。
from __future__ import annotations
import dataclasses
from unittest.mock import MagicMock
import pytest
+702
View File
@@ -0,0 +1,702 @@
"""watermark_config 单测.
domain 层水印配置纯逻辑模块,0 FFmpeg 依赖。
覆盖:配置解析、校验、位置计算、滤镜构建等。
"""
from __future__ import annotations
import pytest
from packages.domain.watermark_config import (
DEFAULT_FONT_COLOR,
DEFAULT_FONT_SIZE,
DEFAULT_MARGIN_X,
DEFAULT_MARGIN_Y,
DEFAULT_MODE,
DEFAULT_OPACITY,
DEFAULT_POSITION,
DEFAULT_SCALE,
VALID_POSITIONS,
WATERMARK_POSITIONS,
WatermarkConfig,
build_image_watermark_filter,
build_text_watermark_filter,
calc_position,
calc_scroll_x,
get_position_display_name,
get_position_names,
)
# ═══════════════════════════════════════════════════════════════════════════════
# 常量测试
# ═══════════════════════════════════════════════════════════════════════════════
class TestConstants:
"""常量测试."""
def test_default_position(self):
"""默认位置右下角."""
assert DEFAULT_POSITION == "bottom_right"
def test_default_mode(self):
"""默认模式 text."""
assert DEFAULT_MODE == "text"
def test_default_scale(self):
"""默认缩放 0.2."""
assert DEFAULT_SCALE == 0.2
def test_default_opacity(self):
"""默认透明度 0.8."""
assert DEFAULT_OPACITY == 0.8
def test_default_font_size(self):
"""默认字号 24."""
assert DEFAULT_FONT_SIZE == 24
def test_nine_positions(self):
"""9 个合法位置."""
assert len(VALID_POSITIONS) == 9
assert "top_left" in VALID_POSITIONS
assert "top_center" in VALID_POSITIONS
assert "top_right" in VALID_POSITIONS
assert "center_left" in VALID_POSITIONS
assert "center" in VALID_POSITIONS
assert "center_right" in VALID_POSITIONS
assert "bottom_left" in VALID_POSITIONS
assert "bottom_center" in VALID_POSITIONS
assert "bottom_right" in VALID_POSITIONS
# ═══════════════════════════════════════════════════════════════════════════════
# WatermarkConfig.from_dict
# ═══════════════════════════════════════════════════════════════════════════════
class TestWatermarkConfigFromDict:
"""from_dict 构造测试."""
def test_none_returns_none(self):
"""None 返回 None."""
assert WatermarkConfig.from_dict(None) is None
def test_empty_dict_returns_none(self):
"""空字典返回 None."""
assert WatermarkConfig.from_dict({}) is None
def test_disabled_returns_none(self):
"""enabled=False 返回 None."""
assert WatermarkConfig.from_dict({"enabled": False}) is None
def test_image_mode_requires_image_path(self):
"""图片模式缺少 image_path 返回 None."""
assert WatermarkConfig.from_dict({"enabled": True, "mode": "image"}) is None
def test_text_mode_requires_text(self):
"""文字模式缺少 text 返回 None."""
result = WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": ""})
assert result is None
def test_image_mode_valid(self):
"""图片模式配置有效."""
cfg = WatermarkConfig.from_dict(
{
"enabled": True,
"mode": "image",
"image_path": "/tmp/wm.png",
"position": "top_left",
"scale": 0.3,
"opacity": 0.5,
}
)
assert cfg is not None
assert cfg.mode == "image"
assert cfg.image_path == "/tmp/wm.png"
assert cfg.position == "top_left"
assert cfg.scale == 0.3
assert cfg.opacity == 0.5
def test_text_mode_valid(self):
"""文字模式配置有效."""
cfg = WatermarkConfig.from_dict(
{
"enabled": True,
"mode": "text",
"text": "Hello Watermark",
"font_size": 32,
"font_color": "red",
"position": "bottom_left",
}
)
assert cfg is not None
assert cfg.mode == "text"
assert cfg.text == "Hello Watermark"
assert cfg.font_size == 32
assert cfg.font_color == "red"
assert cfg.position == "bottom_left"
def test_invalid_position_defaults(self):
"""非法 position 回退到默认."""
cfg = WatermarkConfig.from_dict(
{
"enabled": True,
"mode": "text",
"text": "test",
"position": "invalid_pos",
}
)
assert cfg.position == DEFAULT_POSITION
def test_default_values_when_not_provided(self):
"""未提供的字段使用默认值."""
cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": "hi"})
assert cfg is not None
assert cfg.position == DEFAULT_POSITION
assert cfg.opacity == DEFAULT_OPACITY
assert cfg.font_size == DEFAULT_FONT_SIZE
assert cfg.font_color == DEFAULT_FONT_COLOR
assert cfg.margin_x == DEFAULT_MARGIN_X
assert cfg.margin_y == DEFAULT_MARGIN_Y
def test_image_alias_key(self):
"""image 字段作为 image_path 的别名."""
cfg = WatermarkConfig.from_dict(
{
"enabled": True,
"mode": "image",
"image": "/tmp/wm.jpg",
}
)
assert cfg is not None
assert cfg.image_path == "/tmp/wm.jpg"
def test_scroll_config_parsed(self):
"""滚动配置被解析."""
cfg = WatermarkConfig.from_dict(
{
"enabled": True,
"mode": "text",
"text": "scroll",
"scroll": True,
"scroll_speed": 100,
}
)
assert cfg.scroll is True
assert cfg.scroll_speed == 100
def test_scroll_default_false(self):
"""scroll 默认 False."""
cfg = WatermarkConfig.from_dict({"enabled": True, "mode": "text", "text": "hi"})
assert cfg.scroll is False
# ═══════════════════════════════════════════════════════════════════════════════
# WatermarkConfig.validate
# ═══════════════════════════════════════════════════════════════════════════════
class TestWatermarkConfigValidate:
"""validate 校验测试."""
def test_valid_image_config(self):
"""合法图片配置."""
cfg = WatermarkConfig(
mode="image",
image_path="/tmp/wm.png",
opacity=0.8,
scale=0.3,
position="top_left",
)
ok, err = cfg.validate()
assert ok is True
assert err == ""
def test_valid_text_config(self):
"""合法文字配置."""
cfg = WatermarkConfig(
mode="text",
text="Hello",
font_size=24,
opacity=0.8,
position="bottom_right",
)
ok, err = cfg.validate()
assert ok is True
assert err == ""
def test_invalid_position(self):
"""非法位置."""
cfg = WatermarkConfig(mode="text", text="hi", position="invalid")
ok, err = cfg.validate()
assert ok is False
assert "位置" in err
def test_opacity_too_high(self):
"""透明度 > 1."""
cfg = WatermarkConfig(mode="text", text="hi", opacity=1.5)
ok, err = cfg.validate()
assert ok is False
assert "透明度" in err
def test_opacity_negative(self):
"""透明度 < 0."""
cfg = WatermarkConfig(mode="text", text="hi", opacity=-0.1)
ok, err = cfg.validate()
assert ok is False
assert "透明度" in err
def test_opacity_zero_valid(self):
"""透明度 = 0 合法(虽然没效果)."""
cfg = WatermarkConfig(mode="text", text="hi", opacity=0.0)
ok, _ = cfg.validate()
assert ok is True
def test_opacity_one_valid(self):
"""透明度 = 1 合法."""
cfg = WatermarkConfig(mode="text", text="hi", opacity=1.0)
ok, _ = cfg.validate()
assert ok is True
def test_image_no_path(self):
"""图片模式无路径."""
cfg = WatermarkConfig(mode="image", image_path="")
ok, err = cfg.validate()
assert ok is False
assert "图片路径" in err
def test_image_scale_too_small(self):
"""缩放比例太小."""
cfg = WatermarkConfig(mode="image", image_path="/tmp/a.png", scale=0.001)
ok, err = cfg.validate()
assert ok is False
assert "缩放比例" in err
def test_image_scale_too_large(self):
"""缩放比例 > 1."""
cfg = WatermarkConfig(mode="image", image_path="/tmp/a.png", scale=1.5)
ok, err = cfg.validate()
assert ok is False
assert "缩放比例" in err
def test_image_scale_bounds_valid(self):
"""缩放边界值 0.01 和 1.0 合法."""
cfg1 = WatermarkConfig(mode="image", image_path="/tmp/a.png", scale=0.01)
cfg2 = WatermarkConfig(mode="image", image_path="/tmp/a.png", scale=1.0)
assert cfg1.validate()[0] is True
assert cfg2.validate()[0] is True
def test_text_empty(self):
"""文字模式空文字."""
cfg = WatermarkConfig(mode="text", text="")
ok, err = cfg.validate()
assert ok is False
assert "文字" in err
def test_text_font_size_zero(self):
"""字号为 0."""
cfg = WatermarkConfig(mode="text", text="hi", font_size=0)
ok, err = cfg.validate()
assert ok is False
assert "字体大小" in err
def test_text_font_size_negative(self):
"""字号为负."""
cfg = WatermarkConfig(mode="text", text="hi", font_size=-5)
ok, err = cfg.validate()
assert ok is False
assert "字体大小" in err
def test_invalid_mode(self):
"""非法模式."""
cfg = WatermarkConfig(mode="video", text="hi")
ok, err = cfg.validate()
assert ok is False
assert "模式" in err
# ═══════════════════════════════════════════════════════════════════════════════
# WatermarkConfig.has_effect
# ═══════════════════════════════════════════════════════════════════════════════
class TestWatermarkConfigHasEffect:
"""has_effect 测试."""
def test_image_with_path_and_opacity(self):
"""图片水印有路径且透明度>0."""
cfg = WatermarkConfig(mode="image", image_path="/tmp/wm.png", opacity=0.5)
assert cfg.has_effect() is True
def test_image_zero_opacity_no_effect(self):
"""透明度 0 无效果."""
cfg = WatermarkConfig(mode="image", image_path="/tmp/wm.png", opacity=0.0)
assert cfg.has_effect() is False
def test_image_empty_path_no_effect(self):
"""空路径无效果."""
cfg = WatermarkConfig(mode="image", image_path="", opacity=0.8)
assert cfg.has_effect() is False
def test_text_with_content_and_opacity(self):
"""文字水印有内容且透明度>0且字号>0."""
cfg = WatermarkConfig(mode="text", text="Hello", opacity=0.8, font_size=24)
assert cfg.has_effect() is True
def test_text_empty_no_effect(self):
"""空文字无效果."""
cfg = WatermarkConfig(mode="text", text="", opacity=0.8, font_size=24)
assert cfg.has_effect() is False
def test_text_zero_opacity_no_effect(self):
"""透明度 0 无效果."""
cfg = WatermarkConfig(mode="text", text="Hi", opacity=0.0, font_size=24)
assert cfg.has_effect() is False
def test_text_zero_font_size_no_effect(self):
"""字号 0 无效果."""
cfg = WatermarkConfig(mode="text", text="Hi", opacity=0.8, font_size=0)
assert cfg.has_effect() is False
def test_invalid_mode_no_effect(self):
"""非法模式无效果."""
cfg = WatermarkConfig(mode="unknown")
assert cfg.has_effect() is False
# ═══════════════════════════════════════════════════════════════════════════════
# calc_position
# ═══════════════════════════════════════════════════════════════════════════════
class TestCalcPosition:
"""位置计算测试."""
OUT_W = 1080
OUT_H = 1920
WM_W = 200
WM_H = 50
MX = 20
MY = 15
def test_top_left(self):
"""左上角."""
x, y = calc_position("top_left", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == 20
assert y == 15
def test_top_center(self):
"""顶部居中."""
x, y = calc_position("top_center", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == (1080 - 200) // 2
assert y == 15
def test_top_right(self):
"""右上角."""
x, y = calc_position("top_right", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == 1080 - 200 - 20
assert y == 15
def test_center_left(self):
"""左中."""
x, y = calc_position("center_left", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == 20
assert y == (1920 - 50) // 2
def test_center(self):
"""正中心."""
x, y = calc_position("center", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == (1080 - 200) // 2
assert y == (1920 - 50) // 2
def test_center_right(self):
"""右中."""
x, y = calc_position("center_right", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == 1080 - 200 - 20
assert y == (1920 - 50) // 2
def test_bottom_left(self):
"""左下角."""
x, y = calc_position("bottom_left", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == 20
assert y == 1920 - 50 - 15
def test_bottom_center(self):
"""底部居中."""
x, y = calc_position("bottom_center", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == (1080 - 200) // 2
assert y == 1920 - 50 - 15
def test_bottom_right(self):
"""右下角."""
x, y = calc_position("bottom_right", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == 1080 - 200 - 20
assert y == 1920 - 50 - 15
def test_invalid_position_defaults_bottom_right(self):
"""非法位置默认右下角."""
x, y = calc_position("invalid", self.OUT_W, self.OUT_H, self.WM_W, self.WM_H, self.MX, self.MY)
assert x == 1080 - 200 - 20
assert y == 1920 - 50 - 15
def test_zero_margin(self):
"""零边距."""
x, y = calc_position("top_left", 100, 100, 50, 30, 0, 0)
assert x == 0
assert y == 0
def test_large_margin(self):
"""大边距."""
x, y = calc_position("top_left", 100, 100, 50, 30, 10, 10)
assert x == 10
assert y == 10
# ═══════════════════════════════════════════════════════════════════════════════
# calc_scroll_x
# ═══════════════════════════════════════════════════════════════════════════════
class TestCalcScrollX:
"""滚动 x 表达式测试."""
def test_returns_string_expression(self):
"""返回字符串表达式."""
expr = calc_scroll_x("bottom", 1080, 200, 50)
assert isinstance(expr, str)
assert len(expr) > 0
def test_contains_output_width(self):
"""包含输出宽度."""
expr = calc_scroll_x("bottom", 1080, 200, 50)
assert "1080" in expr
def test_contains_wm_width(self):
"""包含水印宽度."""
expr = calc_scroll_x("bottom", 1080, 200, 50)
assert "200" in expr
def test_contains_speed(self):
"""包含速度."""
expr = calc_scroll_x("bottom", 1080, 200, 50)
assert "50" in expr
def test_contains_mod_function(self):
"""包含 mod 函数(跑马灯效果)."""
expr = calc_scroll_x("bottom", 1080, 200, 50)
assert "mod" in expr
# ═══════════════════════════════════════════════════════════════════════════════
# build_image_watermark_filter
# ═══════════════════════════════════════════════════════════════════════════════
class TestBuildImageWatermarkFilter:
"""图片水印滤镜构建测试."""
def test_basic_image_watermark(self):
"""基础图片水印滤镜."""
cfg = WatermarkConfig(
mode="image",
image_path="/tmp/wm.png",
position="bottom_right",
scale=0.2,
opacity=0.8,
margin_x=20,
margin_y=20,
)
filter_str, inputs = build_image_watermark_filter(
"[in_v]",
"/tmp/wm.png",
1080,
1920,
"[out_v]",
cfg,
)
assert "overlay=" in filter_str
assert "scale=" in filter_str
assert inputs == ["-i", "/tmp/wm.png"]
def test_contains_scaled_width(self):
"""包含缩放后的宽度."""
cfg = WatermarkConfig(mode="image", image_path="/a.png", scale=0.2, position="top_left")
filter_str, _ = build_image_watermark_filter("[in]", "/a.png", 1000, 500, "[out]", cfg)
assert "scale=200:" in filter_str # 1000 * 0.2 = 200
def test_full_opacity_no_alpha_filter(self):
"""透明度 1.0 不加 alpha 滤镜."""
cfg = WatermarkConfig(mode="image", image_path="/a.png", opacity=1.0, position="top_left")
filter_str, _ = build_image_watermark_filter("[in]", "/a.png", 1000, 500, "[out]", cfg)
assert "colorchannelmixer" not in filter_str
def test_partial_opacity_has_alpha(self):
"""透明度 < 1.0 加 alpha 滤镜."""
cfg = WatermarkConfig(mode="image", image_path="/a.png", opacity=0.5, position="top_left")
filter_str, _ = build_image_watermark_filter("[in]", "/a.png", 1000, 500, "[out]", cfg)
assert "colorchannelmixer=aa=0.5" in filter_str
def test_position_top_left_coordinates(self):
"""左上角位置的 overlay 坐标."""
cfg = WatermarkConfig(
mode="image", image_path="/a.png", position="top_left", margin_x=10, margin_y=10, scale=0.1
)
filter_str, _ = build_image_watermark_filter("[in]", "/a.png", 1000, 500, "[out]", cfg)
assert "x=10:y=10" in filter_str
def test_scroll_watermark_expression(self):
"""滚动水印使用表达式."""
cfg = WatermarkConfig(
mode="image",
image_path="/a.png",
position="bottom",
scroll=True,
scroll_speed=60,
scale=0.2,
)
filter_str, _ = build_image_watermark_filter("[in]", "/a.png", 1000, 500, "[out]", cfg)
assert "mod(" in filter_str
assert "60" in filter_str # scroll_speed
def test_filter_has_two_parts(self):
"""滤镜包含两部分(缩放 + overlay)."""
cfg = WatermarkConfig(mode="image", image_path="/a.png", position="top_left")
filter_str, _ = build_image_watermark_filter("[in]", "/a.png", 1000, 500, "[out]", cfg)
parts = filter_str.split(";")
assert len(parts) == 2
def test_output_label(self):
"""输出标签正确."""
cfg = WatermarkConfig(mode="image", image_path="/a.png", position="top_left")
filter_str, _ = build_image_watermark_filter("[in]", "/a.png", 1000, 500, "[my_out]", cfg)
assert filter_str.endswith("[my_out]")
# ═══════════════════════════════════════════════════════════════════════════════
# build_text_watermark_filter
# ═══════════════════════════════════════════════════════════════════════════════
class TestBuildTextWatermarkFilter:
"""文字水印滤镜构建测试."""
def test_basic_text_watermark(self):
"""基础文字水印."""
cfg = WatermarkConfig(
mode="text",
text="Hello World",
font_size=24,
font_color="white",
opacity=0.8,
position="bottom_right",
)
result = build_text_watermark_filter("[in_v]", "[out_v]", cfg, 1080, 1920)
assert "drawtext=" in result
assert "text='Hello World'" in result
assert "fontsize=24" in result
def test_text_color_with_opacity(self):
"""字体颜色带透明度."""
cfg = WatermarkConfig(mode="text", text="Hi", font_color="red", opacity=0.5)
result = build_text_watermark_filter("[in]", "[out]", cfg, 1000, 500)
assert "fontcolor=red@0.5" in result
def test_text_escapes_colon(self):
"""文字中的冒号被转义."""
cfg = WatermarkConfig(mode="text", text="time: 10:00")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1000, 500)
assert "time\\: 10\\:00" in result
def test_text_escapes_single_quote(self):
"""文字中的单引号被转义."""
cfg = WatermarkConfig(mode="text", text="it's")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1000, 500)
assert "it\\'s" in result
def test_font_path_included(self):
"""指定字体路径时包含 fontfile."""
cfg = WatermarkConfig(mode="text", text="Hi", font_path="/tmp/font.ttf")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1000, 500)
assert "fontfile=" in result
assert "/tmp/font.ttf" in result
def test_no_font_path_when_empty(self):
"""无字体路径时不含 fontfile."""
cfg = WatermarkConfig(mode="text", text="Hi", font_path="")
result = build_text_watermark_filter("[in]", "[out]", cfg, 1000, 500)
assert "fontfile=" not in result
def test_position_coordinates(self):
"""位置计算正确."""
cfg = WatermarkConfig(
mode="text",
text="Hi",
position="top_left",
font_size=20,
margin_x=10,
margin_y=10,
)
result = build_text_watermark_filter("[in]", "[out]", cfg, 1000, 500)
# "Hi" 2 字 * 20px = 40 宽,高 20
# top_left: x=10, y=10
assert "x=10:y=10" in result
def test_scroll_text_watermark(self):
"""滚动文字水印."""
cfg = WatermarkConfig(mode="text", text="Scroll", scroll=True, scroll_speed=80)
result = build_text_watermark_filter("[in]", "[out]", cfg, 1000, 500)
assert "x=w-mod(" in result
assert "80" in result
def test_output_label(self):
"""输出标签正确."""
cfg = WatermarkConfig(mode="text", text="Hi")
result = build_text_watermark_filter("[in]", "[final]", cfg, 1000, 500)
assert result.endswith("[final]")
def test_input_label(self):
"""输入标签正确."""
cfg = WatermarkConfig(mode="text", text="Hi")
result = build_text_watermark_filter("[video_in]", "[out]", cfg, 1000, 500)
assert result.startswith("[video_in]")
# ═══════════════════════════════════════════════════════════════════════════════
# get_position_names / get_position_display_name
# ═══════════════════════════════════════════════════════════════════════════════
class TestPositionUtilities:
"""位置工具函数测试."""
def test_get_position_names_returns_nine(self):
"""返回 9 个位置名."""
names = get_position_names()
assert len(names) == 9
def test_get_position_names_order(self):
"""从上到下从左到右的顺序."""
names = get_position_names()
assert names[0] == "top_left"
assert names[2] == "top_right"
assert names[4] == "center"
assert names[8] == "bottom_right"
def test_get_position_display_name_valid(self):
"""合法位置返回中文名."""
assert get_position_display_name("top_left") == "左上"
assert get_position_display_name("center") == "中心"
assert get_position_display_name("bottom_right") == "右下"
def test_get_position_display_name_invalid(self):
"""非法位置返回原字符串."""
assert get_position_display_name("invalid") == "invalid"
assert get_position_display_name("") == ""
def test_all_positions_have_display_names(self):
"""所有 9 个位置都有中文显示名."""
for pos in VALID_POSITIONS:
display = get_position_display_name(pos)
assert display != pos
assert len(display) > 0
-1
View File
@@ -4,7 +4,6 @@ from __future__ import annotations
import dataclasses
import pytest
from packages.domain.transition_presets import (