Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c245b5bcd2 | |||
| 6586a8b943 | |||
| 834d9f219e |
@@ -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,
|
||||
|
||||
Executable
+567
@@ -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
|
||||
|
||||
Executable
+702
@@ -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
|
||||
@@ -4,7 +4,6 @@ from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
|
||||
|
||||
import pytest
|
||||
|
||||
from packages.domain.transition_presets import (
|
||||
|
||||
Reference in New Issue
Block a user