Su YijunandClaude Opus 5.5
1488ab667c
- 数据库连接改为运行时初始化：DATABASE_URL > data/db_config.json > 安装向导 - 新增 /api/setup/status|test|save 与安装向导页（连接数据库 → 设置管理员） - 连接失败给出可操作的中文提示；有配置但暂时连不上时自动重试，不需要重填 - docker-compose.yml 不再包含数据库设置；README / DOCKERHUB.md 改为“建库 → 启动 → 网页填写” - CI 增加真实 MySQL 8.4 上的安装向导测试 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
43 lines
1.7 KiB
Python
43 lines
1.7 KiB
Python
import os
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import bootstrap, database
|
|
from app.config import get_settings
|
|
from app.engine import engine
|
|
from app.main import app
|
|
|
|
|
|
def test_build_url_escapes_and_ipv6():
|
|
u = bootstrap.build_url({"host": "fd00::5", "port": 3307, "user": "ro@s", "password": "p@ss:/#", "database": "rosdpi"})
|
|
assert u == "mysql+pymysql://ro%40s:p%40ss%3A%2F%23@[fd00::5]:3307/rosdpi?charset=utf8mb4"
|
|
|
|
|
|
def test_explain_common_errors():
|
|
assert "账号或密码错误" in bootstrap.explain(Exception("(1045, \"Access denied for user 'x'\")"))
|
|
assert "127.0.0.1" in bootstrap.explain(Exception("(2003, \"Can't connect to MySQL server\")"))
|
|
|
|
|
|
def test_configured_via_env_blocks_wizard():
|
|
with TestClient(app) as c:
|
|
st = c.get("/api/setup/status").json()
|
|
assert st == {"configured": True, "source": "env", "config_present": True}
|
|
r = c.post("/api/setup/save", json={"host": "x", "user": "u", "password": "p", "database": "rosdpi"})
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_wizard_mode_without_database(monkeypatch):
|
|
url = os.environ["DATABASE_URL"]
|
|
monkeypatch.setattr(get_settings(), "database_url", "")
|
|
database.reset_engine()
|
|
try:
|
|
c = TestClient(app) # 不触发 lifespan，只测接口
|
|
assert c.get("/api/setup/status").json() == {"configured": False, "source": None, "config_present": False}
|
|
assert c.get("/api/auth/status").status_code == 503
|
|
r = c.post("/api/setup/save", json={"host": "x", "user": "u", "database": "bad name"})
|
|
assert r.status_code == 422
|
|
engine.flush() # 未配置时不报错、也不丢弃内存数据
|
|
finally:
|
|
database.connect(url)
|
|
assert database.is_configured()
|