149 lines
5.0 KiB
Python
149 lines
5.0 KiB
Python
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base
|
|
from app.models.website import Website, WebsiteGroupBinding
|
|
from app.routers.websites import create_website_group, update_website_group
|
|
from app.schemas.website import WebsiteGroupCreate, WebsiteGroupUpdate
|
|
|
|
|
|
@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)
|
|
|
|
|
|
def test_create_and_update_website_group(db_session, monkeypatch):
|
|
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()
|
|
db_session.refresh(w)
|
|
|
|
b = WebsiteGroupBinding(
|
|
website_id=w.id,
|
|
target_group_id="g/123",
|
|
target_group_name="OldName",
|
|
source_groups_json="[]",
|
|
enabled=True
|
|
)
|
|
db_session.add(b)
|
|
db_session.commit()
|
|
db_session.refresh(b)
|
|
|
|
called_create = []
|
|
called_update = []
|
|
|
|
class FakeClient:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
def __enter__(self):
|
|
return self
|
|
def __exit__(self, *args):
|
|
pass
|
|
def create_group(self, body, endpoint):
|
|
called_create.append((body, endpoint))
|
|
return {"id": "new-gid", "name": body["name"], "description": body.get("description")}
|
|
def update_group(self, endpoint_template, group_id, body):
|
|
called_update.append((endpoint_template, group_id, body))
|
|
return {"id": group_id, "name": body["name"], "description": body.get("description")}
|
|
|
|
monkeypatch.setattr("app.routers.websites._client", lambda website: FakeClient())
|
|
|
|
# 1. Test create_website_group
|
|
body_create = WebsiteGroupCreate(name="NewGroup", description="NewDesc")
|
|
res_create = create_website_group(wid=w.id, body=body_create, db=db_session)
|
|
assert res_create["id"] == "new-gid"
|
|
assert res_create["name"] == "NewGroup"
|
|
assert res_create["description"] == "NewDesc"
|
|
assert len(called_create) == 1
|
|
assert called_create[0][0] == {"name": "NewGroup", "description": "NewDesc"}
|
|
|
|
# 2. Test update_website_group
|
|
body_update = WebsiteGroupUpdate(name="UpdatedGroup", description="UpdatedDesc")
|
|
res_update = update_website_group(wid=w.id, group_id="g/123", body=body_update, db=db_session)
|
|
assert res_update["id"] == "g/123"
|
|
assert res_update["name"] == "UpdatedGroup"
|
|
assert res_update["description"] == "UpdatedDesc"
|
|
assert len(called_update) == 1
|
|
assert called_update[0][1] == "g/123"
|
|
assert called_update[0][2] == {"name": "UpdatedGroup", "description": "UpdatedDesc"}
|
|
|
|
# Verify that local binding was updated
|
|
db_session.refresh(b)
|
|
assert b.target_group_name == "UpdatedGroup"
|
|
|
|
|
|
def test_update_website_group_route_handles_slash(db_session, monkeypatch):
|
|
from fastapi.testclient import TestClient
|
|
from app.main import app
|
|
from app.database import get_db
|
|
from app.utils.auth import get_current_user
|
|
|
|
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()
|
|
db_session.refresh(w)
|
|
|
|
called_update = []
|
|
|
|
class FakeClient:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
def __enter__(self):
|
|
return self
|
|
def __exit__(self, *args):
|
|
pass
|
|
def update_group(self, endpoint_template, group_id, body):
|
|
called_update.append((endpoint_template, group_id, body))
|
|
return {"id": group_id, "name": body["name"], "description": body.get("description")}
|
|
|
|
monkeypatch.setattr("app.routers.websites._client", lambda website: FakeClient())
|
|
|
|
# Set dependency overrides
|
|
app.dependency_overrides[get_db] = lambda: db_session
|
|
app.dependency_overrides[get_current_user] = lambda: None
|
|
|
|
try:
|
|
client = TestClient(app)
|
|
# request URL with urlencoded slash: g%2F123
|
|
resp = client.put(
|
|
f"/api/websites/{w.id}/groups/g%2F123",
|
|
json={"name": "UpdatedGroup", "description": "UpdatedDesc"}
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["id"] == "g/123"
|
|
assert data["name"] == "UpdatedGroup"
|
|
assert len(called_update) == 1
|
|
assert called_update[0][1] == "g/123"
|
|
finally:
|
|
app.dependency_overrides.clear()
|