95 lines
3.3 KiB
Python
95 lines
3.3 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="g123",
|
|
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="g123", body=body_update, db=db_session)
|
|
assert res_update["id"] == "g123"
|
|
assert res_update["name"] == "UpdatedGroup"
|
|
assert res_update["description"] == "UpdatedDesc"
|
|
assert len(called_update) == 1
|
|
assert called_update[0][1] == "g123"
|
|
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"
|