devplacepy/tests/api/admin/backups/download.py

# retoor <retoor@molodetz.nl>

import time
import requests
from tests.conftest import BASE_URL
from devplacepy import config
from devplacepy.database import (
    get_table,
    get_primary_admin_uid,
    invalidate_admins_cache,
    refresh_snapshot,
)
from devplacepy.services.backup import store
from devplacepy.utils import clear_user_cache

JSON = {"Accept": "application/json"}
_counter = [0]


def _signup():
    _counter[0] += 1
    name = f"bkp{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,
    )
    refresh_snapshot()
    return get_table("users").find_one(username=name)


def _make_admin():
    row = _signup()
    get_table("users").update({"uid": row["uid"], "role": "Admin"}, ["uid"])
    clear_user_cache(row["uid"])
    invalidate_admins_cache()
    return row["api_key"]


def _primary_admin_key():
    refresh_snapshot()
    uid = get_primary_admin_uid()
    return get_table("users").find_one(uid=uid)["api_key"]


def _seed_done_backup(owner_uid):
    backup_uid = store.create_backup(
        target="database", created_by=owner_uid, job_uid="jobtest"
    )
    target_dir = config.BACKUPS_DIR
    target_dir.mkdir(parents=True, exist_ok=True)
    path = target_dir / f"{backup_uid}.tar.gz"
    path.write_bytes(b"\x1f\x8b\x08\x00test-archive")
    store.finalize_backup(
        backup_uid,
        filename="database-test.tar.gz",
        local_path=str(path),
        stats={"bytes_out": path.stat().st_size, "file_count": 1},
    )
    refresh_snapshot()
    return backup_uid


def test_download_requires_admin_guest(app_server):
    assert (
        requests.get(
            f"{BASE_URL}/admin/backups/nope/download",
            headers=JSON,
            allow_redirects=False,
        ).status_code
        == 401
    )
    assert (
        requests.get(
            f"{BASE_URL}/admin/backups/nope/download", allow_redirects=False
        ).status_code
        == 303
    )


def test_download_denied_for_member(app_server, seeded_db):
    s = requests.Session()
    s.post(
        f"{BASE_URL}/auth/login",
        data={
            "email": seeded_db["bob"]["email"],
            "password": seeded_db["bob"]["password"],
        },
    )
    r = s.get(
        f"{BASE_URL}/admin/backups/nope/download", headers=JSON, allow_redirects=False
    )
    assert r.status_code == 403


def test_download_and_flag_denied_for_non_primary_admin(app_server):
    key = _make_admin()
    headers = {**JSON, "X-API-KEY": key}
    assert (
        requests.get(
            f"{BASE_URL}/admin/backups/nope/download",
            headers=headers,
            allow_redirects=False,
        ).status_code
        == 403
    )
    data = requests.get(f"{BASE_URL}/admin/backups/data", headers=headers).json()
    assert data["can_download_backups"] is False
    assert all(b.get("download_url") is None for b in data["backups"])


def test_primary_admin_flag_true(app_server):
    headers = {**JSON, "X-API-KEY": _primary_admin_key()}
    data = requests.get(f"{BASE_URL}/admin/backups/data", headers=headers).json()
    assert data["can_download_backups"] is True


def test_primary_admin_can_download_archive(app_server):
    uid = get_primary_admin_uid()
    key = get_table("users").find_one(uid=uid)["api_key"]
    backup_uid = _seed_done_backup(uid)
    headers = {**JSON, "X-API-KEY": key}

    data = requests.get(f"{BASE_URL}/admin/backups/data", headers=headers).json()
    rows = [b for b in data["backups"] if b["uid"] == backup_uid]
    assert rows and rows[0]["download_url"] == f"/admin/backups/{backup_uid}/download"

    r = requests.get(f"{BASE_URL}/admin/backups/{backup_uid}/download", headers=headers)
    assert r.status_code == 200
    assert r.headers["content-type"] == "application/gzip"