test: add PHPUnit controller and Python ExApp integration tests

Signed-off-by: Oleksander Piskun <oleksandr2088@icloud.com>

Author: oleksandr-nc

tests/exapp_integration/.gitignore

@@ -0,0 +1,5 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+venv/
+__pycache__/
+.pytest_cache/

tests/exapp_integration/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later

tests/exapp_integration/_client.py

@@ -0,0 +1,60 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Tiny HTTP client wrapping AppAPI's auth contract.
+
+Replaces the parts of nc_py_api that the integration tests actually used:
+the AppAPI auth headers and a `requests`-style call.
+
+No HMAC / request signing — AppAPI accepts the simple base64 auth header for
+ExApp -> Nextcloud calls. See `tests/install_no_init.py` (the existing in-tree
+test ExApp), which hits AppAPI's /log endpoint with these same headers.
+"""
+
+from __future__ import annotations
+
+from base64 import b64encode
+from dataclasses import dataclass, field
+from typing import Any
+
+import requests
+
+
+@dataclass
+class AppAPIClient:
+    base_url: str
+    app_id: str
+    app_secret: str
+    app_version: str = "1.0.0"
+    user: str = "admin"
+    extra_headers: dict[str, str] = field(default_factory=dict)
+    timeout: float = 30.0
+
+    def auth_headers(self) -> dict[str, str]:
+        basic = b64encode(f"{self.user}:{self.app_secret}".encode()).decode()
+        h = {
+            "EX-APP-ID": self.app_id,
+            "EX-APP-VERSION": self.app_version,
+            "AUTHORIZATION-APP-API": basic,
+            "AA-VERSION": "2.0.0",
+            "OCS-APIRequest": "true",
+            "Accept": "application/json",
+        }
+        if self.user != "admin":
+            h["AA-USER-ID"] = self.user
+        h.update(self.extra_headers)
+        return h
+
+    def request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
+        """Hit `<base_url><path>` with AppAPI auth headers.
+
+        Caller-supplied `headers` win over the defaults. Pass `headers={"X": None}`
+        to drop a default header (used in negative-auth tests).
+        """
+        headers = self.auth_headers()
+        for k, v in (kwargs.pop("headers", None) or {}).items():
+            if v is None:
+                headers.pop(k, None)
+            else:
+                headers[k] = v
+        kwargs.setdefault("timeout", self.timeout)
+        return requests.request(method, f"{self.base_url}{path}", headers=headers, **kwargs)

tests/exapp_integration/_test_app.py

@@ -0,0 +1,78 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Minimal test ExApp for AppAPI integration tests.
+
+Replaces the nc_py_api-based `tests/_install.py`. Implements only the
+callbacks AppAPI invokes (`/init`, `/enabled`, `/heartbeat`) plus a few
+introspection endpoints used by the integration tests.
+"""
+
+import os
+from base64 import b64decode
+
+from fastapi import FastAPI, HTTPException, Request as FastAPIRequest
+from fastapi.responses import JSONResponse
+
+APP_ID = os.environ["APP_ID"]
+APP_SECRET = os.environ["APP_SECRET"]
+APP_VERSION = os.environ.get("APP_VERSION", "1.0.0")
+
+APP = FastAPI()
+
+
+def verify_auth(request: FastAPIRequest) -> str:
+    """Validate AppAPI auth headers. Returns the username on success."""
+    ex_app_id = request.headers.get("EX-APP-ID", "")
+    ex_app_version = request.headers.get("EX-APP-VERSION", "")
+    auth_app_api = request.headers.get("AUTHORIZATION-APP-API", "")
+
+    if not ex_app_id or not ex_app_version or not auth_app_api:
+        raise HTTPException(status_code=401, detail="Missing AppAPI headers")
+    if ex_app_id != APP_ID:
+        raise HTTPException(status_code=401, detail=f"Invalid EX-APP-ID: {ex_app_id}")
+    try:
+        decoded = b64decode(auth_app_api).decode("UTF-8")
+        username, secret = decoded.split(":", maxsplit=1)
+    except Exception:
+        raise HTTPException(status_code=401, detail="Malformed AUTHORIZATION-APP-API")
+    if secret != APP_SECRET:
+        raise HTTPException(status_code=401, detail="Invalid app secret")
+    return username
+
+
+@APP.post("/init")
+async def init_callback(request: FastAPIRequest):
+    verify_auth(request)
+    return JSONResponse(content={}, status_code=200)
+
+
+@APP.put("/enabled")
+async def enabled_callback(enabled: bool, request: FastAPIRequest):
+    verify_auth(request)
+    return JSONResponse(content={"error": ""}, status_code=200)
+
+
+@APP.get("/heartbeat")
+async def heartbeat_callback():
+    return JSONResponse(content={"status": "ok"}, status_code=200)
+
+
+@APP.get("/cfg-echo")
+async def cfg_echo(request: FastAPIRequest):
+    """Echo the env contract back. Used by test_app_cfg.py."""
+    verify_auth(request)
+    return {
+        "app_id": APP_ID,
+        "app_version": APP_VERSION,
+        "app_secret_prefix": APP_SECRET[:6],
+    }
+
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(
+        "_test_app:APP",
+        host=os.environ.get("APP_HOST", "0.0.0.0"),
+        port=int(os.environ.get("APP_PORT", "9009")),
+        log_level="info",
+    )

