archy/tests/routstr-provider/test_package.py

"""Offline packaging invariants; not provider/runtime acceptance tests."""
from pathlib import Path
import unittest

import yaml

ROOT = Path(__file__).resolve().parents[2]
SOURCE_REVISION = "8ce9ef69fc1b5ecd6f92f1baf54d8c2620ea0e89"


class ProviderPackageTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.app = yaml.safe_load((ROOT / "apps/routstr/manifest.yml").read_text())["app"]
        cls.dockerfile = (ROOT / "docker/routstr/Dockerfile").read_text()
        cls.env = dict(entry.split("=", 1) for entry in cls.app["environment"])

    def test_api_is_host_local_and_not_launchable(self):
        self.assertEqual(len(self.app["ports"]), 1)
        port = self.app["ports"][0]
        self.assertEqual((port["bind"], port["auth"]), ("127.0.0.1", "local"))
        self.assertEqual(port["host"], 8200)
        self.assertEqual(port["container"], 8000)
        self.assertTrue(all(i["type"] == "api" for i in self.app["interfaces"].values()))

    def test_readonly_unprivileged_runtime(self):
        security = self.app["security"]
        self.assertTrue(security["readonly_root"])
        self.assertTrue(security["no_new_privileges"])
        self.assertEqual(security["capabilities"], [])
        self.assertNotEqual(self.app["container"]["network"], "host")
        self.assertNotEqual(security["network_policy"], "host")

    def test_all_persistent_state_is_app_owned(self):
        binds = [v for v in self.app["volumes"] if v["type"] == "bind"]
        self.assertEqual(len(binds), 1)
        self.assertEqual(binds[0]["source"], "/var/lib/archipelago/routstr")
        self.assertEqual(binds[0]["target"], "/data")
        self.assertIn("WORKDIR /data", self.dockerfile)
        self.assertEqual(self.env["DATABASE_URL"], "sqlite+aiosqlite:////data/keys.db")
        self.assertEqual(self.env["ROUTSTR_SECRET_KEY_FILE"], "/data/routstr_secret.key")
        self.assertEqual(self.env["MODELS_PATH"], "/data/models.json")

    def test_admin_uses_dedicated_secret_not_appliance_identity(self):
        container = self.app["container"]
        self.assertEqual(container["generated_secrets"], [
            {"name": "routstr-admin-password", "kind": "hex32"}])
        self.assertEqual(container["secret_env"], [
            {"key": "ADMIN_PASSWORD", "secret_file": "routstr-admin-password"},
            {"key": "UPSTREAM_API_KEY", "secret_file": "llama-server-api-key", "optional": True}])
        self.assertNotIn("ADMIN_PASSWORD", self.env)
        self.assertNotIn("NSEC", self.env)
        self.assertEqual(self.env["AUTO_GENERATE_NSEC"], "false")

    def test_fresh_install_is_not_published(self):
        for key in ("HTTP_URL", "ONION_URL", "RELAYS"):
            with self.subTest(key=key):
                self.assertEqual(self.env[key], "")

    def test_upstream_is_the_private_llama_runtime(self):
        self.assertEqual(self.env["UPSTREAM_BASE_URL"], "http://llama-server:8080/v1")
        self.assertIn({"app_id": "llama-server"}, self.app["dependencies"])

    def test_mints_and_payout_come_from_the_node_wallet(self):
        # An empty CASHU_MINTS makes Routstr fall back to a built-in public
        # mint, so the mints must always be rendered, never left empty.
        self.assertEqual(self.app["container"]["derived_env"], [
            {"key": "CASHU_MINTS", "template": "{{ECASH_MINTS}}"},
            {"key": "RECEIVE_LN_ADDRESS", "template": "{{ECASH_LN_ADDRESS}}"}])
        self.assertNotIn("CASHU_MINTS", self.env)
        self.assertNotIn("RECEIVE_LN_ADDRESS", self.env)
        self.assertEqual(self.env["PRIMARY_MINT_URL"], "")

    def test_start_wrapper_reapplies_wallet_settings_and_prices_once(self):
        wrapper = (ROOT / "docker/routstr/archy-start.py").read_text()
        self.assertIn('CMD ["/opt/routstr/.venv/bin/python", "/opt/archy/archy-start.py"]',
                      self.dockerfile)
        self.assertIn('"/admin/api/settings"', wrapper)
        # Re-states the model every start, keeping an existing price.
        self.assertIn("keep = existing if existing and priced(existing) else None", wrapper)
        self.assertIn('pricing = keep["pricing"] if keep else', wrapper)
        # Name and answer limit belong to the operator ("What you sell");
        # only the context, which llama-server fixes, is re-stated.
        self.assertIn('name = existing.get("name") or name', wrapper)
        self.assertIn("max_completion = min(max_completion, context)", wrapper)
        self.assertEqual(self.env["ARCHY_MODEL_ID"], "archy-llama")
        for key in ("ARCHY_DEFAULT_PROMPT_USD_PER_TOKEN", "ARCHY_DEFAULT_COMPLETION_USD_PER_TOKEN"):
            self.assertGreater(float(self.env[key]), 0)
        self.assertEqual(self.env["ENABLE_ANALYTICS_SHARING"], "false")
        self.assertEqual(self.env["UPSTREAM_5XX_RETRY_ATTEMPTS"], "0")

    def test_advertised_context_and_timeouts_match_the_runtime(self):
        llama = yaml.safe_load((ROOT / "apps/llama-server/manifest.yml").read_text())
        args = llama["app"]["container"]["custom_args"]
        self.assertEqual(args[args.index("--ctx-size") + 1], self.env["ARCHY_MODEL_CONTEXT"])
        self.assertNotIn("--threads", args)
        # Routstr keeps an X-Cashu token without saving change when its caller
        # hangs up, so it must time out (and refund) before the node gateway
        # (260s) does.
        self.assertLess(float(self.env["UPSTREAM_READ_TIMEOUT"]), 260)

    def test_base_images_are_pinned_by_digest(self):
        for dockerfile in ("docker/routstr/Dockerfile", "docker/llama-server/Dockerfile"):
            for line in (ROOT / dockerfile).read_text().splitlines():
                if line.startswith("FROM "):
                    self.assertRegex(line.split()[1], r"@sha256:[0-9a-f]{64}$", dockerfile)

    def test_build_has_immutable_source_and_locked_dependencies(self):
        self.assertIn(f"git fetch --depth 1 origin {SOURCE_REVISION}", self.dockerfile)
        self.assertIn(f'"$(git rev-parse HEAD)" = {SOURCE_REVISION}', self.dockerfile)
        self.assertIn("uv sync --frozen --no-dev --no-install-project", self.dockerfile)
        self.assertNotIn(":latest", self.dockerfile)
        self.assertNotIn("image", self.app["container"])

    def test_build_context_matches_runtime_payload(self):
        context = self.app["container"]["build"]["context"]
        self.assertEqual(context, "/opt/archipelago/docker/routstr")
        self.assertTrue((ROOT / context.removeprefix("/opt/archipelago/") / "Dockerfile").is_file())

    def test_migrations_resolve_independently_of_data_workdir(self):
        # cwd is /data; a cwd-relative script_location fails every startup.
        self.assertIn("script_location = %(here)s/migrations", self.dockerfile)

    def test_health_does_not_claim_paid_inference_readiness(self):
        self.assertEqual(self.app["health_check"]["type"], "tcp")
        self.assertEqual(self.app["health_check"]["endpoint"], "localhost:8000")


class LlamaPackageTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.app = yaml.safe_load((ROOT / "apps/llama-server/manifest.yml").read_text())["app"]
        cls.dockerfile = (ROOT / "docker/llama-server/Dockerfile").read_text()

    def test_runtime_has_no_host_listener_or_browser_surface(self):
        self.assertEqual(self.app["ports"], [])
        self.assertNotIn("interfaces", self.app)
        self.assertNotEqual(self.app["container"]["network"], "host")
        self.assertIn("--no-webui", self.app["container"]["custom_args"])
        self.assertEqual(self.app["health_check"]["path"], "/health")

    def test_runtime_requires_dedicated_api_secret(self):
        self.assertEqual(self.app["container"]["secret_env"], [
            {"key": "LLAMA_API_KEY", "secret_file": "llama-server-api-key"}])
        self.assertEqual(self.app["container"]["generated_secrets"], [
            {"name": "llama-server-api-key", "kind": "hex32"}])

    def test_model_is_explicit_local_readonly_input(self):
        binds = [v for v in self.app["volumes"] if v["type"] == "bind"]
        self.assertEqual(binds, [{"type": "bind",
            "source": "/var/lib/archipelago/llama-server/models",
            "target": "/models", "options": ["ro"]}])
        args = self.app["container"]["custom_args"]
        self.assertEqual(args[args.index("--model") + 1], "/models/model.gguf")
        self.assertNotIn("--hf-repo", args)
        self.assertTrue(self.app["security"]["readonly_root"])
        self.assertTrue(self.app["security"]["no_new_privileges"])
        self.assertEqual(self.app["security"]["capabilities"], [])

    def test_cpu_build_has_fixed_source_and_no_downloaded_ui(self):
        revision = "fdb1db877c526ec90f668eca1b858da5dba85560"
        self.assertIn(f"git fetch --depth 1 origin {revision}", self.dockerfile)
        self.assertIn(f'"$(git rev-parse HEAD)" = {revision}', self.dockerfile)
        for option in ("-DGGML_NATIVE=OFF", "-DLLAMA_BUILD_UI=OFF", "-DLLAMA_USE_PREBUILT_UI=OFF"):
            self.assertIn(option, self.dockerfile)
        self.assertNotIn(":latest", self.dockerfile)
        context = self.app["container"]["build"]["context"]
        self.assertEqual(context, "/opt/archipelago/docker/llama-server")
        self.assertTrue((ROOT / context.removeprefix("/opt/archipelago/") / "Dockerfile").is_file())

    def test_cpu_build_runs_on_every_x86_64_tier(self):
        # A single AVX2 build dies with SIGILL on nodes without AVX.
        for option in ("-DGGML_BACKEND_DL=ON", "-DGGML_CPU_ALL_VARIANTS=ON"):
            self.assertIn(option, self.dockerfile)
        self.assertNotIn("-DBUILD_SHARED_LIBS=OFF", self.dockerfile)
        self.assertIn("LD_LIBRARY_PATH=/opt/llama", self.dockerfile)


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