frappe_docker/tests/test_frappe_docker.py

import os
from pathlib import Path
from typing import Any

import pytest

from tests.conftest import S3ServiceResult
from tests.utils import Compose, check_url_content

BACKEND_SERVICES = (
    "backend",
    "queue-short",
    "queue-default",
    "queue-long",
    "scheduler",
)


@pytest.mark.parametrize("service", BACKEND_SERVICES)
def test_links_in_backends(service: str, compose: Compose, python_path: str):
    filename = "_check_connections.py"
    compose("cp", f"tests/{filename}", f"{service}:/tmp/")
    compose.exec(service, python_path, f"/tmp/{filename}")


def index_cb(text: str):
    if "404 page not found" not in text:
        return text[:200]


def api_cb(text: str):
    if '"message"' in text:
        return text


def assets_cb(text: str):
    if text:
        return text[:200]


@pytest.mark.parametrize(
    ("url", "callback"), (("/", index_cb), ("/api/method/version", api_cb))
)
def test_endpoints(url: str, callback: Any, frappe_site: str):
    check_url_content(
        url=f"http://127.0.0.1{url}", callback=callback, site_name=frappe_site
    )


@pytest.mark.skipif(
    os.environ["FRAPPE_VERSION"][0:3] == "v12", reason="v12 doesn't have the asset"
)
def test_assets_endpoint(frappe_site: str):
    check_url_content(
        url=f"http://127.0.0.1/assets/frappe/images/frappe-framework-logo.svg",
        callback=assets_cb,
        site_name=frappe_site,
    )


def test_files_reachable(frappe_site: str, tmp_path: Path, compose: Compose):
    content = "lalala\n"
    file_path = tmp_path / "testfile.txt"

    with file_path.open("w") as f:
        f.write(content)

    compose(
        "cp",
        str(file_path),
        f"backend:/home/frappe/frappe-bench/sites/{frappe_site}/public/files/",
    )

    def callback(text: str):
        if text == content:
            return text

    check_url_content(
        url=f"http://127.0.0.1/files/{file_path.name}",
        callback=callback,
        site_name=frappe_site,
    )


@pytest.mark.parametrize("service", BACKEND_SERVICES)
@pytest.mark.usefixtures("frappe_site")
def test_frappe_connections_in_backends(
    service: str, python_path: str, compose: Compose
):
    filename = "_ping_frappe_connections.py"
    compose("cp", f"tests/{filename}", f"{service}:/tmp/")
    compose.exec(service, python_path, f"/tmp/{filename}")


def test_push_backup(
    python_path: str,
    frappe_site: str,
    s3_service: S3ServiceResult,
    compose: Compose,
):
    compose.bench("--site", frappe_site, "backup", "--with-files")
    compose.exec(
        "backend",
        "push-backup",
        "--site",
        frappe_site,
        "--bucket",
        "frappe",
        "--region-name",
        "us-east-1",
        "--endpoint-url",
        "http://minio:9000",
        "--aws-access-key-id",
        s3_service.access_key,
        "--aws-secret-access-key",
        s3_service.secret_key,
    )
    compose("cp", "tests/_check_backup_files.py", "backend:/tmp")
    compose.exec(
        "-e",
        f"S3_ACCESS_KEY={s3_service.access_key}",
        "-e",
        f"S3_SECRET_KEY={s3_service.secret_key}",
        "-e",
        f"SITE_NAME={frappe_site}",
        "backend",
        python_path,
        "/tmp/_check_backup_files.py",
    )


def test_https(frappe_site: str, compose: Compose):
    compose("-f", "overrides/compose.https.yaml", "up", "-d")
    check_url_content(url="https://127.0.0.1", callback=index_cb, site_name=frappe_site)


@pytest.mark.usefixtures("erpnext_setup")
class TestErpnext:
    @pytest.mark.parametrize(
        ("url", "callback"),
        (
            (
                "/api/method/erpnext.templates.pages.product_search.get_product_list",
                api_cb,
            ),
            ("/assets/erpnext/js/setup_wizard.js", assets_cb),
        ),
    )
    def test_endpoints(self, url: str, callback: Any, erpnext_site: str):
        check_url_content(
            url=f"http://127.0.0.1{url}", callback=callback, site_name=erpnext_site
        )


@pytest.mark.usefixtures("postgres_setup")
class TestPostgres:
    def test_site_creation(self, compose: Compose):
        compose.bench(
            "new-site",
            "test_pg_site",
            "--db-type",
            "postgres",
            "--admin-password",
            "admin",
        )