fix: remove unused request params, strengthen admin route tests
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -161,7 +161,6 @@ async def new_config_form(request: Request):
|
|||||||
|
|
||||||
@app.post("/configs/new")
|
@app.post("/configs/new")
|
||||||
async def create_config(
|
async def create_config(
|
||||||
request: Request,
|
|
||||||
name: str = Form(...),
|
name: str = Form(...),
|
||||||
base_yaml: str = Form(...),
|
base_yaml: str = Form(...),
|
||||||
db: AsyncSession = Depends(get_db),
|
db: AsyncSession = Depends(get_db),
|
||||||
|
|||||||
@@ -146,7 +146,7 @@ async def test_create_config_redirects(http_client, db_session):
|
|||||||
data={"name": "MyConfig", "base_yaml": "proxies: []\nrules: []\nproxy-groups: []\n"},
|
data={"name": "MyConfig", "base_yaml": "proxies: []\nrules: []\nproxy-groups: []\n"},
|
||||||
follow_redirects=False,
|
follow_redirects=False,
|
||||||
)
|
)
|
||||||
assert resp.status_code in (302, 303)
|
assert resp.status_code == 303
|
||||||
|
|
||||||
result = await db_session.execute(select(Config).where(Config.name == "MyConfig"))
|
result = await db_session.execute(select(Config).where(Config.name == "MyConfig"))
|
||||||
config = result.scalar_one_or_none()
|
config = result.scalar_one_or_none()
|
||||||
@@ -179,7 +179,7 @@ async def test_update_config_base_yaml(http_client, db_session):
|
|||||||
data={"base_yaml": "proxies: []\nrules: []\n"},
|
data={"base_yaml": "proxies: []\nrules: []\n"},
|
||||||
follow_redirects=False,
|
follow_redirects=False,
|
||||||
)
|
)
|
||||||
assert resp.status_code in (302, 303)
|
assert resp.status_code == 303
|
||||||
|
|
||||||
await db_session.refresh(config)
|
await db_session.refresh(config)
|
||||||
assert "rules" in config.base_yaml
|
assert "rules" in config.base_yaml
|
||||||
@@ -190,16 +190,27 @@ async def test_delete_config(http_client, db_session):
|
|||||||
|
|
||||||
config = Config(name="c", token=str(uuid.uuid4()), base_yaml="proxies: []\n")
|
config = Config(name="c", token=str(uuid.uuid4()), base_yaml="proxies: []\n")
|
||||||
db_session.add(config)
|
db_session.add(config)
|
||||||
|
await db_session.flush()
|
||||||
|
|
||||||
|
sub = Subscription(config_id=config.id, name="s", url="https://example.com/sub")
|
||||||
|
log = ExportLog(config_id=config.id, node_count=1, success=True)
|
||||||
|
db_session.add(sub)
|
||||||
|
db_session.add(log)
|
||||||
await db_session.commit()
|
await db_session.commit()
|
||||||
config_id = config.id
|
config_id = config.id
|
||||||
|
sub_id = sub.id
|
||||||
|
log_id = log.id
|
||||||
|
|
||||||
with patch("main.write_and_reload_mihomo", new_callable=AsyncMock):
|
with patch("main.write_and_reload_mihomo", new_callable=AsyncMock):
|
||||||
resp = await http_client.post(f"/configs/{config_id}/delete", follow_redirects=False)
|
resp = await http_client.post(f"/configs/{config_id}/delete", follow_redirects=False)
|
||||||
assert resp.status_code in (302, 303)
|
assert resp.status_code == 303
|
||||||
|
|
||||||
result = await db_session.execute(select(Config).where(Config.id == config_id))
|
result = await db_session.execute(select(Config).where(Config.id == config_id))
|
||||||
assert result.scalar_one_or_none() is None
|
assert result.scalar_one_or_none() is None
|
||||||
|
|
||||||
|
assert (await db_session.get(Subscription, sub_id)) is None
|
||||||
|
assert (await db_session.get(ExportLog, log_id)) is None
|
||||||
|
|
||||||
|
|
||||||
async def test_add_subscription(http_client, db_session):
|
async def test_add_subscription(http_client, db_session):
|
||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
@@ -213,7 +224,7 @@ async def test_add_subscription(http_client, db_session):
|
|||||||
data={"name": "mysub", "url": "https://example.com/sub"},
|
data={"name": "mysub", "url": "https://example.com/sub"},
|
||||||
follow_redirects=False,
|
follow_redirects=False,
|
||||||
)
|
)
|
||||||
assert resp.status_code in (302, 303)
|
assert resp.status_code == 303
|
||||||
|
|
||||||
result = await db_session.execute(
|
result = await db_session.execute(
|
||||||
select(Subscription).where(Subscription.config_id == config.id)
|
select(Subscription).where(Subscription.config_id == config.id)
|
||||||
@@ -236,7 +247,7 @@ async def test_delete_subscription(http_client, db_session):
|
|||||||
sub_id = sub.id
|
sub_id = sub.id
|
||||||
|
|
||||||
resp = await http_client.post(f"/subscriptions/{sub_id}/delete", follow_redirects=False)
|
resp = await http_client.post(f"/subscriptions/{sub_id}/delete", follow_redirects=False)
|
||||||
assert resp.status_code in (302, 303)
|
assert resp.status_code == 303
|
||||||
|
|
||||||
result = await db_session.execute(select(Subscription).where(Subscription.id == sub_id))
|
result = await db_session.execute(select(Subscription).where(Subscription.id == sub_id))
|
||||||
assert result.scalar_one_or_none() is None
|
assert result.scalar_one_or_none() is None
|
||||||
@@ -256,6 +267,8 @@ async def test_logs_page_returns_200(http_client, db_session):
|
|||||||
|
|
||||||
|
|
||||||
async def test_force_refresh(http_client, db_session):
|
async def test_force_refresh(http_client, db_session):
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
config = Config(name="c", token=str(uuid.uuid4()), base_yaml="proxies: []\n")
|
config = Config(name="c", token=str(uuid.uuid4()), base_yaml="proxies: []\n")
|
||||||
db_session.add(config)
|
db_session.add(config)
|
||||||
await db_session.flush()
|
await db_session.flush()
|
||||||
@@ -268,4 +281,8 @@ async def test_force_refresh(http_client, db_session):
|
|||||||
mock_mc.refresh_and_collect = AsyncMock(return_value=[])
|
mock_mc.refresh_and_collect = AsyncMock(return_value=[])
|
||||||
resp = await http_client.post(f"/configs/{config.id}/refresh", follow_redirects=False)
|
resp = await http_client.post(f"/configs/{config.id}/refresh", follow_redirects=False)
|
||||||
|
|
||||||
assert resp.status_code in (302, 303)
|
assert resp.status_code == 303
|
||||||
|
assert mock_mc.refresh_and_collect.await_count == 1
|
||||||
|
|
||||||
|
logs = (await db_session.execute(select(ExportLog))).scalars().all()
|
||||||
|
assert logs == []
|
||||||
|
|||||||
Reference in New Issue
Block a user