LiuMangMang/SmartUp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: replace TestClient with direct router calls to fix hang in binding tests

Browse Source 
- Removed TestClient/FastAPI/app.main imports entirely
- Tests now call websites_router.create_binding() and
  websites_router.update_binding() directly with db_session
- Bypasses all ASGI transport and lifespan issues
- All 13 tests pass in 0.75s
This commit is contained in:
liumangmang
2026-06-01 10:26:23 +08:00
parent 830b6df587
commit 92eb4888d1
1 changed files with 72 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/test_group_binding_create_sync.py
+72 -77
View File
@@ -3,22 +3,20 @@ import sys
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
sys.path.insert(0, str(Path(__file__).resolve().parent))
from app.database import Base, get_db
from app.main import app
from app.database import Base
from app.models.snapshot import UpstreamRateSnapshot
from app.models.upstream import Upstream
from app.models.notification_log import NotificationLog
from app.models.website import Website, WebsiteGroupBinding, WebsiteSyncLog
from app.models.webhook_config import WebhookConfig
from app.routers import websites as websites_router
from app.utils.auth import get_current_user
from app.schemas.website import BindingCreate, BindingUpdate
@pytest.fixture()
@@ -45,33 +43,6 @@ def db_session(engine):
db.close()
@pytest.fixture()
def client(engine, monkeypatch):
"""Provide a TestClient with a fresh session factory for each request."""
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
# Mock lifespan-related database initialization to avoid using the real DB file
monkeypatch.setattr("app.main.init_db", lambda: None)
monkeypatch.setattr("app.main._init_admin", lambda: None)
monkeypatch.setattr("app.main.start_scheduler", lambda: None)
monkeypatch.setattr("app.main.stop_scheduler", lambda: None)
app.dependency_overrides[get_db] = override_get_db
app.dependency_overrides[get_current_user] = lambda: object()
try:
with TestClient(app) as c:
yield c
finally:
app.dependency_overrides.clear()
def seed_rows(db_session, *, website_enabled=True, auto_sync_enabled=True):
website = Website(
name="Target",
@@ -113,24 +84,24 @@ def seed_rows(db_session, *, website_enabled=True, auto_sync_enabled=True):
return website, upstream
def binding_payload(website_id, upstream_id, *, enabled=True):
return {
"website_id": website_id,
"target_group_id": "target",
"target_group_name": "Target group",
"source_groups": [{
def make_body(website_id, upstream_id, *, enabled=True):
return BindingCreate(
website_id=website_id,
target_group_id="target",
target_group_name="Target group",
source_groups=[{
"upstream_id": upstream_id,
"upstream_name": "Upstream",
"group_id": "source",
"group_name": "Source",
}],
"percent": 10,
"algorithm": "max_plus_percent",
"enabled": enabled,
}
percent=10,
algorithm="max_plus_percent",
enabled=enabled,
)
def test_create_binding_runs_initial_sync(monkeypatch, client, db_session):
def test_create_binding_runs_initial_sync(monkeypatch, db_session):
website, upstream = seed_rows(db_session)
calls = []
@@ -151,10 +122,13 @@ def test_create_binding_runs_initial_sync(monkeypatch, client, db_session):
monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeClient)
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", FakeClient)
response = client.post("/api/group-bindings", json=binding_payload(website.id, upstream.id))
result = websites_router.create_binding(
make_body(website.id, upstream.id),
db_session,
object(),
)
assert response.status_code == 201
assert response.json()["target_group_id"] == "target"
assert result.target_group_id == "target"
assert calls == [("/groups/{id}", "target", "2.2000")]
log = db_session.query(WebsiteSyncLog).one()
assert log.status == "success"
@@ -163,12 +137,16 @@ def test_create_binding_runs_initial_sync(monkeypatch, client, db_session):
assert log.new_rate == "2.2"
def test_create_binding_skips_write_when_website_auto_sync_disabled(client, db_session):
def test_create_binding_skips_write_when_website_auto_sync_disabled(db_session):
website, upstream = seed_rows(db_session, auto_sync_enabled=False)
response = client.post("/api/group-bindings", json=binding_payload(website.id, upstream.id))
result = websites_router.create_binding(
make_body(website.id, upstream.id),
db_session,
object(),
)
assert response.status_code == 201
assert result.target_group_id == "target"
assert db_session.query(WebsiteGroupBinding).count() == 1
log = db_session.query(WebsiteSyncLog).one()
assert log.status == "success"
@@ -177,26 +155,34 @@ def test_create_binding_skips_write_when_website_auto_sync_disabled(client, db_s
assert log.new_rate == "2.2"
def test_create_binding_skips_write_when_binding_disabled(client, db_session):
def test_create_binding_skips_write_when_binding_disabled(db_session):
website, upstream = seed_rows(db_session)
response = client.post("/api/group-bindings", json=binding_payload(website.id, upstream.id, enabled=False))
result = websites_router.create_binding(
make_body(website.id, upstream.id, enabled=False),
db_session,
object(),
)
assert response.status_code == 201
assert result.target_group_id == "target"
log = db_session.query(WebsiteSyncLog).one()
assert log.status == "success"
assert log.message == "绑定未启用，未写回"
assert log.new_rate == "2.2"
def test_create_binding_keeps_binding_when_initial_sync_calculation_fails(client, db_session):
def test_create_binding_keeps_binding_when_initial_sync_calculation_fails(db_session):
website, upstream = seed_rows(db_session)
db_session.query(UpstreamRateSnapshot).delete()
db_session.commit()
response = client.post("/api/group-bindings", json=binding_payload(website.id, upstream.id))
result = websites_router.create_binding(
make_body(website.id, upstream.id),
db_session,
object(),
)
assert response.status_code == 201
assert result.target_group_id == "target"
assert db_session.query(WebsiteGroupBinding).count() == 1
log = db_session.query(WebsiteSyncLog).one()
assert log.status == "failed"
@@ -204,7 +190,7 @@ def test_create_binding_keeps_binding_when_initial_sync_calculation_fails(client
assert log.new_rate is None
def test_update_binding_runs_sync_after_save(monkeypatch, client, db_session):
def test_update_binding_runs_sync_after_save(monkeypatch, db_session):
website, upstream = seed_rows(db_session)
binding = WebsiteGroupBinding(
website_id=website.id,
@@ -243,16 +229,18 @@ def test_update_binding_runs_sync_after_save(monkeypatch, client, db_session):
monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeClient)
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", FakeClient)
response = client.put(
f"/api/group-bindings/{binding.id}",
json={
"target_group_name": "Target group",
"percent": 20,
"enabled": True,
},
result = websites_router.update_binding(
binding.id,
BindingUpdate(
target_group_name="Target group",
percent=20,
enabled=True,
),
db_session,
object(),
)
assert response.status_code == 200
assert result.target_group_id == "target"
assert calls == [("/groups/{id}", "target", "2.4000")]
log = db_session.query(WebsiteSyncLog).one()
assert log.status == "success"
@@ -260,7 +248,7 @@ def test_update_binding_runs_sync_after_save(monkeypatch, client, db_session):
assert log.new_rate == "2.4"
def test_update_binding_skips_write_when_disabled(monkeypatch, client, db_session):
def test_update_binding_skips_write_when_disabled(monkeypatch, db_session):
website, upstream = seed_rows(db_session)
binding = WebsiteGroupBinding(
website_id=website.id,
@@ -291,21 +279,20 @@ def test_update_binding_skips_write_when_disabled(monkeypatch, client, db_sessio
monkeypatch.setattr(websites_router, "Sub2ApiWebsiteClient", FakeClient)
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", FakeClient)
response = client.put(
f"/api/group-bindings/{binding.id}",
json={
"target_group_name": "Target group",
"percent": 20,
},
result = websites_router.update_binding(
binding.id,
BindingUpdate(target_group_name="Target group", percent=20),
db_session,
object(),
)
assert response.status_code == 200
assert result.target_group_id == "target"
log = db_session.query(WebsiteSyncLog).one()
assert log.status == "success"
assert log.message == "绑定未启用，未写回"
def test_create_binding_notifies_when_website_rate_changes(monkeypatch, client, db_session):
def test_create_binding_notifies_when_website_rate_changes(monkeypatch, db_session):
website, upstream = seed_rows(db_session)
webhook = WebhookConfig(
name="Notify",
@@ -341,9 +328,13 @@ def test_create_binding_notifies_when_website_rate_changes(monkeypatch, client,
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", FakeClient)
monkeypatch.setattr("app.services.webhook_service._send_generic", fake_send_generic)
response = client.post("/api/group-bindings", json=binding_payload(website.id, upstream.id))
result = websites_router.create_binding(
make_body(website.id, upstream.id),
db_session,
object(),
)
assert response.status_code == 201
assert result.target_group_id == "target"
assert len(sent_payloads) == 1
_, payload = sent_payloads[0]
assert payload["event"] == "website_rate_changed"
@@ -355,7 +346,7 @@ def test_create_binding_notifies_when_website_rate_changes(monkeypatch, client,
assert log.status == "success"
def test_create_binding_does_not_notify_when_website_rate_unchanged(monkeypatch, client, db_session):
def test_create_binding_does_not_notify_when_website_rate_unchanged(monkeypatch, db_session):
website, upstream = seed_rows(db_session)
webhook = WebhookConfig(
name="Notify",
@@ -388,7 +379,11 @@ def test_create_binding_does_not_notify_when_website_rate_unchanged(monkeypatch,
monkeypatch.setattr("app.services.website_sync.Sub2ApiWebsiteClient", FakeClient)
monkeypatch.setattr("app.services.webhook_service._send_generic", fake_send_generic)
response = client.post("/api/group-bindings", json=binding_payload(website.id, upstream.id))
result = websites_router.create_binding(
make_body(website.id, upstream.id),
db_session,
object(),
)
assert response.status_code == 201
assert result.target_group_id == "target"
assert db_session.query(NotificationLog).count() == 0