# retoor <retoor@molodetz.nl>
import time
import requests
from tests.conftest import BASE_URL
from devplacepy.database import (
get_table,
get_primary_admin_uid,
invalidate_admins_cache,
refresh_snapshot,
)
from devplacepy.utils import clear_user_cache
JSON = {"Accept": "application/json"}
_counter = [0]
def _signup():
_counter[0] += 1
name = f"pcm{int(time.time() * 1000)}{_counter[0]}"
requests.post(
f"{BASE_URL}/auth/signup",
data={
"username": name,
"email": f"{name}@t.dev",
"password": "secret123",
"confirm_password": "secret123",
},
allow_redirects=True,
)
row = get_table("users").find_one(username=name)
return row["uid"], row["api_key"]
def _make_admin():
uid, key = _signup()
get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
clear_user_cache(uid)
invalidate_admins_cache()
return uid, key
def _primary_admin_key():
refresh_snapshot()
uid = get_primary_admin_uid()
return get_table("users").find_one(uid=uid)["api_key"]
def _headers(key):
return {**JSON, "X-API-KEY": key}
def _create_project(key, title, is_private=False):
data = {
"title": title,
"description": "project container manage isolation test",
"project_type": "software",
"status": "In Development",
}
if is_private:
data["is_private"] = "on"
r = requests.post(f"{BASE_URL}/projects/create", headers=_headers(key), data=data)
assert r.status_code == 200, r.text
return r.json()["data"]
def _insert_instance(project_uid, created_by):
uid = f"pcm-{int(time.time() * 1000000)}"
get_table("instances").insert(
{
"uid": uid,
"slug": uid,
"name": "manage-test",
"project_uid": project_uid,
"created_by": created_by,
"owner_uid": "",
"status": "created",
"container_id": "",
"workspace_dir": "",
"restart_policy": "never",
"deleted_at": None,
"deleted_by": None,
}
)
refresh_snapshot()
return uid
def _cleanup(inst_uid):
get_table("instances").delete(uid=inst_uid)
get_table("instance_schedules").delete(instance_uid=inst_uid)
refresh_snapshot()
def _url(slug, inst_uid, tail=""):
base = f"{BASE_URL}/projects/{slug}/containers/instances/{inst_uid}"
return f"{base}{tail}" if tail else base
def test_delete_denied_for_non_owner_admin_on_public_project(app_server):
owner_uid, owner_key = _make_admin()
_, other_key = _make_admin()
project = _create_project(owner_key, "Project Manage Delete Denied", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(_url(project["slug"], inst_uid, "/delete"), headers=_headers(other_key))
assert r.status_code == 403
assert get_table("instances").find_one(uid=inst_uid) is not None
finally:
_cleanup(inst_uid)
def test_delete_allowed_for_owner(app_server):
owner_uid, owner_key = _make_admin()
project = _create_project(owner_key, "Project Manage Delete Owner", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(_url(project["slug"], inst_uid, "/delete"), headers=_headers(owner_key))
assert r.status_code == 200
finally:
_cleanup(inst_uid)
def test_lifecycle_actions_denied_for_non_owner_admin(app_server):
owner_uid, owner_key = _make_admin()
_, other_key = _make_admin()
project = _create_project(owner_key, "Project Manage Lifecycle Denied", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
for action in ("start", "stop", "pause", "resume", "restart"):
r = requests.post(_url(project["slug"], inst_uid, f"/{action}"), headers=_headers(other_key))
assert r.status_code == 403, action
finally:
_cleanup(inst_uid)
def test_lifecycle_actions_allowed_for_owner(app_server):
owner_uid, owner_key = _make_admin()
project = _create_project(owner_key, "Project Manage Lifecycle Owner", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
for action in ("start", "stop", "pause", "resume", "restart"):
r = requests.post(_url(project["slug"], inst_uid, f"/{action}"), headers=_headers(owner_key))
assert r.status_code == 200, action
finally:
_cleanup(inst_uid)
def test_exec_denied_for_non_owner_admin(app_server):
owner_uid, owner_key = _make_admin()
_, other_key = _make_admin()
project = _create_project(owner_key, "Project Manage Exec Denied", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(
_url(project["slug"], inst_uid, "/exec"),
headers=_headers(other_key),
data={"command": "ls"},
)
assert r.status_code == 403
finally:
_cleanup(inst_uid)
def test_exec_allowed_for_owner_reaches_running_check(app_server):
owner_uid, owner_key = _make_admin()
project = _create_project(owner_key, "Project Manage Exec Owner", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(
_url(project["slug"], inst_uid, "/exec"),
headers=_headers(owner_key),
data={"command": "ls"},
)
assert r.status_code == 400
assert "not running" in r.json()["error"]["message"]
finally:
_cleanup(inst_uid)
def test_sync_denied_for_non_owner_admin(app_server):
owner_uid, owner_key = _make_admin()
_, other_key = _make_admin()
project = _create_project(owner_key, "Project Manage Sync Denied", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(_url(project["slug"], inst_uid, "/sync"), headers=_headers(other_key))
assert r.status_code == 403
finally:
_cleanup(inst_uid)
def test_sync_allowed_for_owner_reaches_workspace_check(app_server):
owner_uid, owner_key = _make_admin()
project = _create_project(owner_key, "Project Manage Sync Owner", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(_url(project["slug"], inst_uid, "/sync"), headers=_headers(owner_key))
assert r.status_code == 400
assert "workspace" in r.json()["error"]["message"]
finally:
_cleanup(inst_uid)
def test_schedule_create_denied_for_non_owner_admin(app_server):
owner_uid, owner_key = _make_admin()
_, other_key = _make_admin()
project = _create_project(owner_key, "Project Manage Schedule Denied", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(
_url(project["slug"], inst_uid, "/schedules"),
headers=_headers(other_key),
data={"action": "start", "kind": "cron", "cron": "0 3 * * *"},
)
assert r.status_code == 403
assert get_table("instance_schedules").find_one(instance_uid=inst_uid) is None
finally:
_cleanup(inst_uid)
def test_schedule_create_allowed_for_owner(app_server):
owner_uid, owner_key = _make_admin()
project = _create_project(owner_key, "Project Manage Schedule Owner", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(
_url(project["slug"], inst_uid, "/schedules"),
headers=_headers(owner_key),
data={"action": "start", "kind": "cron", "cron": "0 3 * * *"},
)
assert r.status_code == 200
assert get_table("instance_schedules").find_one(instance_uid=inst_uid) is not None
finally:
_cleanup(inst_uid)
def test_schedule_delete_denied_for_non_owner_admin(app_server):
owner_uid, owner_key = _make_admin()
_, other_key = _make_admin()
project = _create_project(owner_key, "Project Manage Schedule Delete Denied", is_private=False)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
created = requests.post(
_url(project["slug"], inst_uid, "/schedules"),
headers=_headers(owner_key),
data={"action": "start", "kind": "cron", "cron": "0 3 * * *"},
)
sched_uid = created.json()["data"]["schedule"]["uid"]
r = requests.post(
_url(project["slug"], inst_uid, f"/schedules/{sched_uid}/delete"),
headers=_headers(other_key),
)
assert r.status_code == 403
assert get_table("instance_schedules").find_one(uid=sched_uid) is not None
finally:
_cleanup(inst_uid)
def test_primary_admin_can_manage_others_private_instance(app_server):
owner_uid, owner_key = _make_admin()
primary_key = _primary_admin_key()
project = _create_project(owner_key, "Project Manage Primary Private", is_private=True)
inst_uid = _insert_instance(project["uid"], owner_uid)
try:
r = requests.post(_url(project["slug"], inst_uid, "/stop"), headers=_headers(primary_key))
assert r.status_code == 200
finally:
_cleanup(inst_uid)
Reference in New Issue View Git Blame Copy Permalink