tests/test_bootstrap_hf.py

import os
import subprocess
import tempfile
import textwrap
import unittest
from pathlib import Path


class BootstrapHfScriptTests(unittest.TestCase):
    @staticmethod
    def _repo_root() -> Path:
        return Path(__file__).resolve().parents[1]

    @classmethod
    def _script_path(cls) -> Path:
        return cls._repo_root() / "scripts" / "bootstrap-hf.sh"

    @staticmethod
    def _write_executable(path: Path, content: str) -> None:
        path.write_text(content, encoding="utf-8")
        path.chmod(0o755)

    def _prepare_fake_tools(self, bin_dir: Path) -> None:
        self._write_executable(
            bin_dir / "hf",
            textwrap.dedent(
                """\
                #!/usr/bin/env bash
                set -euo pipefail
                echo "hf $*" >>"${TRACE_FILE}"
                if [[ "${1:-}" == "auth" && "${2:-}" == "whoami" ]]; then
                  if [[ "${HF_WHOAMI_EXIT_CODE:-0}" != "0" ]]; then
                    echo "not logged in" >&2
                    exit "${HF_WHOAMI_EXIT_CODE}"
                  fi
                  if [[ -n "${HF_WHOAMI_OUTPUT:-}" ]]; then
                    echo "${HF_WHOAMI_OUTPUT}"
                  else
                    echo "user:  demo-user"
                  fi
                  exit 0
                fi
                if [[ "${1:-}" == "auth" && "${2:-}" == "login" ]]; then
                  exit 0
                fi
                if [[ "${1:-}" == "version" ]]; then
                  echo "hf 9.9.9"
                  exit 0
                fi
                if [[ "${1:-}" == "repo" && "${2:-}" == "create" ]]; then
                  exit 0
                fi
                if [[ "${1:-}" == "upload" ]]; then
                  exit 0
                fi
                echo "unexpected hf command: $*" >&2
                exit 1
                """
            ),
        )
        self._write_executable(
            bin_dir / "python3",
            textwrap.dedent(
                """\
                #!/usr/bin/env bash
                set -euo pipefail
                echo "python3 $*" >>"${TRACE_FILE}"
                if [[ "${1:-}" == "--version" ]]; then
                  echo "Python 3.12.0"
                  exit 0
                fi
                if [[ -n "${SPACE_VARIABLE_KEY:-}" ]]; then
                  echo "space-variable ${SPACE_VARIABLE_KEY}=${SPACE_VARIABLE_VALUE}" >>"${TRACE_FILE}"
                fi
                if [[ -n "${SPACE_SECRET_KEY:-}" ]]; then
                  echo "space-secret ${SPACE_SECRET_KEY}=${SPACE_SECRET_VALUE}" >>"${TRACE_FILE}"
                fi
                if [[ -n "${SPACE_RESTART_REPO_ID:-}" ]]; then
                  echo "space-restart ${SPACE_RESTART_REPO_ID}" >>"${TRACE_FILE}"
                  if [[ "${FAIL_SPACE_RESTART:-0}" == "1" ]]; then
                    echo "restart failed" >&2
                    exit 1
                  fi
                fi
                exit 0
                """
            ),
        )
        self._write_executable(
            bin_dir / "openssl",
            textwrap.dedent(
                """\
                #!/usr/bin/env bash
                set -euo pipefail
                if [[ "${1:-}" == "rand" && "${2:-}" == "-hex" && "${3:-}" == "16" ]]; then
                  echo "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                  exit 0
                fi
                if [[ "${1:-}" == "rand" && "${2:-}" == "-hex" && "${3:-}" == "8" ]]; then
                  echo "bbbbbbbbbbbbbbbb"
                  exit 0
                fi
                echo "unexpected openssl call: $*" >&2
                exit 1
                """
            ),
        )
        self._write_executable(
            bin_dir / "git",
            textwrap.dedent(
                """\
                #!/usr/bin/env bash
                set -euo pipefail
                if [[ "${1:-}" == "--version" ]]; then
                  echo "git version 2.47.0"
                  exit 0
                fi
                echo "unexpected git call: $*" >&2
                exit 1
                """
            ),
        )

    def test_interactive_bootstrap_configures_space_secrets_and_variables(self):
        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = Path(tmp)
            home_dir = tmp_path / "home"
            token_file = home_dir / ".cache" / "huggingface" / "token"
            bin_dir = tmp_path / "bin"
            trace_file = tmp_path / "trace.log"

            bin_dir.mkdir(parents=True)
            token_file.parent.mkdir(parents=True)
            token_file.write_text("hf-current-token\n", encoding="utf-8")
            trace_file.write_text("", encoding="utf-8")
            self._prepare_fake_tools(bin_dir)

            env = os.environ.copy()
            env.update(
                {
                    "HOME": str(home_dir),
                    "TRACE_FILE": str(trace_file),
                    "PATH": f"{bin_dir}:{env.get('PATH', '')}",
                }
            )

            # use current HF user(yes), space_name, dataset_name,
            # openclaw_version(blank -> default latest), gateway_token(blank),
            # gateway_password(blank), configure_llm(no), sshx_auto_start(yes), proceed(yes)
            user_input = "\ndemo-space\ndemo-backup\n\n\n\nn\ny\ny\n"

            result = subprocess.run(
                [str(self._script_path())],
                cwd=self._repo_root(),
                env=env,
                input=user_input,
                text=True,
                capture_output=True,
                check=False,
            )

            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}",
            )

            trace = trace_file.read_text(encoding="utf-8")
            self.assertIn("hf repo create demo-user/demo-space --repo-type space --space-sdk docker --private --exist-ok", trace)
            self.assertIn("hf repo create demo-user/demo-backup --repo-type dataset --private --exist-ok", trace)
            self.assertIn("hf upload demo-user/demo-space . --repo-type space --exclude .git/** --exclude .git --commit-message feat: deploy Gemma 4 to hf space", trace)

            self.assertIn("space-variable OPENCLAW_BACKUP_DATASET_REPO=demo-user/demo-backup", trace)
            self.assertIn("space-variable OPENCLAW_VERSION=latest", trace)
            self.assertIn("space-variable OPENCLAW_GATEWAY_CONTROLUI_ALLOW_INSECURE_AUTH=false", trace)
            self.assertIn("space-variable OPENCLAW_GATEWAY_CONTROLUI_DANGEROUSLY_DISABLE_DEVICE_AUTH=false", trace)
            self.assertIn("space-variable OPENCLAW_SSHX_AUTO_START=true", trace)
            self.assertIn("space-secret HF_TOKEN=hf-current-token", trace)
            self.assertIn("space-secret OPENCLAW_GATEWAY_TOKEN=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", trace)
            self.assertIn("space-secret OPENCLAW_GATEWAY_PASSWORD=bbbbbbbbbbbbbbbb", trace)
            self.assertNotIn("space-restart demo-user/demo-space", trace)

            self.assertIn("Generated OPENCLAW_GATEWAY_TOKEN=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", result.stdout)
            self.assertIn("Generated OPENCLAW_GATEWAY_PASSWORD=bbbbbbbbbbbbbbbb", result.stdout)
            self.assertIn("Hugging Face Space: https://huggingface.co/spaces/demo-user/demo-space", result.stdout)

    def test_prompts_for_hf_username_when_whoami_output_is_unparseable(self):
        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = Path(tmp)
            home_dir = tmp_path / "home"
            token_file = home_dir / ".cache" / "huggingface" / "token"
            bin_dir = tmp_path / "bin"
            trace_file = tmp_path / "trace.log"

            bin_dir.mkdir(parents=True)
            token_file.parent.mkdir(parents=True)
            token_file.write_text("hf-current-token\n", encoding="utf-8")
            trace_file.write_text("", encoding="utf-8")
            self._prepare_fake_tools(bin_dir)

            env = os.environ.copy()
            env.update(
                {
                    "HOME": str(home_dir),
                    "TRACE_FILE": str(trace_file),
                    "HF_WHOAMI_OUTPUT": "whoami output without username",
                    "PATH": f"{bin_dir}:{env.get('PATH', '')}",
                }
            )

            # use current HF user(yes), hf_username (fallback prompt), space_name, dataset_name,
            # openclaw_version(blank), gateway token/password(empty),
            # configure_llm(no), sshx_auto_start(yes), proceed(yes)
            user_input = "\nmanual-user\ndemo-space\ndemo-backup\n\n\n\nn\ny\ny\n"

            result = subprocess.run(
                [str(self._script_path())],
                cwd=self._repo_root(),
                env=env,
                input=user_input,
                text=True,
                capture_output=True,
                check=False,
            )

            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}",
            )

            trace = trace_file.read_text(encoding="utf-8")
            self.assertIn("hf repo create manual-user/demo-space --repo-type space --space-sdk docker --private --exist-ok", trace)
            self.assertIn("hf repo create manual-user/demo-backup --repo-type dataset --private --exist-ok", trace)
            self.assertIn("HF user: manual-user", result.stdout)

    def test_does_not_restart_space(self):
        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = Path(tmp)
            home_dir = tmp_path / "home"
            token_file = home_dir / ".cache" / "huggingface" / "token"
            bin_dir = tmp_path / "bin"
            trace_file = tmp_path / "trace.log"

            bin_dir.mkdir(parents=True)
            token_file.parent.mkdir(parents=True)
            token_file.write_text("hf-current-token\n", encoding="utf-8")
            trace_file.write_text("", encoding="utf-8")
            self._prepare_fake_tools(bin_dir)

            env = os.environ.copy()
            env.update(
                {
                    "HOME": str(home_dir),
                    "TRACE_FILE": str(trace_file),
                    "FAIL_SPACE_RESTART": "1",
                    "PATH": f"{bin_dir}:{env.get('PATH', '')}",
                }
            )

            # use current HF user(yes), configure_llm(no), sshx_auto_start(no), proceed(yes)
            user_input = "\ndemo-space\ndemo-backup\n\n\n\nn\nn\ny\n"

            result = subprocess.run(
                [str(self._script_path())],
                cwd=self._repo_root(),
                env=env,
                input=user_input,
                text=True,
                capture_output=True,
                check=False,
            )

            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}",
            )
            trace = trace_file.read_text(encoding="utf-8")
            self.assertIn("space-variable OPENCLAW_SSHX_AUTO_START=false", trace)
            self.assertNotIn("space-restart demo-user/demo-space", trace)
            self.assertNotIn(
                "Space restart failed. You can restart manually from Hugging Face Space Settings.",
                result.stderr,
            )

    def test_prompts_for_hf_token_when_not_logged_in(self):
        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = Path(tmp)
            home_dir = tmp_path / "home"
            bin_dir = tmp_path / "bin"
            trace_file = tmp_path / "trace.log"

            bin_dir.mkdir(parents=True)
            trace_file.write_text("", encoding="utf-8")
            self._prepare_fake_tools(bin_dir)

            env = os.environ.copy()
            env.update(
                {
                    "HOME": str(home_dir),
                    "TRACE_FILE": str(trace_file),
                    "HF_WHOAMI_EXIT_CODE": "1",
                    "PATH": f"{bin_dir}:{env.get('PATH', '')}",
                }
            )

            # hf_token_for_login(required), hf_username(fallback prompt), space_name, dataset_name,
            # openclaw_version(blank), gateway token/password(blank), configure_llm(no),
            # sshx_auto_start(yes), proceed(yes)
            user_input = "hf-login-token\nmanual-user\ndemo-space\ndemo-backup\n\n\n\nn\ny\ny\n"

            result = subprocess.run(
                [str(self._script_path())],
                cwd=self._repo_root(),
                env=env,
                input=user_input,
                text=True,
                capture_output=True,
                check=False,
            )

            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}",
            )

            trace = trace_file.read_text(encoding="utf-8")
            self.assertIn("hf auth login --token hf-login-token", trace)
            self.assertIn("space-secret HF_TOKEN=hf-login-token", trace)
            self.assertIn("HF user: manual-user", result.stdout)

    def test_switches_logged_in_user_and_restores_original_token(self):
        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = Path(tmp)
            home_dir = tmp_path / "home"
            token_file = home_dir / ".cache" / "huggingface" / "token"
            bin_dir = tmp_path / "bin"
            trace_file = tmp_path / "trace.log"

            bin_dir.mkdir(parents=True)
            token_file.parent.mkdir(parents=True)
            token_file.write_text("hf-current-token\n", encoding="utf-8")
            trace_file.write_text("", encoding="utf-8")
            self._prepare_fake_tools(bin_dir)

            env = os.environ.copy()
            env.update(
                {
                    "HOME": str(home_dir),
                    "TRACE_FILE": str(trace_file),
                    "PATH": f"{bin_dir}:{env.get('PATH', '')}",
                }
            )

            # use current HF user(no), switch_hf_token(required), space_name, dataset_name,
            # openclaw_version(blank), gateway token/password(blank), configure_llm(no),
            # sshx_auto_start(yes), proceed(yes)
            user_input = "n\nhf-switch-token\ndemo-space\ndemo-backup\n\n\n\nn\ny\ny\n"

            result = subprocess.run(
                [str(self._script_path())],
                cwd=self._repo_root(),
                env=env,
                input=user_input,
                text=True,
                capture_output=True,
                check=False,
            )

            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}",
            )

            trace = trace_file.read_text(encoding="utf-8")
            self.assertIn("hf auth login --token hf-switch-token", trace)
            self.assertIn("hf auth login --token hf-current-token", trace)
            self.assertEqual(trace.count("hf auth login --token"), 2)
            self.assertIn("space-secret HF_TOKEN=hf-switch-token", trace)

    def test_prefers_user_configured_openclaw_version_from_env(self):
        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = Path(tmp)
            home_dir = tmp_path / "home"
            token_file = home_dir / ".cache" / "huggingface" / "token"
            bin_dir = tmp_path / "bin"
            trace_file = tmp_path / "trace.log"

            bin_dir.mkdir(parents=True)
            token_file.parent.mkdir(parents=True)
            token_file.write_text("hf-current-token\n", encoding="utf-8")
            trace_file.write_text("", encoding="utf-8")
            self._prepare_fake_tools(bin_dir)

            env = os.environ.copy()
            env.update(
                {
                    "HOME": str(home_dir),
                    "TRACE_FILE": str(trace_file),
                    "OPENCLAW_VERSION": "2026.3.23-2",
                    "PATH": f"{bin_dir}:{env.get('PATH', '')}",
                }
            )

            # use current HF user(yes), space_name, dataset_name,
            # openclaw_version(blank -> use env default), gateway token/password(blank),
            # configure_llm(no), sshx_auto_start(yes), proceed(yes)
            user_input = "\ndemo-space\ndemo-backup\n\n\n\nn\ny\ny\n"

            result = subprocess.run(
                [str(self._script_path())],
                cwd=self._repo_root(),
                env=env,
                input=user_input,
                text=True,
                capture_output=True,
                check=False,
            )

            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}",
            )

            trace = trace_file.read_text(encoding="utf-8")
            self.assertIn("space-variable OPENCLAW_VERSION=2026.3.23-2", trace)
            self.assertIn("OPENCLAW_VERSION: 2026.3.23-2", result.stdout)

    def test_user_can_cancel_before_remote_execution(self):
        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = Path(tmp)
            home_dir = tmp_path / "home"
            token_file = home_dir / ".cache" / "huggingface" / "token"
            bin_dir = tmp_path / "bin"
            trace_file = tmp_path / "trace.log"

            bin_dir.mkdir(parents=True)
            token_file.parent.mkdir(parents=True)
            token_file.write_text("hf-current-token\n", encoding="utf-8")
            trace_file.write_text("", encoding="utf-8")
            self._prepare_fake_tools(bin_dir)

            env = os.environ.copy()
            env.update(
                {
                    "HOME": str(home_dir),
                    "TRACE_FILE": str(trace_file),
                    "PATH": f"{bin_dir}:{env.get('PATH', '')}",
                }
            )

            # use current HF user(yes), configure_llm(no), sshx_auto_start(yes), proceed(no)
            user_input = "\ndemo-space\ndemo-backup\n\n\n\nn\ny\nn\n"

            result = subprocess.run(
                [str(self._script_path())],
                cwd=self._repo_root(),
                env=env,
                input=user_input,
                text=True,
                capture_output=True,
                check=False,
            )

            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}",
            )

            trace = trace_file.read_text(encoding="utf-8")
            self.assertNotIn("hf repo create", trace)
            self.assertNotIn("hf upload", trace)
            self.assertNotIn("space-variable ", trace)
            self.assertNotIn("space-secret ", trace)
            self.assertIn("Cancelled by user before creating/updating Space or Dataset.", result.stdout)


if __name__ == "__main__":
    unittest.main()