bruno
0861f1fdbf
- Scan de tous les appels `fetch` non-GET du front par préfixe : 12 préfixes n'ont AUCUN call site sans `X-CSRF-Token` → sortis de `EXCLUDED_PATHS` (`/db/`, `/workspace`, `/workspace/automations`, `/board/api/favorites`, `/api/pages`, `/api/recents`, `/api/notifications`, `/api/comments`, `/api/automations`, `/api/github`, `/api/admin`, `/api/onboarding`) - `welcome.html` : les 2 POST `/api/onboarding/*` reçoivent le header (`adminFetch` prouve que `/api/admin` était déjà couvert) - reste A19 (front à équiper) : `/api/workspace` (+`/api/workspaces`), `/api/local-workspace`, `/api/settings`, `/api/gitea`, `/api/agent` — 49 fetch - tests : helper `anon_csrf()` (anonyme + CSRF valide) pour isoler les 401 de route du 403 middleware — 4 tests d'anonymat ajustés - suite **1026/1026** · `ruff check app tests` OK
557 lines
23 KiB
Python
557 lines
23 KiB
Python
"""FlowDeck — v7.0.0 Automations v2 (steps) + Workers lite.
|
|
|
|
Covers migration 26, steps CRUD + validation + auth, trigger modes any/all,
|
|
chained actions with interpolation, condition/delay steps, new actions
|
|
(slack/email/forge_issue/agent_trigger, secret encryption), native DB button,
|
|
legacy no-double-run compat, workers CRUD/run/sandbox/budget/fork/usage/cron.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import secrets
|
|
|
|
import pytest
|
|
from conftest import anon, anon_csrf
|
|
|
|
from app.db import get_conn
|
|
from app.services import automations as auto_svc
|
|
|
|
# ── helpers ────────────────────────────────────────────────────────────────
|
|
|
|
def _login(client):
|
|
from app.auth.session import SessionManager
|
|
login = f"v70_{secrets.token_hex(4)}"
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"INSERT INTO users (login, full_name, email, is_admin) VALUES (?, 'V70', ?, 0)",
|
|
(login, f"{login}@test.com"),
|
|
)
|
|
uid = conn.execute("SELECT id FROM users WHERE login=?", (login,)).fetchone()["id"]
|
|
conn.commit()
|
|
session = SessionManager.create_session({"id": uid, "login": login})
|
|
return session, uid
|
|
|
|
|
|
def _cookies(session):
|
|
return {"flowdeck_session": session}
|
|
|
|
|
|
def _mkcollection(client, name="Tasks"):
|
|
r = client.post("/db/api", json={"name": name,
|
|
"schema": [{"name": "Status", "type": "text"}]})
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
def _mkrow(client, cid, title="Row"):
|
|
r = client.post(f"/db/{cid}/pages/api", json={"title": title})
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
def _mkauto(client, session, **kw):
|
|
body = {"name": "Auto", "trigger_type": "event", "event": "page.created",
|
|
"actions": []}
|
|
body.update(kw)
|
|
r = client.post("/workspace/automations", json=body, cookies=_cookies(session))
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
def _add_step(client, session, aid, kind, config, position=None):
|
|
body = {"kind": kind, "config": config}
|
|
if position is not None:
|
|
body["position"] = position
|
|
r = client.post(f"/workspace/automations/{aid}/steps", json=body,
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 200, r.text
|
|
return r.json()["id"]
|
|
|
|
|
|
def _runs(aid):
|
|
with get_conn() as conn:
|
|
return conn.execute(
|
|
"SELECT * FROM automation_runs WHERE automation_id=? ORDER BY id", (aid,)
|
|
).fetchall()
|
|
|
|
|
|
# ── migration ──────────────────────────────────────────────────────────────
|
|
|
|
def test_migration_26_tables(client):
|
|
with get_conn() as conn:
|
|
tables = {r[0] for r in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()}
|
|
for t in ("automation_steps", "workers", "worker_runs"):
|
|
assert t in tables
|
|
with get_conn() as conn:
|
|
auto_cols = {r[1] for r in conn.execute("PRAGMA table_info(automations)").fetchall()}
|
|
prop_cols = {r[1] for r in conn.execute(
|
|
"PRAGMA table_info(collection_properties)").fetchall()}
|
|
assert "trigger_mode" in auto_cols
|
|
assert "button_automation_id" in prop_cols
|
|
with get_conn() as conn:
|
|
v = conn.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]
|
|
assert v >= 26
|
|
|
|
|
|
# ── steps CRUD ─────────────────────────────────────────────────────────────
|
|
|
|
def test_steps_crud_and_order(client):
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
s1 = _add_step(client, session, aid, "trigger", {"event": "page.created"}, position=0)
|
|
s2 = _add_step(client, session, aid, "action",
|
|
{"type": "notify", "message": "hi"}, position=1)
|
|
r = client.get(f"/workspace/automations/{aid}/steps", cookies=_cookies(session))
|
|
assert [s["id"] for s in r.json()["steps"]] == [s1, s2]
|
|
r = client.put(f"/workspace/automations/steps/{s2}",
|
|
json={"kind": "action", "config": {"type": "notify", "message": "yo"}},
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 200
|
|
r = client.delete(f"/workspace/automations/steps/{s2}", cookies=_cookies(session))
|
|
assert r.status_code == 200
|
|
r = client.get(f"/workspace/automations/{aid}/steps", cookies=_cookies(session))
|
|
assert len(r.json()["steps"]) == 1
|
|
|
|
|
|
def test_steps_validation_and_auth(client):
|
|
anon_csrf(client)
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
r = client.post(f"/workspace/automations/{aid}/steps",
|
|
json={"kind": "nope", "config": {}}, cookies=_cookies(session))
|
|
assert r.status_code == 400
|
|
r = client.post(f"/workspace/automations/{aid}/steps",
|
|
json={"kind": "action", "config": {"type": "nope"}},
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 400
|
|
r = client.post(f"/workspace/automations/{aid}/steps",
|
|
json={"kind": "trigger", "config": {}})
|
|
assert r.status_code == 401
|
|
r = client.post("/workspace/automations/999999/steps",
|
|
json={"kind": "trigger", "config": {"event": "x"}},
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 404
|
|
|
|
|
|
def test_trigger_mode_endpoint(client):
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
r = client.put(f"/workspace/automations/{aid}/mode", json={"mode": "all"},
|
|
cookies=_cookies(session))
|
|
assert r.json()["trigger_mode"] == "all"
|
|
r = client.put(f"/workspace/automations/{aid}/mode", json={"mode": "sometimes"},
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 400
|
|
|
|
|
|
# ── multi-trigger any/all ──────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mode_any_two_triggers(client):
|
|
auto_svc.reset_all_pending()
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "trigger", {"event": "page.created"})
|
|
_add_step(client, session, aid, "trigger", {"event": "form.submitted"})
|
|
_add_step(client, session, aid, "action", {"type": "notify", "message": "fired"})
|
|
await auto_svc.fire_event("page.created", {"collection_id": 0})
|
|
await auto_svc.fire_event("form.submitted", {"collection_id": 0})
|
|
assert len([r for r in _runs(aid) if r["status"] == "fired"]) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mode_all_needs_every_trigger(client):
|
|
auto_svc.reset_all_pending()
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
client.put(f"/workspace/automations/{aid}/mode", json={"mode": "all"},
|
|
cookies=_cookies(session))
|
|
_add_step(client, session, aid, "trigger", {"event": "page.created"})
|
|
_add_step(client, session, aid, "trigger", {"event": "form.submitted"})
|
|
_add_step(client, session, aid, "action", {"type": "notify", "message": "both"})
|
|
await auto_svc.fire_event("page.created", {"collection_id": 0})
|
|
assert _runs(aid) == []
|
|
await auto_svc.fire_event("form.submitted", {"collection_id": 0})
|
|
assert len([r for r in _runs(aid) if r["status"] == "fired"]) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_form_submitted_trigger_end_to_end(client):
|
|
auto_svc.reset_all_pending()
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "trigger", {"event": "form.submitted"})
|
|
_add_step(client, session, aid, "action", {"type": "notify", "message": "form in"})
|
|
await auto_svc.fire_event("form.submitted", {"collection_id": 0, "row_id": 5})
|
|
runs = _runs(aid)
|
|
assert len(runs) == 1 and runs[0]["status"] == "fired"
|
|
|
|
|
|
# ── chains: order, interpolation, conditions, delay ────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_chained_actions_interpolation(client):
|
|
session, uid = _login(client)
|
|
cid = _mkcollection(client)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "trigger", {"event": "page.created"})
|
|
_add_step(client, session, aid, "action",
|
|
{"type": "create_page", "collection_id": cid, "title": "Copy of {{title}}"})
|
|
_add_step(client, session, aid, "action",
|
|
{"type": "notify", "message": "made [[Copy of {{title}}]]"})
|
|
res = await auto_svc.run_automation(aid, "manual",
|
|
{"collection_id": cid, "title": "Alpha"})
|
|
assert res["status"] == "fired"
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT title FROM collection_pages WHERE collection_id=?",
|
|
(cid,)).fetchone()
|
|
assert row and row["title"] == "Copy of Alpha"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_condition_step_blocks(client):
|
|
session, _ = _login(client)
|
|
cid = _mkcollection(client)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "trigger", {"event": "page.created"})
|
|
_add_step(client, session, aid, "condition",
|
|
{"property": "Status", "op": "eq", "value": "Done"})
|
|
_add_step(client, session, aid, "action", {"type": "notify", "message": "x"})
|
|
res = await auto_svc.run_automation(
|
|
aid, "event", {"collection_id": cid, "properties": {"Status": "Todo"}})
|
|
assert res["status"] == "skipped"
|
|
assert _runs(aid)[0]["status"] == "skipped"
|
|
res = await auto_svc.run_automation(
|
|
aid, "event", {"collection_id": cid, "properties": {"Status": "Done"}})
|
|
assert res["status"] == "fired"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delay_step(client):
|
|
import time as _t
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "action", {"type": "notify", "message": "a"})
|
|
_add_step(client, session, aid, "delay", {"seconds": 1})
|
|
_add_step(client, session, aid, "action", {"type": "notify", "message": "b"})
|
|
start = _t.time()
|
|
res = await auto_svc.run_automation(aid, "manual", {})
|
|
assert res["status"] == "fired"
|
|
assert _t.time() - start >= 1.0
|
|
assert "delay 1s" in res["detail"]
|
|
|
|
|
|
# ── new actions ────────────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_slack_action(client, monkeypatch):
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
seen = {}
|
|
async def fake(url, text):
|
|
seen["url"] = url
|
|
seen["text"] = text
|
|
return "slack → (200)"
|
|
monkeypatch.setattr(auto_svc, "_post_slack", fake)
|
|
_add_step(client, session, aid, "action",
|
|
{"type": "slack", "webhook_url": "https://hooks.test/x", "text": "Hi {{title}}"})
|
|
res = await auto_svc.run_automation(aid, "manual", {"title": "Bob"})
|
|
assert res["status"] == "fired"
|
|
assert seen == {"url": "https://hooks.test/x", "text": "Hi Bob"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_slack_secret_encrypted_at_rest(client, monkeypatch):
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
seen = {}
|
|
async def fake(url, text):
|
|
seen["url"] = url
|
|
return "ok"
|
|
monkeypatch.setattr(auto_svc, "_post_slack", fake)
|
|
_add_step(client, session, aid, "action",
|
|
{"type": "slack", "webhook_url": "https://hooks.test/secret"})
|
|
with get_conn() as conn:
|
|
stored = conn.execute(
|
|
"SELECT config_json FROM automation_steps WHERE automation_id=?", (aid,)).fetchone()[0]
|
|
assert "hooks.test/secret" not in stored # encrypted at rest
|
|
res = await auto_svc.run_automation(aid, "manual", {})
|
|
assert res["status"] == "fired"
|
|
assert seen["url"] == "https://hooks.test/secret" # decrypted on execute
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_email_action_no_smtp_skips(client):
|
|
session, uid = _login(client)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "action",
|
|
{"type": "email", "to": f"user:{uid}", "subject": "S", "body": "B"})
|
|
res = await auto_svc.run_automation(aid, "manual", {"created_by": uid})
|
|
# user has email but SMTP unconfigured in tests → skipped, not error
|
|
assert res["status"] == "fired"
|
|
assert "email" in res["detail"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_forge_issue_action(client, monkeypatch):
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
async def fake(provider, owner, repo, title, body, labels=None, user_id=None):
|
|
return f"{provider} issue #7 in {owner}/{repo}"
|
|
monkeypatch.setattr(auto_svc, "_create_forge_issue", fake)
|
|
_add_step(client, session, aid, "action",
|
|
{"type": "forge_issue", "provider": "gitea", "owner": "o", "repo": "r",
|
|
"title": "Bug {{title}}"})
|
|
res = await auto_svc.run_automation(aid, "manual", {"title": "X"})
|
|
assert res["status"] == "fired"
|
|
assert "gitea issue #7 in o/r" in res["detail"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_forge_issue_needs_token(client):
|
|
session, uid = _login(client)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "action",
|
|
{"type": "forge_issue", "provider": "github", "owner": "o", "repo": "r",
|
|
"title": "T"})
|
|
res = await auto_svc.run_automation(aid, "manual", {"created_by": uid})
|
|
assert res["status"] == "error"
|
|
assert "token" in res["detail"].lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_trigger_action(client, monkeypatch):
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
async def fake(agent_id, user_id, workspace_id, message, context):
|
|
return f"agent {agent_id} ran: {message}"
|
|
monkeypatch.setattr(auto_svc, "_run_linked_agent", fake)
|
|
with get_conn() as conn:
|
|
conn.execute("INSERT INTO agents (name, system_instructions) VALUES ('A', 'Do X')")
|
|
agid = conn.execute("SELECT id FROM agents WHERE name='A'").fetchone()["id"]
|
|
conn.commit()
|
|
_add_step(client, session, aid, "action",
|
|
{"type": "agent_trigger", "agent_id": agid, "message": "go"})
|
|
res = await auto_svc.run_automation(aid, "manual", {})
|
|
assert res["status"] == "fired"
|
|
assert f"agent {agid} ran: go" in res["detail"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent_trigger_missing_agent_errors(client):
|
|
session, _ = _login(client)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "action", {"type": "agent_trigger", "agent_id": 999999})
|
|
res = await auto_svc.run_automation(aid, "manual", {})
|
|
assert res["status"] == "error"
|
|
|
|
|
|
# ── native button ──────────────────────────────────────────────────────────
|
|
|
|
def _make_button(client, cid, aid, name="Ship it"):
|
|
r = client.post(f"/db/{cid}/properties/api",
|
|
json={"name": name, "prop_type": "button"})
|
|
assert r.status_code == 200, r.text
|
|
pid = r.json()["id"]
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE collection_properties SET button_automation_id=? WHERE id=?",
|
|
(aid, pid))
|
|
conn.commit()
|
|
return pid
|
|
|
|
|
|
def test_press_button_runs_automation(client):
|
|
session, _ = _login(client)
|
|
cid = _mkcollection(client)
|
|
rid = _mkrow(client, cid)
|
|
aid = _mkauto(client, session)
|
|
_add_step(client, session, aid, "action", {"type": "notify", "message": "shipped"})
|
|
prop_id = _make_button(client, cid, aid)
|
|
r = client.post("/api/automations/press-button",
|
|
json={"collection_id": cid, "row_id": rid, "property_id": prop_id})
|
|
assert r.status_code == 200, r.text
|
|
assert r.json()["status"] == "fired"
|
|
assert len([x for x in _runs(aid) if x["status"] == "fired"]) == 1
|
|
|
|
|
|
def test_press_button_validation(client):
|
|
session, _ = _login(client)
|
|
cid = _mkcollection(client)
|
|
rid = _mkrow(client, cid)
|
|
r = client.post(f"/db/{cid}/properties/api",
|
|
json={"name": "Plain", "prop_type": "text"})
|
|
text_pid = r.json()["id"]
|
|
r = client.post("/api/automations/press-button",
|
|
json={"collection_id": cid, "row_id": rid, "property_id": text_pid})
|
|
assert r.status_code == 400 # not a button
|
|
r = client.post("/api/automations/press-button",
|
|
json={"collection_id": cid, "row_id": rid, "property_id": 999999})
|
|
assert r.status_code == 400 # unknown
|
|
aid = _mkauto(client, session)
|
|
unlinked = _make_button(client, cid, aid, name="Unlinked")
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE collection_properties SET button_automation_id=NULL WHERE id=?",
|
|
(unlinked,))
|
|
conn.commit()
|
|
r = client.post("/api/automations/press-button",
|
|
json={"collection_id": cid, "row_id": rid, "property_id": unlinked})
|
|
assert r.status_code == 400 # no linked automation
|
|
|
|
|
|
# ── legacy compat: no double run ───────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_legacy_automation_single_run(client):
|
|
session, _ = _login(client)
|
|
cid = _mkcollection(client)
|
|
r = client.post("/workspace/automations",
|
|
json={"name": "Legacy", "trigger_type": "event", "event": "page.created",
|
|
"collection_id": cid,
|
|
"actions": [{"type": "notify", "message": "legacy"}]},
|
|
cookies=_cookies(session))
|
|
aid = r.json()["id"]
|
|
await auto_svc.fire_event("page.created", {"collection_id": cid})
|
|
assert len([x for x in _runs(aid) if x["status"] == "fired"]) == 1
|
|
|
|
|
|
# ── workers ────────────────────────────────────────────────────────────────
|
|
|
|
def _mkworker(client, session, **kw):
|
|
body = {"name": "W", "code_py": "result['x'] = 1"}
|
|
body.update(kw)
|
|
r = client.post("/api/v2/workers", json=body, cookies=_cookies(session))
|
|
assert r.status_code == 201, r.text
|
|
return r.json()
|
|
|
|
|
|
def test_workers_crud_and_auth(client):
|
|
anon(client)
|
|
session, _ = _login(client)
|
|
w = _mkworker(client, session, name="Hello")
|
|
assert w["slug"].startswith("hello") or w["slug"]
|
|
wid = w["id"]
|
|
r = client.get(f"/api/v2/workers/{wid}", cookies=_cookies(session))
|
|
assert r.status_code == 200
|
|
assert r.json()["code_py"] == "result['x'] = 1" # owner sees code
|
|
r = client.patch(f"/api/v2/workers/{wid}", json={"shared": True},
|
|
cookies=_cookies(session))
|
|
assert r.json()["shared"] == 1
|
|
r = client.get("/api/v2/workers", cookies=_cookies(session))
|
|
assert "X-Total-Count" in r.headers
|
|
r = client.post("/api/v2/workers", json={"name": "X", "code_py": "x = 1"})
|
|
assert r.status_code == 401
|
|
r = client.delete(f"/api/v2/workers/{wid}", cookies=_cookies(session))
|
|
assert r.status_code == 200
|
|
|
|
|
|
def test_worker_rejects_bad_code(client):
|
|
session, _ = _login(client)
|
|
r = client.post("/api/v2/workers",
|
|
json={"name": "Bad", "code_py": "import os\nresult['x']=1"},
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 400
|
|
r = client.post("/api/v2/workers",
|
|
json={"name": "Bad2", "code_py": "open('/etc/passwd').read()"},
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 400
|
|
r = client.post("/api/v2/workers",
|
|
json={"name": "Bad3", "code_py": "def broken(:\n pass"},
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 400
|
|
|
|
|
|
def test_worker_run_ok(client):
|
|
session, _ = _login(client)
|
|
w = _mkworker(client, session, code_py="log('hello'); result['total'] = sum([1, 2, 3])")
|
|
r = client.post(f"/api/v2/workers/{w['id']}/run", json={"ctx": {}},
|
|
cookies=_cookies(session))
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["status"] == "ok"
|
|
assert body["result"] == {"total": 6}
|
|
r = client.get(f"/api/v2/workers/{w['id']}/runs", cookies=_cookies(session))
|
|
assert r.json()["runs"][0]["status"] == "ok"
|
|
|
|
|
|
def test_worker_run_error(client):
|
|
session, _ = _login(client)
|
|
w = _mkworker(client, session, code_py="1 / 0")
|
|
r = client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(session))
|
|
assert r.json()["status"] == "error"
|
|
assert "ZeroDivision" in r.json()["error"]
|
|
|
|
|
|
def test_worker_run_timeout(client, monkeypatch):
|
|
from app.services import workers as wsvc
|
|
monkeypatch.setattr(wsvc, "RUN_TIMEOUT_S", 1)
|
|
session, _ = _login(client)
|
|
w = _mkworker(client, session, code_py="while True:\n pass")
|
|
r = client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(session))
|
|
assert r.json()["status"] == "timeout"
|
|
|
|
|
|
def test_worker_budget_enforced(client):
|
|
session, _ = _login(client)
|
|
w = _mkworker(client, session, code_py="result['x'] = 1")
|
|
with get_conn() as conn:
|
|
conn.execute("UPDATE workers SET daily_budget_s=1 WHERE id=?", (w["id"],))
|
|
conn.execute("INSERT INTO worker_runs (worker_id, status, duration_ms)"
|
|
" VALUES (?, 'ok', 60000)", (w["id"],))
|
|
conn.commit()
|
|
r = client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(session))
|
|
assert r.status_code == 429
|
|
|
|
|
|
def test_worker_fork_and_privacy(client):
|
|
s1, _ = _login(client)
|
|
s2, _ = _login(client)
|
|
w = _mkworker(client, s1, name="Private")
|
|
r = client.post(f"/api/v2/workers/{w['id']}/fork", cookies=_cookies(s2))
|
|
assert r.status_code == 403 # private
|
|
client.patch(f"/api/v2/workers/{w['id']}", json={"shared": True},
|
|
cookies=_cookies(s1))
|
|
r = client.post(f"/api/v2/workers/{w['id']}/fork", cookies=_cookies(s2))
|
|
assert r.status_code == 201
|
|
assert r.json()["from"] == w["id"]
|
|
|
|
|
|
def test_worker_private_run_forbidden(client):
|
|
s1, _ = _login(client)
|
|
s2, _ = _login(client)
|
|
w = _mkworker(client, s1)
|
|
r = client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(s2))
|
|
assert r.status_code == 403
|
|
|
|
|
|
def test_workers_usage(client):
|
|
session, _ = _login(client)
|
|
w = _mkworker(client, session)
|
|
client.post(f"/api/v2/workers/{w['id']}/run", json={}, cookies=_cookies(session))
|
|
r = client.get("/api/v2/workers-usage", cookies=_cookies(session))
|
|
assert r.json()["used_seconds_today"] >= 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_due_workers(client):
|
|
from app.services import workers as wsvc
|
|
session, _ = _login(client)
|
|
w = _mkworker(client, session, code_py="result['cron'] = True",
|
|
schedule_cron="@hourly")
|
|
fired = await wsvc.run_due_workers()
|
|
assert fired >= 1
|
|
with get_conn() as conn:
|
|
row = conn.execute("SELECT status FROM worker_runs WHERE worker_id=? ORDER BY id DESC",
|
|
(w["id"],)).fetchone()
|
|
assert row and row["status"] == "ok"
|
|
|
|
|
|
def test_worker_code_hidden_from_strangers(client):
|
|
s1, _ = _login(client)
|
|
s2, _ = _login(client)
|
|
w = _mkworker(client, s1)
|
|
client.patch(f"/api/v2/workers/{w['id']}", json={"shared": True},
|
|
cookies=_cookies(s1))
|
|
r = client.get(f"/api/v2/workers/{w['id']}", cookies=_cookies(s2))
|
|
assert r.status_code == 200
|
|
assert "code_py" not in r.json() # no include_code for non-owner
|