devplacepy/tests/api/admin/containers/manage.py

# 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"acm{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",
            "birth_date": "1990-01-01",
            "accept_terms": "1",
        },
        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": "admin 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"acm-{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)
    refresh_snapshot()


def test_data_flags_can_manage_for_owner_and_hides_for_other_on_public_project(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    project = _create_project(owner_key, "Manage Data Public", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        owner_rows = requests.get(
            f"{BASE_URL}/admin/containers/data", headers=_headers(owner_key)
        ).json()["instances"]
        other_rows = requests.get(
            f"{BASE_URL}/admin/containers/data", headers=_headers(other_key)
        ).json()["instances"]
        owner_row = next(r for r in owner_rows if r["uid"] == inst_uid)
        other_row = next(r for r in other_rows if r["uid"] == inst_uid)
        assert owner_row["can_manage"] is True
        assert other_row["can_manage"] is False
    finally:
        _cleanup(inst_uid)


def test_data_excludes_others_private_project_instance(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    project = _create_project(owner_key, "Manage Data Private", is_private=True)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        other_rows = requests.get(
            f"{BASE_URL}/admin/containers/data", headers=_headers(other_key)
        ).json()["instances"]
        assert all(r["uid"] != inst_uid for r in other_rows)
        owner_rows = requests.get(
            f"{BASE_URL}/admin/containers/data", headers=_headers(owner_key)
        ).json()["instances"]
        assert any(r["uid"] == inst_uid for r in owner_rows)
    finally:
        _cleanup(inst_uid)


def test_primary_admin_sees_and_can_manage_others_private_instance(app_server):
    owner_uid, owner_key = _make_admin()
    primary_key = _primary_admin_key()
    project = _create_project(owner_key, "Manage Data Primary", is_private=True)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        rows = requests.get(
            f"{BASE_URL}/admin/containers/data", headers=_headers(primary_key)
        ).json()["instances"]
        row = next(r for r in rows if r["uid"] == inst_uid)
        assert row["can_manage"] is True
    finally:
        _cleanup(inst_uid)


def test_instance_detail_404_for_non_viewer_on_private_project(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    project = _create_project(owner_key, "Manage Detail Private", is_private=True)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.get(
            f"{BASE_URL}/admin/containers/{inst_uid}", headers=_headers(other_key)
        )
        assert r.status_code == 404
        r = requests.get(
            f"{BASE_URL}/admin/containers/{inst_uid}", headers=_headers(owner_key)
        )
        assert r.status_code == 200
    finally:
        _cleanup(inst_uid)


def test_instance_detail_view_only_for_non_owner_on_public_project(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    project = _create_project(owner_key, "Manage Detail Public", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.get(
            f"{BASE_URL}/admin/containers/{inst_uid}", headers=_headers(other_key)
        )
        assert r.status_code == 200
        assert r.json()["can_manage"] is False

        r = requests.get(
            f"{BASE_URL}/admin/containers/{inst_uid}", headers=_headers(owner_key)
        )
        assert r.json()["can_manage"] is True
    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, "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(
                f"{BASE_URL}/admin/containers/{inst_uid}/{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, "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(
                f"{BASE_URL}/admin/containers/{inst_uid}/{action}",
                headers=_headers(owner_key),
            )
            assert r.status_code == 200, action
    finally:
        _cleanup(inst_uid)


def test_delete_denied_for_non_owner_admin(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    project = _create_project(owner_key, "Manage Delete Denied", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.post(
            f"{BASE_URL}/admin/containers/{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, "Manage Delete Owner", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.post(
            f"{BASE_URL}/admin/containers/{inst_uid}/delete", headers=_headers(owner_key)
        )
        assert r.status_code == 200
    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, "Manage Sync Denied", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.post(
            f"{BASE_URL}/admin/containers/{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, "Manage Sync Owner", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.post(
            f"{BASE_URL}/admin/containers/{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_configure_denied_for_non_owner_admin(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    project = _create_project(owner_key, "Manage Configure Denied", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.post(
            f"{BASE_URL}/admin/containers/{inst_uid}/edit",
            headers=_headers(other_key),
            data={"restart_policy": "always"},
        )
        assert r.status_code == 403
    finally:
        _cleanup(inst_uid)


def test_configure_allowed_for_owner(app_server):
    owner_uid, owner_key = _make_admin()
    project = _create_project(owner_key, "Manage Configure Owner", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.post(
            f"{BASE_URL}/admin/containers/{inst_uid}/edit",
            headers=_headers(owner_key),
            data={"restart_policy": "always"},
        )
        assert r.status_code == 200
    finally:
        _cleanup(inst_uid)


def test_edit_page_redirects_non_owner_admin(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    project = _create_project(owner_key, "Manage Edit Page Denied", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.get(
            f"{BASE_URL}/admin/containers/{inst_uid}/edit",
            headers=_headers(other_key),
            allow_redirects=False,
        )
        assert r.status_code == 302
        assert r.headers["location"] == f"/admin/containers/{inst_uid}"
    finally:
        _cleanup(inst_uid)


def test_edit_page_allowed_for_owner(app_server):
    owner_uid, owner_key = _make_admin()
    project = _create_project(owner_key, "Manage Edit Page Owner", is_private=False)
    inst_uid = _insert_instance(project["uid"], owner_uid)
    try:
        r = requests.get(
            f"{BASE_URL}/admin/containers/{inst_uid}/edit", headers=_headers(owner_key)
        )
        assert r.status_code == 200
    finally:
        _cleanup(inst_uid)


def test_create_rejects_others_private_project(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    project = _create_project(owner_key, "Manage Create Denied", is_private=True)
    r = requests.post(
        f"{BASE_URL}/admin/containers/create",
        headers=_headers(other_key),
        data={"project_slug": project["slug"], "name": "denied-instance"},
    )
    assert r.status_code == 404
    assert get_table("instances").find_one(name="denied-instance") is None


def test_projects_search_excludes_others_private_project(app_server):
    owner_uid, owner_key = _make_admin()
    _, other_key = _make_admin()
    title = f"Manage Search Unique {int(time.time() * 1000)}"
    project = _create_project(owner_key, title, is_private=True)
    r = requests.get(
        f"{BASE_URL}/admin/containers/projects/search",
        headers=_headers(other_key),
        params={"q": title},
    )
    assert all(res["slug"] != project["slug"] for res in r.json()["results"])
    r = requests.get(
        f"{BASE_URL}/admin/containers/projects/search",
        headers=_headers(owner_key),
        params={"q": title},
    )
    assert any(res["slug"] == project["slug"] for res in r.json()["results"])