devplacepy/tests/e2e/admin/backups/index.py

# retoor <retoor@molodetz.nl>

import time
import requests
from tests.conftest import BASE_URL, login_user
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

_counter = [0]


def _make_admin():
    _counter[0] += 1
    name = f"bke{int(time.time() * 1000)}{_counter[0]}"
    password = "secret123"
    requests.post(
        f"{BASE_URL}/auth/signup",
        data={
            "username": name,
            "email": f"{name}@t.dev",
            "password": password,
            "confirm_password": password,
            "birth_date": "1990-01-01",
            "accept_terms": "1",
        },
        allow_redirects=True,
    )
    refresh_snapshot()
    row = get_table("users").find_one(username=name)
    get_table("users").update({"uid": row["uid"], "role": "Admin"}, ["uid"])
    clear_user_cache(row["uid"])
    invalidate_admins_cache()
    return {"email": f"{name}@t.dev", "password": password, "uid": row["uid"]}


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


def _row(page, backup_uid):
    return page.locator(
        f"tr:has(button[data-action='delete-backup'][data-uid='{backup_uid}'])"
    )


def test_download_disabled_with_tooltip_for_non_primary_admin(page, app_server):
    _make_admin()
    admin = _make_admin()
    assert get_primary_admin_uid() != admin["uid"]
    backup_uid = _seed_done_backup(admin["uid"])

    login_user(page, admin)
    page.goto(f"{BASE_URL}/admin/backups", wait_until="domcontentloaded")
    row = _row(page, backup_uid)
    row.wait_for(state="visible", timeout=15000)

    blocked = row.locator("button.admin-btn:has-text('Download')")
    blocked.wait_for(state="visible")
    assert blocked.is_disabled()
    assert blocked.get_attribute("title") == "Not available"
    assert row.locator("a.admin-btn:has-text('Download')").count() == 0


def test_download_enabled_for_primary_admin(page, app_server):
    admin = _make_admin()
    refresh_snapshot()
    others = [
        r["uid"]
        for r in get_table("users").find(role="Admin")
        if r["uid"] != admin["uid"]
    ]
    for uid in others:
        get_table("users").update({"uid": uid, "role": "Member"}, ["uid"])
        clear_user_cache(uid)
    invalidate_admins_cache()
    try:
        assert get_primary_admin_uid() == admin["uid"]
        backup_uid = _seed_done_backup(admin["uid"])
        login_user(page, admin)
        link = None
        for _ in range(20):
            page.goto(f"{BASE_URL}/admin/backups", wait_until="domcontentloaded")
            row = _row(page, backup_uid)
            row.wait_for(state="visible", timeout=15000)
            link = row.locator("a.admin-btn:has-text('Download')")
            if link.count() > 0:
                break
            time.sleep(0.3)
        link.wait_for(state="visible")
        assert link.get_attribute("href") == f"/admin/backups/{backup_uid}/download"
        assert row.locator("button.admin-btn[title='Not available']").count() == 0
    finally:
        for uid in others:
            get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
            clear_user_cache(uid)