58 lines
2.0 KiB
Python
58 lines
2.0 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from web.backend.database import Base
|
|
from web.backend.models import Account, LoginTask, User
|
|
from web.backend.routers.cookies import _order_cookie_tasks
|
|
|
|
|
|
class TestCustomCookieOrder:
|
|
def setup_method(self):
|
|
self.engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(self.engine)
|
|
self.db = Session(self.engine)
|
|
self.user = User(username="admin", password_hash="hash", role="super_admin")
|
|
self.db.add(self.user)
|
|
self.db.flush()
|
|
|
|
now = datetime.now(UTC)
|
|
accounts = {
|
|
name: Account(username=name, password="p", email="e", email_password="ep")
|
|
for name in ("account-a", "account-b", "account-c")
|
|
}
|
|
self.db.add_all(accounts.values())
|
|
self.db.flush()
|
|
# 完成时间特意与用户输入顺序相反。
|
|
for index, name in enumerate(("account-a", "account-b", "account-c")):
|
|
self.db.add(
|
|
LoginTask(
|
|
batch_id=f"batch-{name}",
|
|
account_id=accounts[name].id,
|
|
status="success",
|
|
created_by=self.user.id,
|
|
finished_at=now + timedelta(minutes=index),
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
def teardown_method(self):
|
|
self.db.close()
|
|
Base.metadata.drop_all(self.engine)
|
|
self.engine.dispose()
|
|
|
|
def test_custom_cookie_order_follows_input_not_finished_time(self):
|
|
selected_names = ["account-c", "account-a", "account-b"]
|
|
tasks = _order_cookie_tasks(
|
|
self.db.query(LoginTask)
|
|
.join(Account, LoginTask.account_id == Account.id)
|
|
.filter(Account.username.in_(selected_names)),
|
|
selected_names,
|
|
).all()
|
|
|
|
account_names = {
|
|
account.id: account.username for account in self.db.query(Account).all()
|
|
}
|
|
assert [account_names[task.account_id] for task in tasks] == selected_names
|