govoplan-files/dev/connectors/smoke.py

from __future__ import annotations

import argparse
import os
import socket
from pathlib import Path

import httpx

from govoplan_files.backend.storage.connector_browse import browse_connector_profile
from govoplan_files.backend.storage.connector_imports import read_connector_file
from govoplan_files.backend.storage.connector_profiles import connector_profiles_from_payload


ROOT = Path(__file__).resolve().parent


def main() -> int:
    parser = argparse.ArgumentParser(description="Smoke-test GovOPlaN connector helpers against the local dev compose stack.")
    parser.add_argument("--require-smb", action="store_true", help="Fail if the SMB connector cannot browse/import.")
    parser.add_argument("--require-s3", action="store_true", help="Fail if the S3 connector cannot browse/import.")
    parser.add_argument("--debug-smb", action="store_true", help="Print direct smbclient probes before the connector smoke check.")
    args = parser.parse_args()

    _load_dotenv()
    _default_env()
    preflight_failures = _preflight_services(require_smb=args.require_smb, require_s3=args.require_s3)
    if preflight_failures:
        for failure in preflight_failures:
            print(f"FAIL {failure}")
        print("Start or recreate the connector stack from this directory with: docker compose up -d nextcloud nextcloud-db webdav smb minio")
        return 1
    _seed_webdav_fixture()
    _seed_smb_fixture()
    try:
        _seed_nextcloud_fixture()
    except httpx.HTTPError as exc:
        print(f"FAIL dev-nextcloud seed failed: {exc}")
        return 1
    try:
        _seed_s3_fixture()
    except Exception as exc:
        if args.require_s3:
            print(f"FAIL dev-s3 seed failed: {exc}")
            return 1
        print(f"SKIP dev-s3 seed failed: {exc}")

    profiles = {profile.id: profile for profile in connector_profiles_from_payload({"profiles": _profile_payloads()})}
    if args.debug_smb:
        _debug_smb(profiles["dev-smb"])
    failures: list[str] = []
    for profile_id, folder, file_path, expected in (
        ("dev-webdav", "GovOPlaN", "GovOPlaN/webdav-live.txt", "webdav live fixture"),
        ("dev-nextcloud", "GovOPlaN", "GovOPlaN/nextcloud-live.txt", "nextcloud live fixture"),
    ):
        try:
            _exercise_profile(profiles[profile_id], folder=folder, file_path=file_path, expected=expected)
        except Exception as exc:
            failures.append(f"{profile_id}: {exc}")

    try:
        _exercise_profile(profiles["dev-smb"], folder="GovOPlaN", file_path="GovOPlaN/smb-live.txt", expected="smb live fixture")
    except Exception as exc:
        message = f"dev-smb: {exc}"
        if args.require_smb:
            failures.append(message)
        else:
            print(f"SKIP {message}")

    try:
        _exercise_profile(profiles["dev-s3"], folder="GovOPlaN", file_path="GovOPlaN/s3-live.txt", expected="s3 live fixture")
    except Exception as exc:
        message = f"dev-s3: {exc}"
        if args.require_s3:
            failures.append(message)
        else:
            print(f"SKIP {message}")

    if failures:
        for failure in failures:
            print(f"FAIL {failure}")
        return 1
    print("OK connector dev stack smoke checks passed")
    return 0


def _default_env() -> None:
    defaults = {
        "NEXTCLOUD_ADMIN_USER": "admin",
        "NEXTCLOUD_ADMIN_PASSWORD": "govoplan-nextcloud-admin",
        "WEBDAV_USER": "govoplan",
        "WEBDAV_PASSWORD": "govoplan-webdav",
        "SMB_SHARE_NAME": "files",
        "SMB_USER": "govoplan",
        "SMB_PASSWORD": "govoplan-smb",
        "MINIO_ROOT_USER": "govoplan",
        "MINIO_ROOT_PASSWORD": "govoplan-minio",
        "MINIO_BUCKET": "govoplan",
    }
    for key, value in defaults.items():
        os.environ.setdefault(key, value)


def _load_dotenv() -> None:
    path = ROOT / ".env"
    if not path.exists():
        return
    for line in path.read_text(encoding="utf-8").splitlines():
        text = line.strip()
        if not text or text.startswith("#") or "=" not in text:
            continue
        key, value = text.split("=", 1)
        key = key.strip()
        if not key or key in os.environ:
            continue
        os.environ[key] = value.strip().strip("\"'")