tests/exapp_integration/conftest.py

@@ -0,0 +1,69 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Pytest fixtures for AppAPI integration tests.
+
+These tests assume:
+  1. A Nextcloud instance reachable at NEXTCLOUD_URL.
+  2. The test ExApp (`_test_app.py`) is running on TEST_APP_HOST:TEST_APP_PORT.
+  3. The ExApp has been registered & enabled through OCC against the
+     manual_daemon (see register_test_exapp.sh).
+
+In the dev VM all three are arranged before running pytest. See README.md.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+import pytest
+
+from ._client import AppAPIClient
+
+NEXTCLOUD_URL = os.environ.get("NEXTCLOUD_URL", "http://nextcloud.appapi")
+APP_ID = os.environ.get("APP_ID", "test_appapi")
+APP_VERSION = os.environ.get("APP_VERSION", "1.0.0")
+APP_SECRET = os.environ["APP_SECRET"]
+TEST_APP_URL = os.environ.get("TEST_APP_URL", "http://127.0.0.1:9009")
+
+OCC = ["docker", "exec", "appapi-nextcloud-1",
+       "sudo", "-u", "www-data", "php", "occ"]
+
+
+@pytest.fixture(scope="session")
+def client() -> AppAPIClient:
+    return AppAPIClient(
+        base_url=NEXTCLOUD_URL, app_id=APP_ID,
+        app_secret=APP_SECRET, app_version=APP_VERSION,
+    )
+
+
+@pytest.fixture(scope="session")
+def app_id() -> str:
+    return APP_ID
+
+
+@pytest.fixture(scope="session")
+def app_version() -> str:
+    return APP_VERSION
+
+
+@pytest.fixture(scope="session")
+def app_secret() -> str:
+    return APP_SECRET
+
+
+@pytest.fixture(scope="session")
+def test_app_url() -> str:
+    return TEST_APP_URL
+
+
+@pytest.fixture(scope="session", autouse=True)
+def _ensure_test_app_registered() -> None:
+    r = subprocess.run(OCC + ["app_api:app:list"], capture_output=True, text=True, check=True)
+    if APP_ID not in r.stdout:
+        raise RuntimeError(
+            f"ExApp '{APP_ID}' not registered. Run register_test_exapp.sh first."
+        )
+    if "[enabled]" not in next((line for line in r.stdout.splitlines() if APP_ID in line), ""):
+        raise RuntimeError(f"ExApp '{APP_ID}' is registered but not enabled.")

tests/exapp_integration/register_test_exapp.sh

@@ -0,0 +1,59 @@
+#!/bin/bash
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+# Register and enable the integration-test ExApp via OCC against the live
+# Nextcloud container. Idempotent: re-running unregisters first.
+#
+# Required env: APP_SECRET (32+ char random string).
+# Optional env: NEXTCLOUD_CONTAINER (default appapi-nextcloud-1),
+#               DAEMON_NAME       (default manual_daemon),
+#               APP_ID            (default test_appapi),
+#               APP_VERSION       (default 1.0.0),
+#               APP_PORT          (default 9009),
+#               APP_HOST          (default host.docker.internal — host as seen
+#                                  from inside the Nextcloud container).
+#
+# The ExApp itself (uvicorn _test_app:APP) must be running and reachable from
+# the Nextcloud container at $APP_HOST:$APP_PORT BEFORE this script runs,
+# because `app_api:app:enable` calls /init synchronously.
+set -euo pipefail
+
+NEXTCLOUD_CONTAINER=${NEXTCLOUD_CONTAINER:-appapi-nextcloud-1}
+DAEMON_NAME=${DAEMON_NAME:-manual_daemon}
+APP_ID=${APP_ID:-test_appapi}
+APP_VERSION=${APP_VERSION:-1.0.0}
+APP_PORT=${APP_PORT:-9009}
+APP_HOST=${APP_HOST:-host.docker.internal}
+
+if [[ -z "${APP_SECRET:-}" ]]; then
+  echo "APP_SECRET must be set (32+ random chars)" >&2
+  exit 2
+fi
+
+OCC=(docker exec "$NEXTCLOUD_CONTAINER" sudo -u www-data php occ)
+
+# Wait for /heartbeat (max 30s)
+for _ in $(seq 1 30); do
+  if docker exec "$NEXTCLOUD_CONTAINER" curl -fs "http://$APP_HOST:$APP_PORT/heartbeat" >/dev/null; then
+    break
+  fi
+  sleep 1
+done
+docker exec "$NEXTCLOUD_CONTAINER" curl -fs "http://$APP_HOST:$APP_PORT/heartbeat" >/dev/null \
+  || { echo "Test ExApp /heartbeat unreachable at $APP_HOST:$APP_PORT" >&2; exit 1; }
+
+# Best-effort unregister so the script is idempotent.
+"${OCC[@]}" app_api:app:unregister "$APP_ID" --silent 2>/dev/null || true
+
+JSON=$(cat <<EOF
+{"id":"$APP_ID","name":"AppAPI Integration Test ExApp","version":"$APP_VERSION","secret":"$APP_SECRET","port":$APP_PORT,"host":"$APP_HOST","protocol":"http","system_app":0}
+EOF
+)
+"${OCC[@]}" app_api:app:register "$APP_ID" "$DAEMON_NAME" --json-info="$JSON" --silent
+"${OCC[@]}" app_api:app:enable "$APP_ID" || true
+
+# Confirm
+"${OCC[@]}" app_api:app:list | grep -q "^$APP_ID .* \[enabled\]$" \
+  || { echo "ExApp not in enabled state after register" >&2; exit 1; }
+
+echo "Registered & enabled: $APP_ID (secret=${APP_SECRET:0:6}…)"

