|
# retoor <retoor@molodetz.nl>
|
|
|
|
import time
|
|
|
|
import requests
|
|
|
|
from tests.conftest import BASE_URL
|
|
|
|
JSON_dt = {"Accept": "application/json"}
|
|
_counter_dt = [0]
|
|
|
|
|
|
def _unique_dt(prefix="dt"):
|
|
_counter_dt[0] += 1
|
|
return f"{prefix}{int(time.time() * 1000)}{_counter_dt[0]}"
|
|
|
|
|
|
def _member_dt():
|
|
name = _unique_dt("dtmem")
|
|
session = requests.Session()
|
|
session.post(
|
|
f"{BASE_URL}/auth/signup",
|
|
data={
|
|
"username": name,
|
|
"email": f"{name}@t.dev",
|
|
"password": "secret123",
|
|
"confirm_password": "secret123",
|
|
},
|
|
allow_redirects=True,
|
|
)
|
|
return session, name
|
|
|
|
|
|
def _admin_dt(seeded_db):
|
|
session = requests.Session()
|
|
creds = seeded_db["alice"]
|
|
session.post(
|
|
f"{BASE_URL}/auth/login",
|
|
data={"email": creds["email"], "password": creds["password"]},
|
|
allow_redirects=True,
|
|
)
|
|
return session
|
|
|
|
|
|
def _seed_task(owner_uid, enabled=True):
|
|
from devplacepy.database import db
|
|
from devplacepy.services.devii.tasks.schedule import now_utc, to_iso
|
|
from devplacepy.utils import generate_uid
|
|
|
|
uid = generate_uid()
|
|
db["devii_tasks"].insert(
|
|
{
|
|
"uid": uid,
|
|
"owner_kind": "user",
|
|
"owner_id": owner_uid,
|
|
"label": _unique_dt("task"),
|
|
"prompt": "work",
|
|
"enabled": enabled,
|
|
"status": "pending" if enabled else "disabled",
|
|
"kind": "interval",
|
|
"every_seconds": 900,
|
|
"max_runs": 10,
|
|
"run_count": 3,
|
|
"failure_count": 0,
|
|
"created_at": to_iso(now_utc()),
|
|
"next_run_at": to_iso(now_utc()),
|
|
"expires_at": None,
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
return uid
|
|
|
|
|
|
def _admin_uid(seeded_db):
|
|
from devplacepy.database import get_table
|
|
|
|
return get_table("users").find_one(username=seeded_db["alice"]["username"])["uid"]
|
|
|
|
|
|
def test_devii_tasks_requires_admin(seeded_db):
|
|
member, _ = _member_dt()
|
|
response = member.get(f"{BASE_URL}/admin/devii-tasks", allow_redirects=False)
|
|
assert response.status_code in (302, 303)
|
|
|
|
|
|
def test_devii_tasks_lists_a_task_with_its_owner(seeded_db):
|
|
uid = _seed_task(_admin_uid(seeded_db))
|
|
admin = _admin_dt(seeded_db)
|
|
body = admin.get(f"{BASE_URL}/admin/devii-tasks", headers=JSON_dt).json()
|
|
assert body["admin_section"] == "devii-tasks"
|
|
entry = next(item for item in body["items"] if item["uid"] == uid)
|
|
assert entry["owner"] == seeded_db["alice"]["username"]
|
|
assert entry["owner_is_admin"] is True
|
|
assert entry["schedule"] == "every 900s"
|
|
assert entry["run_count"] == 3
|
|
|
|
|
|
def test_devii_tasks_reports_the_configured_bounds(seeded_db):
|
|
admin = _admin_dt(seeded_db)
|
|
body = admin.get(f"{BASE_URL}/admin/devii-tasks", headers=JSON_dt).json()
|
|
for key in ("max_concurrent", "max_per_owner", "max_failures", "idle_days"):
|
|
assert key in body["limits"]
|
|
|
|
|
|
def test_devii_tasks_inactive_filter_excludes_enabled(seeded_db):
|
|
enabled = _seed_task(_admin_uid(seeded_db))
|
|
disabled = _seed_task(_admin_uid(seeded_db), enabled=False)
|
|
admin = _admin_dt(seeded_db)
|
|
body = admin.get(
|
|
f"{BASE_URL}/admin/devii-tasks", headers=JSON_dt, params={"state": "inactive"}
|
|
).json()
|
|
uids = {item["uid"] for item in body["items"]}
|
|
assert disabled in uids
|
|
assert enabled not in uids
|
|
|
|
|
|
def test_admin_can_disable_a_task(seeded_db):
|
|
from devplacepy.database import db
|
|
|
|
uid = _seed_task(_admin_uid(seeded_db))
|
|
admin = _admin_dt(seeded_db)
|
|
response = admin.post(
|
|
f"{BASE_URL}/admin/devii-tasks/{uid}/disable",
|
|
headers=JSON_dt,
|
|
allow_redirects=False,
|
|
)
|
|
assert response.status_code in (200, 302, 303)
|
|
row = db["devii_tasks"].find_one(uid=uid)
|
|
assert not row["enabled"]
|
|
assert row["status"] == "disabled"
|
|
|
|
|
|
def test_admin_can_delete_a_task(seeded_db):
|
|
from devplacepy.database import db
|
|
|
|
uid = _seed_task(_admin_uid(seeded_db))
|
|
admin = _admin_dt(seeded_db)
|
|
response = admin.post(
|
|
f"{BASE_URL}/admin/devii-tasks/{uid}/delete",
|
|
headers=JSON_dt,
|
|
allow_redirects=False,
|
|
)
|
|
assert response.status_code in (200, 302, 303)
|
|
assert db["devii_tasks"].find_one(uid=uid, deleted_at=None) is None
|
|
|
|
|
|
def test_member_cannot_disable_a_task(seeded_db):
|
|
from devplacepy.database import db
|
|
|
|
uid = _seed_task(_admin_uid(seeded_db))
|
|
member, _ = _member_dt()
|
|
response = member.post(
|
|
f"{BASE_URL}/admin/devii-tasks/{uid}/disable", allow_redirects=False
|
|
)
|
|
assert response.status_code in (302, 303)
|
|
assert db["devii_tasks"].find_one(uid=uid)["enabled"]
|