test: establish monitoring configuration contract

This commit is contained in:
2026-08-04 11:01:09 +08:00
parent 63bf68e73f
commit 44bf6e003e
8 changed files with 127 additions and 14 deletions
+25 -9
View File
@@ -1,20 +1,33 @@
# 运行环境
ENVIRONMENT=development
# 数据库(开发用 SQLite,生产用 PostgreSQL
DATABASE_URL=sqlite+aiosqlite:///./pingwatch.db
# DATABASE_URL=postgresql+asyncpg://user:pass@localhost/pingwatch
# 企业微信
WECOM_CORP_ID=your_corp_id
WECOM_AGENT_ID=1000001
WECOM_APP_SECRET=your_app_secret
WECOM_NOTIFICATION_ENABLED=false
WECOM_CORP_ID=
WECOM_AGENT_ID=0
WECOM_APP_SECRET=
WECOM_TO_PARTY=
WECOM_NOTIFICATION_MAX_ATTEMPTS=5
WECOM_RETRY_BASE_SECONDS=30
# Casdoor
CASDOOR_ENDPOINT=https://casdoor.dhdx.fun
CASDOOR_CLIENT_ID=e46b9e1eb893027bdf2a
CASDOOR_CLIENT_SECRET=b12c7e1688ed51481f3b5c5dae4191b6edbba916
CASDOOR_CLIENT_ID=
CASDOOR_CLIENT_SECRET=
CASDOOR_CERTIFICATE=
CASDOOR_ORGANIZATION=dahua
CASDOOR_APPLICATION=PingWatch
CASDOOR_REDIRECT_URI=http://10.10.10.7:5173/login
CASDOOR_REDIRECT_URI=
# 连通性监测
PROBE_PACKETS_PER_ROUND=3
OFFLINE_CONSECUTIVE_ROUNDS=2
DEGRADED_WINDOW_ROUNDS=5
DEGRADED_LOSS_PERCENT=20.0
RECOVERY_CONSECUTIVE_CLEAN_ROUNDS=3
# Ping 引擎
PING_INTERVAL_SECONDS=30
@@ -31,5 +44,8 @@ OFFLINE_SUPPRESS_RATIO=0.9
PING_RECORD_RETENTION_DAYS=90
ALERT_RETENTION_DAYS=365
# JWT 密钥(请改成随机字符串
SECRET_KEY=change-me-to-a-long-random-string
# JWT 密钥(生产环境必须由受控运行环境注入
SECRET_KEY=
# LogHive(可选)
LOGHIVE_API_KEY=
+33 -4
View File
@@ -1,10 +1,16 @@
"""应用配置,通过环境变量注入,不支持 .env 文件"""
"""Application configuration loaded from environment variables or a local .env file."""
from pydantic import Field
from pydantic_settings import BaseSettings
from typing import Optional
LEGACY_DEFAULT_SECRET = "change-me-to-a-long-random-string"
class Settings(BaseSettings):
# ---------- 运行环境 ----------
environment: str = "development"
# ---------- 数据库 ----------
DATABASE_URL: str = "sqlite+aiosqlite:///./pingwatch.db"
# PostgreSQL: "postgresql+asyncpg://user:pass@localhost/pingwatch"
@@ -15,7 +21,7 @@ class Settings(BaseSettings):
WECOM_APP_SECRET: str = ""
# ---------- Casdoor ----------
CASDOOR_ENDPOINT: str = "https://casdoor.dhdx.fun"
casdoor_endpoint: str = "https://casdoor.dhdx.fun"
CASDOOR_CLIENT_ID: str = ""
CASDOOR_CLIENT_SECRET: str = ""
CASDOOR_CERTIFICATE: str = "" # 可选,用于验证 id_token 签名
@@ -23,6 +29,19 @@ class Settings(BaseSettings):
CASDOOR_APPLICATION: str = "PingWatch"
CASDOOR_REDIRECT_URI: str = "http://10.10.10.7:5173/login" # 前端回调地址
# ---------- 连通性监测 ----------
probe_packets_per_round: int = Field(default=3, ge=1, le=10)
offline_consecutive_rounds: int = Field(default=2, ge=1, le=10)
degraded_window_rounds: int = Field(default=5, ge=2, le=60)
degraded_loss_percent: float = Field(default=20.0, ge=1, le=100)
recovery_consecutive_clean_rounds: int = Field(default=3, ge=1, le=20)
# ---------- 企业微信投递 ----------
wecom_notification_enabled: bool = False
wecom_notification_max_attempts: int = Field(default=5, ge=1, le=10)
wecom_retry_base_seconds: int = Field(default=30, ge=1, le=3600)
WECOM_TO_PARTY: str = ""
# ---------- Ping 引擎 ----------
PING_INTERVAL_SECONDS: int = 30
PING_TIMEOUT_SECONDS: float = 5.0
@@ -40,7 +59,7 @@ class Settings(BaseSettings):
ALERT_RETENTION_DAYS: int = 365
# ---------- JWT ----------
SECRET_KEY: str = "change-me-to-a-long-random-string"
secret_key: str = LEGACY_DEFAULT_SECRET
ACCESS_TOKEN_EXPIRE_MINUTES: int = 480
# ---------- LogHive ----------
@@ -53,5 +72,15 @@ class Settings(BaseSettings):
model_config = {"env_file": ".env", "env_file_encoding": "utf-8"}
@property
def CASDOOR_ENDPOINT(self) -> str:
"""Compatibility accessor for existing uppercase configuration consumers."""
return self.casdoor_endpoint
@property
def SECRET_KEY(self) -> str:
"""Compatibility accessor for existing uppercase configuration consumers."""
return self.secret_key
settings = Settings()
@@ -0,0 +1,21 @@
"""Runtime validation for security-sensitive application settings."""
from app.config import LEGACY_DEFAULT_SECRET, Settings
def validate_runtime_settings(settings: Settings) -> None:
"""Reject unsafe settings before scheduling monitoring work."""
if settings.environment == "production":
if settings.secret_key == LEGACY_DEFAULT_SECRET:
raise ValueError("SECRET_KEY must be provided by the runtime environment")
if not settings.casdoor_endpoint.startswith("https://"):
raise ValueError("CASDOOR endpoint must use TLS in production")
if settings.wecom_notification_enabled:
credentials = (
settings.WECOM_CORP_ID,
settings.WECOM_AGENT_ID,
settings.WECOM_APP_SECRET,
)
if not all(credentials):
raise ValueError("WECOM delivery requires corp ID, agent ID, and app secret")
+4
View File
@@ -0,0 +1,4 @@
[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
testpaths = tests
+3 -1
View File
@@ -1,7 +1,7 @@
fastapi>=0.110.0
uvicorn[standard]>=0.29.0
sqlalchemy[asyncio]>=2.0.30
aiosqlite>=0.20.0
aiosqlite==0.20.0
asyncpg>=0.29.0
pydantic>=2.7.0
pydantic-settings>=2.2.0
@@ -9,3 +9,5 @@ python-jose[cryptography]>=3.3.0
httpx>=0.27.0
python-multipart>=0.0.9
websockets>=12.0
pytest==8.3.5
pytest-asyncio==0.25.3
+1
View File
@@ -0,0 +1 @@
"""Shared test configuration for PingWatch backend tests."""
+38
View File
@@ -0,0 +1,38 @@
import pytest
from pydantic import ValidationError
from app.config import Settings
from app.services.settings_validation import validate_runtime_settings
def test_production_rejects_legacy_jwt_secret():
settings = Settings(
environment="production",
secret_key="change-me-to-a-long-random-string",
)
with pytest.raises(ValueError, match="SECRET_KEY"):
validate_runtime_settings(settings)
def test_probe_packet_count_must_be_positive():
with pytest.raises(ValidationError):
Settings(probe_packets_per_round=0)
def test_production_rejects_tls_disabled_casdoor():
settings = Settings(
environment="production",
secret_key="a-safe-production-secret",
casdoor_endpoint="http://casdoor.internal",
)
with pytest.raises(ValueError, match="CASDOOR"):
validate_runtime_settings(settings)
def test_enabled_wecom_delivery_requires_all_credentials():
settings = Settings(wecom_notification_enabled=True, WECOM_CORP_ID="corp")
with pytest.raises(ValueError, match="WECOM"):
validate_runtime_settings(settings)