tests/exapp_integration/test_app_cfg.py

@@ -0,0 +1,34 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Env contract: the running ExApp received APP_ID/APP_VERSION/APP_SECRET.
+
+Replaces nc_py_api/tests/actual_tests/nc_app_test.py::test_app_cfg.
+
+We do not assert from the AppAPI side — the contract is "the deploy daemon
+populated these env vars in the ExApp container". Test by asking the test
+ExApp to echo them back.
+"""
+
+import requests
+
+
+def test_test_app_received_env(test_app_url: str, app_id: str, app_version: str, app_secret: str) -> None:
+    # Use the same simple base64 auth the ExApp validates incoming requests with.
+    from base64 import b64encode
+    auth = b64encode(f"admin:{app_secret}".encode()).decode()
+    r = requests.get(
+        f"{test_app_url}/cfg-echo",
+        headers={
+            "EX-APP-ID": app_id,
+            "EX-APP-VERSION": app_version,
+            "AUTHORIZATION-APP-API": auth,
+        },
+        timeout=10,
+    )
+    assert r.status_code == 200
+    body = r.json()
+    assert body["app_id"] == app_id
+    assert body["app_version"] == app_version
+    # Don't transmit/log the full secret; first 6 chars is enough to prove
+    # the same value was injected.
+    assert body["app_secret_prefix"] == app_secret[:6]

tests/exapp_integration/test_lifecycle.py

@@ -0,0 +1,40 @@
+# SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+"""Lifecycle callbacks: /init and /enabled fire on the registered ExApp.
+
+Replaces the implicit lifecycle coverage from nc_py_api/tests/_install.py and
+its `set_handlers` indirection. We use AppAPI's /ex-app/{appId}/enabled OCS
+endpoint to flip enabled state and verify the test ExApp returned 200 to the
+callback (otherwise the OCC enable command itself would fail).
+"""
+
+import subprocess
+
+OCC = ["docker", "exec", "appapi-nextcloud-1",
+       "sudo", "-u", "www-data", "php", "occ"]
+
+
+def _is_enabled(app_id: str) -> bool:
+    r = subprocess.run(OCC + ["app_api:app:list"], capture_output=True, text=True, check=True)
+    line = next((ln for ln in r.stdout.splitlines() if app_id in ln), "")
+    return "[enabled]" in line
+
+
+def test_disable_then_reenable_calls_callbacks(app_id: str) -> None:
+    """Toggle enabled state via OCC and confirm AppAPI invoked the ExApp's
+    /enabled callback (the OCC command would error out non-zero if the
+    callback returned anything other than 200)."""
+    assert _is_enabled(app_id), "test fixture must start enabled"
+
+    try:
+        r = subprocess.run(OCC + ["app_api:app:disable", app_id], capture_output=True, text=True)
+        assert r.returncode == 0, f"disable failed: stdout={r.stdout!r} stderr={r.stderr!r}"
+        assert not _is_enabled(app_id)
+
+        r = subprocess.run(OCC + ["app_api:app:enable", app_id], capture_output=True, text=True)
+        assert r.returncode == 0, f"enable failed: stdout={r.stdout!r} stderr={r.stderr!r}"
+        assert _is_enabled(app_id)
+    finally:
+        # Make sure we leave the fixture enabled even on test failure.
+        if not _is_enabled(app_id):
+            subprocess.run(OCC + ["app_api:app:enable", app_id], check=False)