def _preflight_services(*, require_smb: bool, require_s3: bool) -> list[str]:
    failures: list[str] = []
    nextcloud_url = f"http://127.0.0.1:{os.getenv('NEXTCLOUD_HOST_PORT', '9081')}/status.php"
    try:
        response = httpx.get(nextcloud_url, timeout=3.0)
        if response.status_code != 200:
            failures.append(f"dev-nextcloud expected HTTP 200 at {nextcloud_url}, got HTTP {response.status_code}")
    except httpx.HTTPError as exc:
        failures.append(f"dev-nextcloud is not reachable at {nextcloud_url}: {exc}")

    webdav_url = f"http://127.0.0.1:{os.getenv('WEBDAV_HOST_PORT', '9083')}/"
    try:
        response = httpx.get(webdav_url, timeout=3.0)
        if response.status_code not in {200, 401}:
            failures.append(f"dev-webdav expected HTTP 200/401 at {webdav_url}, got HTTP {response.status_code}")
    except httpx.HTTPError as exc:
        failures.append(f"dev-webdav is not reachable at {webdav_url}: {exc}")

    if require_smb:
        smb_port = int(os.getenv("SMB_HOST_PORT", "1445"))
        try:
            with socket.create_connection(("127.0.0.1", smb_port), timeout=3.0):
                pass
        except OSError as exc:
            failures.append(f"dev-smb is not reachable at 127.0.0.1:{smb_port}: {exc}")
    if require_s3:
        minio_url = f"http://127.0.0.1:{os.getenv('MINIO_API_HOST_PORT', '9000')}/minio/health/live"
        try:
            response = httpx.get(minio_url, timeout=3.0)
            if response.status_code != 200:
                failures.append(f"dev-s3 expected HTTP 200 at {minio_url}, got HTTP {response.status_code}")
        except httpx.HTTPError as exc:
            failures.append(f"dev-s3 is not reachable at {minio_url}: {exc}")
    return failures


def _profile_payloads() -> list[dict[str, object]]:
    return [
        {
            "id": "dev-webdav",
            "provider": "webdav",
            "endpoint_url": f"http://127.0.0.1:{os.getenv('WEBDAV_HOST_PORT', '9083')}/",
            "credential_mode": "basic",
            "username": os.getenv("WEBDAV_USER", "govoplan"),
            "password_env": "WEBDAV_PASSWORD",
        },
        {
            "id": "dev-nextcloud",
            "provider": "nextcloud",
            "endpoint_url": f"http://127.0.0.1:{os.getenv('NEXTCLOUD_HOST_PORT', '9081')}/remote.php/dav/files/{os.getenv('NEXTCLOUD_ADMIN_USER', 'admin')}/",
            "credential_mode": "basic",
            "username": os.getenv("NEXTCLOUD_ADMIN_USER", "admin"),
            "password_env": "NEXTCLOUD_ADMIN_PASSWORD",
        },
        {
            "id": "dev-smb",
            "provider": "smb",
            "endpoint_url": f"smb://127.0.0.1:{os.getenv('SMB_HOST_PORT', '1445')}/{os.getenv('SMB_SHARE_NAME', 'files')}",
            "credential_mode": "basic",
            "username": os.getenv("SMB_USER", "govoplan"),
            "password_env": "SMB_PASSWORD",
        },
        {
            "id": "dev-s3",
            "provider": "s3",
            "endpoint_url": f"http://127.0.0.1:{os.getenv('MINIO_API_HOST_PORT', '9000')}",
            "credential_mode": "basic",
            "username": os.getenv("MINIO_ROOT_USER", "govoplan"),
            "password_env": "MINIO_ROOT_PASSWORD",
            "metadata": {
                "bucket": os.getenv("MINIO_BUCKET", "govoplan"),
                "region": "us-east-1",
                "path_style": True,
                "verify_tls": False,
            },
        },
    ]


