320 lines
12 KiB
Python
320 lines
12 KiB
Python
import json
|
|
import pytest
|
|
import httpx
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base
|
|
from app.models.upstream import Upstream
|
|
from app.models.upstream_key import UpstreamGeneratedKey
|
|
from app.models.website import Website
|
|
from app.routers.websites import preview_cleanup_invalid_accounts, execute_cleanup_invalid_accounts
|
|
from app.services.website_client import Sub2ApiWebsiteClient
|
|
|
|
|
|
@pytest.fixture()
|
|
def db_session():
|
|
engine = create_engine(
|
|
"sqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
from app.models import admin_user, upstream, snapshot, webhook_config, notification_log, custom_page, website, revoked_token, upstream_key
|
|
Base.metadata.create_all(bind=engine)
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
db = TestingSessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
|
|
class MockResponse:
|
|
def __init__(self, status_code, lines, raise_exc=None):
|
|
self.status_code = status_code
|
|
self._lines = lines
|
|
self._raise_exc = raise_exc
|
|
|
|
def __enter__(self):
|
|
if self._raise_exc:
|
|
raise self._raise_exc
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
pass
|
|
|
|
def iter_lines(self):
|
|
return self._lines
|
|
|
|
|
|
def mock_stream(status_code, lines, raise_exc=None):
|
|
def mock_fn(*args, **kwargs):
|
|
return MockResponse(status_code, lines, raise_exc)
|
|
return mock_fn
|
|
|
|
|
|
def test_website_client_test_account_sse_scenarios(monkeypatch):
|
|
"""测试 Sub2ApiWebsiteClient.test_account 在各种 SSE 场景下的解析、处理、ID 转义和 API 日志记录。"""
|
|
c = Sub2ApiWebsiteClient(
|
|
base_url="http://w1",
|
|
api_prefix="api/v1",
|
|
auth_type="bearer",
|
|
auth_config={"token": "t1"},
|
|
)
|
|
|
|
log_calls = []
|
|
def mock_log(**kwargs):
|
|
log_calls.append(kwargs)
|
|
monkeypatch.setattr("app.services.website_client.log_external_api_call", mock_log)
|
|
|
|
# 1. success=true，包含带有斜杠的 ID 校验以测试 URL quote 转义
|
|
monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: test_complete", "data: {\"success\": true}"]))
|
|
res = c.test_account("acc/1")
|
|
assert res["status"] == "success"
|
|
assert res["cleanup_allowed"] is False
|
|
# 验证日志记录和 URL 转义：acc/1 应被转义为 acc%2F1
|
|
assert len(log_calls) == 1
|
|
assert log_calls[-1]["method"] == "POST"
|
|
assert log_calls[-1]["path"] == "/api/v1/accounts/acc%2F1/test"
|
|
assert log_calls[-1]["success"] is True
|
|
|
|
# 2. success=false
|
|
monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: test_complete", "data: {\"success\": false}"]))
|
|
res = c.test_account("acc/1")
|
|
assert res["status"] == "failed"
|
|
assert res["cleanup_allowed"] is True
|
|
assert len(log_calls) == 2
|
|
assert log_calls[-1]["success"] is True
|
|
|
|
# 3. HTTP 404
|
|
monkeypatch.setattr(c._client, "stream", mock_stream(404, []))
|
|
res = c.test_account("acc/1")
|
|
assert res["status"] == "404"
|
|
assert res["cleanup_allowed"] is True
|
|
assert len(log_calls) == 3
|
|
assert log_calls[-1]["status_code"] == 404
|
|
assert log_calls[-1]["success"] is False
|
|
assert log_calls[-1]["error_type"] == "HTTPStatus"
|
|
assert log_calls[-1]["error_message"] == "HTTP 404"
|
|
|
|
# 4. error event
|
|
monkeypatch.setattr(c._client, "stream", mock_stream(200, ["event: error", "data: \"internal error\""]))
|
|
res = c.test_account("acc/1")
|
|
assert res["status"] == "error"
|
|
assert res["cleanup_allowed"] is False
|
|
assert len(log_calls) == 4
|
|
assert log_calls[-1]["success"] is False
|
|
assert log_calls[-1]["error_type"] == "WebsiteError"
|
|
|
|
# 5. 超时
|
|
monkeypatch.setattr(c._client, "stream", mock_stream(200, [], raise_exc=httpx.TimeoutException("Connection timed out")))
|
|
res = c.test_account("acc/1")
|
|
assert res["status"] == "timeout"
|
|
assert res["cleanup_allowed"] is False
|
|
assert len(log_calls) == 5
|
|
assert log_calls[-1]["success"] is False
|
|
assert log_calls[-1]["error_type"] == "TimeoutException"
|
|
|
|
# 6. 非法 SSE (没有 test_complete)
|
|
monkeypatch.setattr(c._client, "stream", mock_stream(200, ["data: hello world"]))
|
|
res = c.test_account("acc/1")
|
|
assert res["status"] == "invalid_sse"
|
|
assert res["cleanup_allowed"] is False
|
|
assert len(log_calls) == 6
|
|
assert log_calls[-1]["success"] is False
|
|
assert log_calls[-1]["error_type"] == "WebsiteError"
|
|
|
|
|
|
def test_cleanup_invalid_preview_and_execute(db_session, monkeypatch):
|
|
"""测试预览与执行清理的完整工作流。
|
|
- 包含本地 orphaned 键。
|
|
- 包含远端 SmartUp 备注孤立账号。
|
|
- 包含无 SmartUp 备注的手工账号（不参与候选）。
|
|
- 包含测试通过与失败的不同流程分支。
|
|
"""
|
|
w = Website(
|
|
name="W1",
|
|
site_type="sub2api",
|
|
base_url="http://w1",
|
|
enabled=True,
|
|
auth_config_json="{}",
|
|
timeout_seconds=30
|
|
)
|
|
u1 = Upstream(name="U1", base_url="http://u1")
|
|
db_session.add_all([w, u1])
|
|
db_session.commit()
|
|
db_session.refresh(w)
|
|
db_session.refresh(u1)
|
|
|
|
# 1. 本地 orphaned 键 k1：对应的远端账号测试将失败 (可清理)
|
|
k1 = UpstreamGeneratedKey(
|
|
upstream_id=u1.id,
|
|
group_id="G1",
|
|
group_name="G1-Name",
|
|
key_name="Key-G1",
|
|
key_value="sk-g1",
|
|
status="orphaned",
|
|
imported_website_id=w.id,
|
|
imported_account_id="acc-k1",
|
|
error="original-error-k1"
|
|
)
|
|
# 2. 本地 orphaned 键 k2：对应的远端账号测试将成功 (不可清理)
|
|
k2 = UpstreamGeneratedKey(
|
|
upstream_id=u1.id,
|
|
group_id="G2",
|
|
group_name="G2-Name",
|
|
key_name="Key-G2",
|
|
key_value="sk-g2",
|
|
status="orphaned",
|
|
imported_website_id=w.id,
|
|
imported_account_id="acc-k2",
|
|
error="original-error-k2"
|
|
)
|
|
# 3. 本地 imported 正常键 k3（不进入候选）
|
|
k3 = UpstreamGeneratedKey(
|
|
upstream_id=u1.id,
|
|
group_id="G3",
|
|
group_name="G3-Name",
|
|
key_name="Key-G3",
|
|
key_value="sk-g3",
|
|
status="imported",
|
|
imported_website_id=w.id,
|
|
imported_account_id="acc-k3"
|
|
)
|
|
db_session.add_all([k1, k2, k3])
|
|
db_session.commit()
|
|
|
|
deleted_accounts = []
|
|
test_calls = {}
|
|
|
|
class MockClient:
|
|
def __init__(self, **kwargs): pass
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): pass
|
|
|
|
def list_accounts(self):
|
|
# 远端列表：
|
|
# acc-k1 (Condition A, k1)
|
|
# acc-k2 (Condition A, k2)
|
|
# acc-k3 (k3 正常)
|
|
# acc-orphan-remote (Condition B, notes 带备注，本地不存在关联)
|
|
# acc-manual (无备注手工账号，notes 没匹配上，不应进候选)
|
|
return [
|
|
{"id": "acc-k1", "name": "Acc-K1", "notes": f"Imported by SmartUp from upstream key #{k1.id}"},
|
|
{"id": "acc-k2", "name": "Acc-K2", "notes": f"Imported by SmartUp from upstream key #{k2.id}"},
|
|
{"id": "acc-k3", "name": "Acc-K3", "notes": f"Imported by SmartUp from upstream key #{k3.id}"},
|
|
{"id": "acc-orphan-remote", "name": "Orphan-Remote", "notes": "Imported by SmartUp from upstream key #9999"},
|
|
{"id": "acc-manual", "name": "Manual-Acc", "notes": "Manually created account"},
|
|
]
|
|
|
|
def extract_id(self, val):
|
|
return val.get("id") if isinstance(val, dict) else str(val)
|
|
|
|
def test_account(self, account_id):
|
|
test_calls[account_id] = test_calls.get(account_id, 0) + 1
|
|
if account_id in ("acc-k1", "acc-orphan-remote"):
|
|
# 测试失败 -> 可删除
|
|
return {"status": "failed", "cleanup_allowed": True, "message": "API check failed"}
|
|
else:
|
|
# 其他测试成功
|
|
return {"status": "success", "cleanup_allowed": False, "message": "API check OK"}
|
|
|
|
def delete_account(self, account_id):
|
|
deleted_accounts.append(account_id)
|
|
return {"status": "deleted", "message": f"Account {account_id} has been deleted"}
|
|
|
|
monkeypatch.setattr("app.routers.websites._client", lambda row: MockClient())
|
|
|
|
# ── 1. 测试预览 ──
|
|
preview_res = preview_cleanup_invalid_accounts(wid=w.id, db=db_session)
|
|
assert preview_res.success is True
|
|
# 候选账号应为 3 个：acc-k1, acc-k2, acc-orphan-remote (acc-k3 正常, acc-manual 无备注，均不进入候选)
|
|
items = preview_res.items
|
|
assert len(items) == 3
|
|
|
|
item_k1 = next(i for i in items if i.account_id == "acc-k1")
|
|
assert item_k1.cleanup_allowed is True
|
|
assert item_k1.test_status == "failed"
|
|
|
|
item_k2 = next(i for i in items if i.account_id == "acc-k2")
|
|
assert item_k2.cleanup_allowed is False
|
|
assert item_k2.test_status == "test_passed"
|
|
|
|
item_orphan = next(i for i in items if i.account_id == "acc-orphan-remote")
|
|
assert item_orphan.cleanup_allowed is True
|
|
assert item_orphan.test_status == "failed"
|
|
assert item_orphan.upstream_key_id == 9999
|
|
|
|
# 确认没有调用过删除
|
|
assert len(deleted_accounts) == 0
|
|
|
|
# ── 2. 测试执行 ──
|
|
# 重置调用计数
|
|
test_calls.clear()
|
|
execute_res = execute_cleanup_invalid_accounts(wid=w.id, db=db_session)
|
|
assert execute_res.success is True
|
|
assert "成功清理 2 个" in execute_res.message
|
|
assert "跳过 1 个" in execute_res.message
|
|
|
|
# 验证是否重新跑了测试
|
|
assert "acc-k1" in test_calls
|
|
assert "acc-k2" in test_calls
|
|
assert "acc-orphan-remote" in test_calls
|
|
|
|
# 验证被删除的账号 (只有测试失败/可清理的才删：acc-k1 和 acc-orphan-remote)
|
|
assert len(deleted_accounts) == 2
|
|
assert "acc-k1" in deleted_accounts
|
|
assert "acc-orphan-remote" in deleted_accounts
|
|
assert "acc-k2" not in deleted_accounts
|
|
|
|
# 验证本地数据库的 DB key 清空逻辑：
|
|
db_session.refresh(k1)
|
|
db_session.refresh(k2)
|
|
db_session.refresh(k3)
|
|
|
|
# k1 应该被清空 imported 字段，但保留 status='orphaned' 及原始 error
|
|
assert k1.imported_website_id is None
|
|
assert k1.imported_account_id is None
|
|
assert k1.status == "orphaned"
|
|
assert k1.error == "original-error-k1"
|
|
|
|
# k2 由于测试成功跳过删除，DB 不变
|
|
assert k2.imported_website_id == w.id
|
|
assert k2.imported_account_id == "acc-k2"
|
|
assert k2.status == "orphaned"
|
|
|
|
|
|
def test_cleanup_list_accounts_failure(db_session, monkeypatch):
|
|
"""测试 list_accounts() 失败时，整个清理预览与执行直接阻断报错。"""
|
|
w = Website(
|
|
name="W1",
|
|
site_type="sub2api",
|
|
base_url="http://w1",
|
|
enabled=True,
|
|
auth_config_json="{}",
|
|
timeout_seconds=30
|
|
)
|
|
db_session.add(w)
|
|
db_session.commit()
|
|
|
|
class FailClient:
|
|
def __init__(self, **kwargs): pass
|
|
def __enter__(self): return self
|
|
def __exit__(self, *a): pass
|
|
def list_accounts(self):
|
|
return None # 模拟获取失败
|
|
|
|
monkeypatch.setattr("app.routers.websites._client", lambda row: FailClient())
|
|
|
|
preview_res = preview_cleanup_invalid_accounts(wid=w.id, db=db_session)
|
|
assert preview_res.success is False
|
|
assert "拉取远端账号列表失败" in preview_res.message
|
|
assert len(preview_res.items) == 0
|
|
|
|
execute_res = execute_cleanup_invalid_accounts(wid=w.id, db=db_session)
|
|
assert execute_res.success is False
|
|
assert "拉取远端账号列表失败" in execute_res.message
|
|
assert len(execute_res.items) == 0
|