def _exercise_profile(profile, *, folder: str, file_path: str, expected: str) -> None:
    items = browse_connector_profile(profile, path=folder)
    paths = {item.path for item in items}
    if file_path not in paths:
        raise RuntimeError(f"{file_path!r} not found; saw {sorted(paths)!r}")
    downloaded = read_connector_file(profile, library_id="", path=file_path, max_bytes=1024 * 1024)
    text = downloaded.data.decode("utf-8").strip()
    if text != expected:
        raise RuntimeError(f"{file_path!r} content mismatch: {text!r}")
    print(f"OK {profile.id} browse/import {file_path}")


def _debug_smb(profile) -> None:
    try:
        import smbclient
    except ImportError as exc:
        print(f"DEBUG smbclient import failed: {exc}")
        return
    from govoplan_files.backend.storage.connector_browse import _smb_client_kwargs, _smb_location, _smb_unc_path

    location = _smb_location(profile)
    kwargs = _smb_client_kwargs(profile, location)
    print(f"DEBUG SMB endpoint=//{location.server}:{location.port}/{location.share} root_path={location.root_path!r}")
    print(f"DEBUG SMB user={kwargs.get('username')!r} require_signing={kwargs.get('require_signing')!r} auth_protocol={kwargs.get('auth_protocol')!r}")
    try:
        smbclient.reset_connection_cache()
    except Exception as exc:
        print(f"DEBUG SMB reset cache failed: {exc}")
    for path in (_smb_unc_path(location, ""), _smb_unc_path(location, "GovOPlaN")):
        try:
            print(f"DEBUG SMB list {path}: {smbclient.listdir(path, **kwargs)}")
        except Exception as exc:
            print(f"DEBUG SMB list {path} failed: {type(exc).__name__}: {exc}")


def _seed_webdav_fixture() -> None:
    path = ROOT / "data" / "webdav" / "GovOPlaN"
    path.mkdir(parents=True, exist_ok=True)
    (path / "webdav-live.txt").write_text("webdav live fixture\n", encoding="utf-8")


def _seed_smb_fixture() -> None:
    path = ROOT / "data" / "smb" / "GovOPlaN"
    path.mkdir(parents=True, exist_ok=True)
    (path / "smb-live.txt").write_text("smb live fixture\n", encoding="utf-8")


def _seed_nextcloud_fixture() -> None:
    base_url = f"http://127.0.0.1:{os.getenv('NEXTCLOUD_HOST_PORT', '9081')}/remote.php/dav/files/{os.getenv('NEXTCLOUD_ADMIN_USER', 'admin')}/GovOPlaN"
    auth = (os.getenv("NEXTCLOUD_ADMIN_USER", "admin"), os.getenv("NEXTCLOUD_ADMIN_PASSWORD", "govoplan-nextcloud-admin"))
    response = httpx.request("MKCOL", base_url, auth=auth, timeout=15.0)
    if response.status_code not in {201, 405}:
        raise RuntimeError(f"Nextcloud MKCOL failed with HTTP {response.status_code}: {response.text[:200]}")
    response = httpx.put(f"{base_url}/nextcloud-live.txt", content=b"nextcloud live fixture\n", auth=auth, timeout=15.0)
    if response.status_code not in {200, 201, 204}:
        raise RuntimeError(f"Nextcloud PUT failed with HTTP {response.status_code}: {response.text[:200]}")


def _seed_s3_fixture() -> None:
    try:
        import boto3
        from botocore.config import Config
        from botocore.exceptions import ClientError
    except ImportError as exc:
        raise RuntimeError("boto3 is not installed; install govoplan-files[s3]") from exc
    bucket = os.getenv("MINIO_BUCKET", "govoplan")
    client = boto3.client(
        "s3",
        endpoint_url=f"http://127.0.0.1:{os.getenv('MINIO_API_HOST_PORT', '9000')}",
        aws_access_key_id=os.getenv("MINIO_ROOT_USER", "govoplan"),
        aws_secret_access_key=os.getenv("MINIO_ROOT_PASSWORD", "govoplan-minio"),
        region_name="us-east-1",
        config=Config(s3={"addressing_style": "path"}),
    )
    try:
        client.create_bucket(Bucket=bucket)
    except ClientError as exc:
        code = exc.response.get("Error", {}).get("Code")
        if code not in {"BucketAlreadyOwnedByYou", "BucketAlreadyExists"}:
            raise
    client.put_object(Bucket=bucket, Key="GovOPlaN/s3-live.txt", Body=b"s3 live fixture\n", ContentType="text/plain")


if __name__ == "__main__":
    raise SystemExit(main())