supaspin/tests/test_supaspin.py

"""Unit tests for the pure parts of supaspin. Run: python3 -m unittest discover -s tests"""
import base64
import hashlib
import hmac
import importlib.machinery
import importlib.util
import collections
import contextlib
import io
import json
import os
import random
import re
import shutil
import sys
import tempfile
import unittest
from unittest import mock

import yaml

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
loader = importlib.machinery.SourceFileLoader("supaspin", os.path.join(ROOT, "supaspin"))
spec = importlib.util.spec_from_loader("supaspin", loader)
sp = importlib.util.module_from_spec(spec)
loader.exec_module(sp)

CFG = {
    "storage_root": "/data/supabase", "node_name": "node1", "lan_ip": "192.168.10.30",
    "acme_email": "a@b.c", "ingress_class": "public",
    "slots": {1: "supa1.example.com", 2: "supa2.example.com", 3: "supa3.example.com"},
    "ssh_host": "203.0.113.5", "ssh_port": 2222, "ssh_user": "igor",
    "chart_version": "0.8.0",
}
MAIL = {"host": "smtp.x.com", "port": 587, "user": "u", "password": "p",
        "admin_email": "admin@x.com", "sender_name": "Sender"}


def template():
    with open(os.path.join(ROOT, "templates", "values.yaml.tpl")) as f:
        return f.read()


class Numbering(unittest.TestCase):
    def test_lowest_free_number_fills_gaps(self):
        self.assertEqual(sp.lowest_free_number(CFG["slots"], set()), 1)
        self.assertEqual(sp.lowest_free_number(CFG["slots"], {1, 3}), 2)
        self.assertEqual(sp.lowest_free_number(CFG["slots"], {1, 2}), 3)

    def test_no_free_slot(self):
        self.assertIsNone(sp.lowest_free_number(CFG["slots"], {1, 2, 3}))

    def test_slots_are_keyed_by_the_hostname_number(self):
        # The digit in the name is the slot number, so instance N, its ports and the
        # hostname stay in step on every machine.
        self.assertEqual(sp.parse_slots("supa4.didgibot.com, supa5.didgibot.com"),
                         {4: "supa4.didgibot.com", 5: "supa5.didgibot.com"})

    def test_plain_names_take_their_position(self):
        self.assertEqual(sp.parse_slots("db.example.com, db2.example.com"),
                         {1: "db.example.com", 2: "db2.example.com"})
        self.assertEqual(sp.parse_slots("supa4.didgibot.com, spare.example.com"),
                         {4: "supa4.didgibot.com", 2: "spare.example.com"})

    def test_duplicates_are_an_error(self):
        with self.assertRaises(ValueError):
            sp.parse_slots("supa4.didgibot.com, supa4.other.com")

    def test_empty_answer_gives_no_slots(self):
        self.assertEqual(sp.parse_slots(""), {})

    def test_ports_and_names(self):
        self.assertEqual((sp.pg_port(1), sp.pg_port(3)), (5441, 5443))
        self.assertEqual((sp.studio_port(1), sp.studio_port(3)), (3001, 3003))
        self.assertEqual(sp.namespace(2), "supaspin-2")
        self.assertEqual(sp.pvc_name(2, "db"), "supa2-supabase-db")
        self.assertEqual(sp.tls_secret(2), "supa2-tls")


class Jwt(unittest.TestCase):
    def test_signature_and_claims(self):
        tok = sp.sign_jwt("s" * 40, "anon", now=1000)
        h, p, s = tok.split(".")
        pad = lambda x: x + "=" * (-len(x) % 4)
        self.assertEqual(json.loads(base64.urlsafe_b64decode(pad(h))), {"alg": "HS256", "typ": "JWT"})
        claims = json.loads(base64.urlsafe_b64decode(pad(p)))
        self.assertEqual((claims["role"], claims["iss"], claims["iat"]), ("anon", "supabase", 1000))
        self.assertGreater(claims["exp"], 1000 + 9 * 365 * 86400)
        want = hmac.new(b"s" * 40, f"{h}.{p}".encode(), hashlib.sha256).digest()
        self.assertEqual(base64.urlsafe_b64decode(pad(s)), want)

    def test_generated_secrets_are_consistent(self):
        s = sp.generate_secrets()
        self.assertEqual(len(s["REALTIME_DB_ENC_KEY"]), 16)
        self.assertGreaterEqual(len(s["JWT_SECRET"]), 32)
        self.assertGreaterEqual(len(s["META_CRYPTO_KEY"]), 32)
        h, p, sig = s["ANON_KEY"].split(".")
        want = hmac.new(s["JWT_SECRET"].encode(), f"{h}.{p}".encode(), hashlib.sha256).digest()
        pad = lambda x: x + "=" * (-len(x) % 4)
        self.assertEqual(base64.urlsafe_b64decode(pad(sig)), want)
        self.assertNotEqual(sp.generate_secrets()["DB_PASSWORD"], s["DB_PASSWORD"])


class Values(unittest.TestCase):
    def setUp(self):
        self.sec = sp.generate_secrets()
        self.v = sp.instance_values(CFG, 2, self.sec, MAIL, template())

    def test_memory_caps_add_up(self):
        req, lim = sp.total_memory_mib()
        # db 4096, analytics 1536, kong 768, realtime 768, vector 512, the other seven 512 each
        self.assertEqual(lim, 4096 + 1536 + 768 + 768 + 512 + 7 * 512)
        self.assertEqual(lim / 1024, 11.0)
        self.assertEqual(req, 1024 + 11 * 128)  # 2.375 GiB
        # Measured on three running instances on 2026-09-22; the ceiling has to stay clear of the
        # highest use, or the kernel kills the container in a loop the way it did on instance 1.
        self.assertGreaterEqual(sp.MEMORY["analytics"][1], 1536)  # used 694 MiB, died at 768
        self.assertGreaterEqual(sp.MEMORY["vector"][1], 512)      # used 290 MiB; the nightly log
        #  retention keeps its delivery backlog from growing past this the way it did on
        #  2026-10-06, when instance 3's full db image killed analytics and vector filled with
        #  the undeliverable backlog every few minutes against 1024 MiB.
        self.assertGreaterEqual(sp.MEMORY["realtime"][1], 768)    # used 300 MiB against 384
        self.assertGreaterEqual(sp.MEMORY["kong"][1], 768)        # used 269 MiB against 512
        self.assertGreaterEqual(sp.DEFAULT_MEMORY[1], 512)        # studio used 203 against 256
        # The requests stay where they were: the node is shared with production and reserving
        # what the instances really use would leave production nothing to schedule into.
        # Only db's went up, from 512 to 1024 MiB, with its limit (2026-11-03).
        self.assertTrue(all(m[0] == 128 for s, m in sp.MEMORY.items() if s != "db"))
        self.assertEqual(req / 1024, 2.375)
        for svc in sp.SERVICES:
            self.assertEqual(self.v["deployment"][svc]["priorityClassName"], "supaspin-low")
            self.assertIn("memory", self.v["deployment"][svc]["resources"]["limits"])

    def test_all_twelve_services_enabled(self):
        self.assertEqual(len(sp.SERVICES), 12)
        self.assertTrue(self.v["deployment"]["analytics"]["enabled"])
        self.assertTrue(self.v["deployment"]["vector"]["enabled"])

    def test_no_group_override_on_db_pod(self):
        # The db image's own user is 100:101; a forced fsGroup would lock Postgres out of its volume.
        self.assertNotIn("podSecurityContext", self.v["deployment"]["db"])
        self.assertEqual((sp.DB_UID, sp.DB_GID), (100, 101))

    def test_volume_caps(self):
        p = self.v["persistence"]
        self.assertEqual(p["db"]["size"], "12288Mi")
        self.assertEqual(p["storage"]["size"], "1024Mi")
        for vol in ("functions", "snippets", "deno", "pgsodium"):
            self.assertEqual(p[vol]["size"], "256Mi")
        self.assertEqual(set(sp.VOLUMES), {"db", "storage", "functions", "snippets", "deno", "pgsodium"})
        self.assertTrue(all(p[v]["storageClassName"] == "supaspin-local-sc" for v in sp.VOLUMES))

    def test_log_retention_sql(self):
        # Logflare stores logs in _analytics.log_events_* and never deletes them; the job deletes
        # rows older than LOG_RETENTION and is scheduled under a fixed name so re-runs overwrite.
        # pg_cron in the supabase image only runs jobs from the `postgres` database, so the job
        # reaches the log tables in `_supabase` over a local dblink connection (trusted there).
        # dblink and the helper function live in the `extensions` schema: in `public` they belong
        # to someone else's routines and the app's `REVOKE ALL ON ALL ROUTINES IN SCHEMA public`
        # fails as the non-superuser postgres role (42501, dblink_connect_u).
        self.assertIn("schemaname = ''_analytics''", sp.RETENTION_SQL)
        self.assertIn("''log_events\\_%''", sp.RETENTION_SQL)
        self.assertIn(f"interval ''{sp.LOG_RETENTION}''", sp.RETENTION_SQL)
        self.assertIn(f"'{sp.LOG_RETENTION_JOB}'", sp.RETENTION_SQL)
        self.assertIn("cron.schedule", sp.RETENTION_SQL)
        self.assertIn("extensions.dblink_connect", sp.RETENTION_SQL)
        self.assertIn("user=supabase_admin", sp.RETENTION_SQL)
        self.assertIn("create extension if not exists dblink with schema extensions", sp.RETENTION_SQL)
        self.assertIn("alter extension dblink set schema extensions", sp.RETENTION_SQL)
        self.assertIn("extensions.supaspin_delete_old_logs", sp.RETENTION_SQL)
        self.assertIn("drop function if exists public.supaspin_delete_old_logs", sp.RETENTION_SQL)
        self.assertNotIn("\ndblink_", sp.RETENTION_SQL)  # no unqualified dblink call remains

    def test_imgproxy_reads_the_storage_volume(self):
        img = self.v["persistence"]["imgproxy"]
        self.assertEqual(img["existingClaim"], "supa2-supabase-storage")
        self.assertNotIn("imgproxy", sp.VOLUMES)  # no separate image file for it

    def test_ingress_and_urls(self):
        ing = self.v["ingress"]
        self.assertEqual(ing["hosts"][0]["host"], "supa2.example.com")
        self.assertEqual(ing["tls"][0]["secretName"], "supa2-tls")
        self.assertEqual(ing["className"], "public")
        self.assertIsNone(ing["annotations"]["nginx.ingress.kubernetes.io/rewrite-target"])
        env = {e["name"]: e["value"] for e in self.v["environment"]["auth"]}
        self.assertEqual(env["API_EXTERNAL_URL"], "https://supa2.example.com")
        self.assertEqual(env["GOTRUE_SMTP_HOST"], "smtp.x.com")
        self.assertEqual(env["GOTRUE_SMTP_PORT"], "587")
        self.assertEqual(self.v["secret"]["smtp"]["password"], "p")

    def test_secrets_reach_values(self):
        self.assertEqual(self.v["secret"]["jwt"]["secret"], self.sec["JWT_SECRET"])
        self.assertEqual(self.v["secret"]["db"]["password"], self.sec["DB_PASSWORD"])

    def test_special_characters_stay_strings(self):
        m = dict(MAIL, password='p"a$s: #x', sender_name="A & B")
        v = sp.instance_values(CFG, 1, self.sec, m, template())
        self.assertEqual(v["secret"]["smtp"]["password"], 'p"a$s: #x')


class IngressShaping(unittest.TestCase):
    """The ingress annotation map per controller and per Let's Encrypt setting."""
    def setUp(self):
        self.sec = sp.generate_secrets()

    def ann(self, cfg=None, n=2):
        return sp.instance_values(dict(CFG, **(cfg or {})), n, self.sec, MAIL, template()) \
            ["ingress"]["annotations"]

    def test_default_is_nginx_with_lets_encrypt(self):
        # Configs written before these settings existed must render exactly as before.
        a = self.ann()
        self.assertEqual(a["cert-manager.io/cluster-issuer"], "supaspin-letsencrypt")
        self.assertIsNone(a["nginx.ingress.kubernetes.io/rewrite-target"])
        self.assertEqual(a["nginx.ingress.kubernetes.io/proxy-body-size"], "50m")
        self.assertEqual(a["nginx.ingress.kubernetes.io/proxy-read-timeout"], "3600")
        self.assertNotIn("traefik.ingress.kubernetes.io/router.middlewares", a)

    def test_traefik_also_keeps_nginx_annotations(self):
        # The nginx keys must be present in traefik mode too: while nginx routes the
        # ingress they enforce the cap, and they are inert the day Traefik takes over.
        a = self.ann({"ingress_controller": "traefik"})
        self.assertEqual(a["traefik.ingress.kubernetes.io/router.middlewares"],
                         "supaspin-2-supa2-buffering@kubernetescrd")
        self.assertIsNone(a["nginx.ingress.kubernetes.io/rewrite-target"])
        self.assertEqual(a["nginx.ingress.kubernetes.io/proxy-body-size"], "50m")
        self.assertEqual(a["nginx.ingress.kubernetes.io/proxy-read-timeout"], "3600")
        self.assertEqual(a["cert-manager.io/cluster-issuer"], "supaspin-letsencrypt")

    def test_lets_encrypt_off_drops_cert_manager_annotation(self):
        self.assertNotIn("cert-manager.io/cluster-issuer", self.ann({"lets_encrypt": False}))
        self.assertNotIn("cert-manager.io/cluster-issuer",
                         self.ann({"lets_encrypt": False, "ingress_controller": "traefik"}))

    def test_middleware_manifest(self):
        m = sp.middleware_manifest(2)
        self.assertEqual(m["apiVersion"], "traefik.io/v1alpha1")
        self.assertEqual(m["kind"], "Middleware")
        self.assertEqual(m["metadata"], {"name": "supa2-buffering", "namespace": "supaspin-2"})
        self.assertEqual(m["spec"]["buffering"]["maxRequestBodyBytes"], 50 * 1024 * 1024)


class ClusterObjects(unittest.TestCase):
    """ensure_cluster_objects skips the ClusterIssuer when Let's Encrypt is off."""
    def created_docs(self, cfg):
        applied = []

        class R:
            returncode = 1
            stdout = stderr = ""

        def fake_kubectl(*args, **kw):
            if args and args[0] == "apply":
                applied.extend(yaml.safe_load_all(kw["input_"]))
            return R()

        with tempfile.TemporaryDirectory() as tmp, \
                mock.patch.object(sp, "STATE_DIR", tmp), \
                mock.patch.object(sp, "kubectl", side_effect=fake_kubectl), \
                mock.patch.object(sp, "helm"):
            sp.ensure_cluster_objects(cfg, lambda *a: None)
        return applied

    def test_issuer_created_when_lets_encrypt_on(self):
        self.assertIn("ClusterIssuer", [d["kind"] for d in self.created_docs(CFG)])

    def test_no_issuer_without_lets_encrypt(self):
        kinds = [d["kind"] for d in self.created_docs(dict(CFG, lets_encrypt=False))]
        self.assertNotIn("ClusterIssuer", kinds)
        self.assertIn("StorageClass", kinds)
        self.assertIn("PriorityClass", kinds)
        self.assertIn("Namespace", kinds)


class SelfSignedCert(unittest.TestCase):
    def test_generated_when_missing(self):
        calls = []

        class R:
            returncode = 1
            stdout = stderr = ""

        def fake_kubectl(*args, **kw):
            calls.append(args)
            return R()

        def fake_run(args, **kw):
            calls.append(("run", args))

            class P:
                returncode = 0
            return P()

        with mock.patch.object(sp, "kubectl", side_effect=fake_kubectl), \
                mock.patch.object(sp, "run", side_effect=fake_run):
            sp.ensure_self_signed_cert(CFG, 2, "supa2.example.com", lambda *a: None)
        openssl = next(c[1] for c in calls if c[0] == "run" and c[1][0] == "openssl")
        self.assertIn("subjectAltName=DNS:supa2.example.com", openssl)
        create = next(c for c in calls if c[:3] == ("create", "secret", "tls"))
        self.assertEqual(create[3], "supa2-tls")
        self.assertIn("supaspin-2", create)

    def test_kept_when_present(self):
        class R:
            returncode = 0
            stdout = stderr = ""

        with mock.patch.object(sp, "kubectl", return_value=R()) as k, \
                mock.patch.object(sp, "run") as r:
            sp.ensure_self_signed_cert(CFG, 2, "supa2.example.com", lambda *a: None)
            k.assert_called_once()
            r.assert_not_called()


class EnvFile(unittest.TestCase):
    def setUp(self):
        self.sec = sp.generate_secrets()
        self.text = sp.render_env(CFG, 1, self.sec, MAIL)
        self.kv = dict(l.split("=", 1) for l in self.text.splitlines() if "=" in l and not l.startswith("#"))

    def test_expected_variables(self):
        for k in ("SUPABASE_URL", "SUPABASE_PUBLIC_URL", "PUBLIC_SUPABASE_URL", "API_EXTERNAL_URL",
                  "PUBLIC_SUPABASE_ANON_KEY", "ANON_KEY", "JWT_ANON_KEY", "SERVICE_ROLE_KEY",
                  "JWT_SERVICE_KEY", "JWT_SECRET", "JWT_EXPIRY", "POSTGRES_PASSWORD", "POSTGRES_DB",
                  "POSTGRES_HOST", "POSTGRES_PORT", "DATABASE_URL", "DASHBOARD_USERNAME",
                  "DASHBOARD_PASSWORD", "GOTRUE_SMTP_HOST", "GOTRUE_SMTP_PORT", "GOTRUE_SMTP_USER",
                  "GOTRUE_SMTP_PASS", "GOTRUE_SMTP_ADMIN_EMAIL", "GOTRUE_SMTP_SENDER_NAME"):
            self.assertIn(k, self.kv)
        self.assertEqual(self.kv["SUPABASE_URL"], "https://supa1.example.com")

    def test_database_url_and_tunnel_comment(self):
        self.assertEqual(self.kv["DATABASE_URL"],
                         f"postgresql://postgres:{self.sec['DB_PASSWORD']}@localhost:5441/postgres")
        self.assertIn("192.168.10.30:5441", self.text)
        self.assertIn("ssh -N -p 2222 -L 5441:192.168.10.30:5441 -L 3001:192.168.10.30:3001 igor@203.0.113.5",
                      self.text)

    def test_mask_hides_secrets_only(self):
        masked = sp.mask_env(self.text)
        self.assertNotIn(self.sec["DB_PASSWORD"], masked)
        self.assertNotIn(self.sec["JWT_SECRET"], masked)
        self.assertIn("SUPABASE_URL=https://supa1.example.com", masked)
        self.assertIn("POSTGRES_PORT=5441", masked)


class Fstab(unittest.TestCase):
    def test_line_and_removal(self):
        line = sp.fstab_line("/r/1/db/db.img", "/r/1/db/data", "/r")
        self.assertEqual(line, "/r/1/db/db.img /r/1/db/data ext4 loop,nofail,x-systemd.requires-mounts-for=/r 0 0")
        text = "UUID=x / ext4 defaults 0 1\n" + line + "\n# " + line + "\n"
        self.assertTrue(sp.fstab_has(text, "/r/1/db/data"))
        out = sp.fstab_without(text, "/r/1/db/data")
        self.assertFalse(sp.fstab_has(out, "/r/1/db/data"))
        self.assertIn("UUID=x / ext4", out)
        self.assertIn("# /r/1/db/db.img", out)  # comments are left alone

    def test_mounts_under_an_instance_folder(self):
        text = ("a /r/1/db/data ext4 x 0 0\n"
                "b /r/1/imgproxy/data ext4 x 0 0\n"   # a volume an older version created
                "c /r/10/db/data ext4 x 0 0\n"        # another instance
                "# d /r/1/old/data ext4 x 0 0\n"
                "e /r/1 ext4 x 0 0\n")                # the folder itself is not below itself
        self.assertEqual(sp.fstab_mounts_under(text, "/r/1"), ["/r/1/db/data", "/r/1/imgproxy/data"])

    def test_similar_paths_are_not_removed(self):
        text = "a /r/1/db/data ext4 x 0 0\nb /r/10/db/data ext4 x 0 0\n"
        self.assertIn("/r/10/db/data", sp.fstab_without(text, "/r/1/db/data"))


RENDERED = """
apiVersion: apps/v1
kind: StatefulSet
metadata: {name: supa1-supabase-db}
spec: {template: {spec: {containers: [{name: db, ports: [{name: http, containerPort: 5432}]}]}}}
---
apiVersion: apps/v1
kind: Deployment
metadata: {name: supa1-supabase-studio}
spec: {template: {spec: {containers: [{name: studio, ports: [{name: http, containerPort: 3000}]}]}}}
---
apiVersion: apps/v1
kind: Deployment
metadata: {name: supa1-supabase-rest}
spec: {template: {spec: {containers: [{name: rest, ports: [{name: http, containerPort: 3000}]}]}}}
---
apiVersion: v1
kind: ConfigMap
metadata: {name: supa1-supabase-kong}
data:
  kong-entrypoint.sh: "echo hi\\necho there\\n"
  kong.yml: |
    _format_version: '2.1'
    services:
      - name: auth-v1
        url: http://auth:9999
        routes: [{name: auth-v1-all, paths: [/auth/v1/]}]
        plugins:
          - name: request-transformer
            config:
              add:
                headers:
                  - "Authorization: ${LUA_AUTH_EXPR}"
      - name: meta
        url: http://meta:8080/
        routes: [{name: meta-all, paths: [/pg/]}]
      - name: dashboard
        url: http://studio:3000/
        routes: [{name: dashboard-all, paths: [/]}]
"""


class PostRender(unittest.TestCase):
    def setUp(self):
        self.docs = list(yaml.safe_load_all(sp.post_render(RENDERED, 1, CFG)))

    def find(self, kind, name):
        return next(d for d in self.docs if d["kind"] == kind and d["metadata"]["name"] == name)

    def test_db_and_studio_get_fixed_host_ports(self):
        db = self.find("StatefulSet", "supa1-supabase-db")["spec"]["template"]["spec"]["containers"][0]["ports"][0]
        self.assertEqual((db["hostPort"], db["hostIP"], db["containerPort"]), (5441, "192.168.10.30", 5432))
        st = self.find("Deployment", "supa1-supabase-studio")["spec"]["template"]["spec"]["containers"][0]["ports"][0]
        self.assertEqual((st["hostPort"], st["hostIP"]), (3001, "192.168.10.30"))

    def test_all_binds_localhost_too(self):
        # `all` omits hostIP: the kubelet then listens on every interface, localhost included.
        docs = list(yaml.safe_load_all(sp.post_render(RENDERED, 1, dict(CFG, port_bind="all"))))
        find = lambda k, name: next(d for d in docs if d["kind"] == k and d["metadata"]["name"] == name)
        db = find("StatefulSet", "supa1-supabase-db")["spec"]["template"]["spec"]["containers"][0]["ports"][0]
        st = find("Deployment", "supa1-supabase-studio")["spec"]["template"]["spec"]["containers"][0]["ports"][0]
        self.assertEqual(db["hostPort"], 5441)
        self.assertNotIn("hostIP", db)
        self.assertEqual(st["hostPort"], 3001)
        self.assertNotIn("hostIP", st)

    def test_local_binds_localhost_only(self):
        docs = list(yaml.safe_load_all(sp.post_render(RENDERED, 1, dict(CFG, port_bind="local"))))
        find = lambda k, name: next(d for d in docs if d["kind"] == k and d["metadata"]["name"] == name)
        db = find("StatefulSet", "supa1-supabase-db")["spec"]["template"]["spec"]["containers"][0]["ports"][0]
        self.assertEqual((db["hostPort"], db["hostIP"]), (5441, "127.0.0.1"))

    def test_other_pods_untouched(self):
        rest = self.find("Deployment", "supa1-supabase-rest")["spec"]["template"]["spec"]["containers"][0]["ports"][0]
        self.assertNotIn("hostPort", rest)

    def test_studio_and_pgmeta_routes_removed(self):
        kong = yaml.safe_load(self.find("ConfigMap", "supa1-supabase-kong")["data"]["kong.yml"])
        self.assertEqual([s["name"] for s in kong["services"]], ["auth-v1"])
        self.assertEqual(kong["_format_version"], "2.1")

    def test_kong_file_is_cut_not_rewritten(self):
        # Kong's start script substitutes ${LUA_AUTH_EXPR} inside these exact double quotes.
        text = self.find("ConfigMap", "supa1-supabase-kong")["data"]["kong.yml"]
        self.assertIn('- "Authorization: ${LUA_AUTH_EXPR}"', text)
        self.assertIn("_format_version: '2.1'", text)
        self.assertNotIn("name: meta", text)
        self.assertNotIn("name: dashboard", text)

    def test_strip_keeps_following_top_level_keys(self):
        src = "services:\n  - name: meta\n    url: x\n  - name: keep\n    url: y\nother: 1\n"
        self.assertEqual(sp.strip_kong_services(src, ("meta",)),
                         "services:\n  - name: keep\n    url: y\nother: 1\n")
        src2 = "services:\n  - name: dashboard\n    url: x\nother: 1\n"
        self.assertEqual(sp.strip_kong_services(src2, ("dashboard",)), "services:\nother: 1\n")

    def test_other_config_keys_survive(self):
        cm = self.find("ConfigMap", "supa1-supabase-kong")
        self.assertEqual(cm["data"]["kong-entrypoint.sh"], "echo hi\necho there\n")

    def test_instance_number_selects_ports(self):
        docs = list(yaml.safe_load_all(sp.post_render(RENDERED.replace("supa1", "supa3"), 3, CFG)))
        db = next(d for d in docs if d["kind"] == "StatefulSet")["spec"]["template"]["spec"]["containers"][0]["ports"][0]
        self.assertEqual((db["hostPort"], db["hostIP"]), (5443, "192.168.10.30"))


class Manifests(unittest.TestCase):
    def test_pv_is_pinned_and_reserved_for_the_chart_claim(self):
        pv = sp.pv_manifest(CFG, 2, "db")
        self.assertEqual(pv["spec"]["claimRef"], {"namespace": "supaspin-2", "name": "supa2-supabase-db"})
        self.assertEqual(pv["spec"]["local"]["path"], "/data/supabase/2/db/data")
        self.assertEqual(pv["spec"]["capacity"]["storage"], "12288Mi")
        self.assertEqual(pv["spec"]["persistentVolumeReclaimPolicy"], "Retain")
        self.assertEqual(pv["spec"]["nodeAffinity"]["required"]["nodeSelectorTerms"][0]
                         ["matchExpressions"][0]["values"], ["node1"])
        self.assertEqual(sp.pv_manifest(CFG, 2, "deno")["spec"]["capacity"]["storage"], "256Mi")

    def test_saved_certificate_keeps_issuer_annotations(self):
        src = {"type": "kubernetes.io/tls",
               "metadata": {"name": "supa1-tls", "namespace": "supaspin-1", "uid": "x",
                            "resourceVersion": "9", "ownerReferences": [{"name": "o"}],
                            "annotations": {"cert-manager.io/issuer-name": "supaspin-letsencrypt",
                                            "cert-manager.io/issuer-kind": "ClusterIssuer",
                                            "cert-manager.io/issuer-group": "cert-manager.io",
                                            "kubectl.kubernetes.io/last-applied-configuration": "{}"}},
               "data": {"tls.crt": "Yg==", "tls.key": "Yw=="}}
        out = sp.carry_secret(src, "supaspin-certs", "supa1-tls")
        self.assertEqual(out["metadata"], {"name": "supa1-tls", "namespace": "supaspin-certs",
                                           "annotations": {"cert-manager.io/issuer-name": "supaspin-letsencrypt",
                                                           "cert-manager.io/issuer-kind": "ClusterIssuer",
                                                           "cert-manager.io/issuer-group": "cert-manager.io"}})
        self.assertEqual(out["data"], src["data"])
        self.assertEqual(out["type"], "kubernetes.io/tls")

    def test_secret_without_annotations_still_copies(self):
        out = sp.carry_secret({"data": {"tls.crt": "a"}}, "ns", "n")
        self.assertNotIn("annotations", out["metadata"])
        self.assertEqual(out["type"], "kubernetes.io/tls")

    def test_tunnel_help_mentions_all_steps(self):
        text = sp.tunnel_help(CFG, 1, "PW")
        self.assertIn("postgresql://postgres:PW@localhost:5441/postgres", text)
        self.assertIn("http://localhost:3001", text)
        self.assertIn("192.168.10.30:5441", text)


class Version(unittest.TestCase):
    def test_version_file_code_and_readme_agree(self):
        with open(os.path.join(ROOT, "VERSION")) as f:
            version = f.read().strip()
        self.assertRegex(version, r"^\d+\.\d+\.\d+$")
        self.assertEqual(sp.VERSION, version)
        with open(os.path.join(ROOT, "README.md")) as f:
            self.assertEqual(f.read().splitlines()[2], f"Version {version}")

    def test_version_command_names_the_app(self):
        out = io.StringIO()
        with mock.patch.object(sp.os, "geteuid", return_value=0), mock.patch.object(sp, "load_config", return_value=CFG), \
                contextlib.redirect_stdout(out):
            self.assertEqual(sp.main(["supaspin", "--version"]), 0)
        self.assertEqual(out.getvalue().strip(), f"supaspin {sp.VERSION}")


class TunnelGuide(unittest.TestCase):
    def test_list_guide_has_the_three_commands_for_this_instance(self):
        text = "\n".join(sp.tunnel_guide(CFG, 2))
        self.assertIn("ssh -N -p 2222 -L 5442:192.168.10.30:5442 -L 3002:192.168.10.30:3002 igor@203.0.113.5", text)
        self.assertIn("ssh -f -N -M -S /tmp/supaspin-2.sock -o ExitOnForwardFailure=yes -p 2222 "
                      "-L 5442:192.168.10.30:5442 -L 3002:192.168.10.30:3002 igor@203.0.113.5", text)
        self.assertIn("ssh -S /tmp/supaspin-2.sock -O exit igor@203.0.113.5", text)
        self.assertIn("If you want to open the tunnel", text)
        self.assertIn("If you want the tunnel to keep running in the background", text)
        self.assertIn("If you no longer need the background tunnel", text)

    def test_windows_limit_is_written_down(self):
        for detailed in (False, True):
            text = "\n".join(sp.tunnel_guide(CFG, 1, detailed))
            self.assertIn("do NOT work on Windows", text)
            self.assertIn("ONLY the first command below", text)
            self.assertIn("(Mac or Linux only)", text)
        self.assertIn("do NOT work on Windows", sp.tunnel_help(CFG, 1))

    def test_list_guide_is_short_and_points_to_the_detailed_help(self):
        text = "\n".join(sp.tunnel_guide(CFG, 1))
        self.assertIn("sudo supaspin tunnel 1", text)
        self.assertNotIn("pkill", text)
        self.assertNotIn("-O check", text)

    def test_detailed_guide_adds_check_and_leftover_tunnel(self):
        text = "\n".join(sp.tunnel_guide(CFG, 1, detailed=True))
        self.assertIn("ssh -S /tmp/supaspin-1.sock -O check igor@203.0.113.5", text)
        self.assertIn('pkill -f "[s]upaspin-1.sock"', text)
        self.assertIn("stops the command from also matching", text)

    def test_without_public_ssh_host_it_says_how_to_set_it(self):
        text = "\n".join(sp.tunnel_guide(dict(CFG, ssh_host=""), 1))
        self.assertIn("Public SSH host for tunnels", text)
        self.assertNotIn("ssh -N", text)

    def test_list_prints_the_guide_under_each_instance(self):
        rows = [{"n": 1, "host": "supa1.example.com", "pods": "12/12", "memory": "1/2 MiB", "disk": "db 1/2M",
                 "env": "/x/1/.env"}]
        out = io.StringIO()
        with mock.patch.object(sp, "list_rows", return_value=rows), contextlib.redirect_stdout(out):
            sp.print_list(CFG)
        text = out.getvalue()
        self.assertIn("supa1.example.com", text)
        self.assertIn("ssh -N -p 2222 -L 5441:192.168.10.30:5441", text)
        self.assertIn("ssh -S /tmp/supaspin-1.sock -O exit", text)
        self.assertIn("do NOT work on Windows", text)


class TargetParsing(unittest.TestCase):
    def test_local_paths(self):
        self.assertEqual(sp.parse_target("/srv/app/.env"), {"kind": "local", "path": "/srv/app/.env"})
        self.assertEqual(sp.parse_target("./app/.env")["kind"], "local")
        self.assertEqual(sp.parse_target("~/app/.env", "/home/igor")["path"], "/home/igor/app/.env")
        self.assertEqual(sp.parse_target("app/.env"), {"kind": "local", "path": "app/.env"})

    def test_remote_paths(self):
        self.assertEqual(sp.parse_target("igor@192.168.10.50:/Users/igor/app/.env"),
                         {"kind": "ssh", "user": "igor", "host": "192.168.10.50", "path": "/Users/igor/app/.env"})
        self.assertEqual(sp.parse_target("mac.local:app/.env")["user"], None)
        self.assertEqual(sp.parse_target("u@[fe80::1]:/x/.env")["host"], "[fe80::1]")

    def test_windows_paths_are_converted(self):
        for typed in ("igor@pc:C:\\Users\\igor\\app\\.env", "igor@pc:C:/Users/igor/app/.env"):
            self.assertEqual(sp.parse_target(typed)["path"], "/C:/Users/igor/app/.env")

    def test_windows_path_without_computer_is_refused(self):
        with self.assertRaises(ValueError) as cm:
            sp.parse_target("C:\\Users\\igor\\.env")
        self.assertIn("Windows path without the computer's name", str(cm.exception))

    def test_nothing_typed(self):
        with self.assertRaises(ValueError):
            sp.parse_target("  ")


PAIRS = [("A", "1"), ("B", "2"), ("C", "3")]


class MergeEnv(unittest.TestCase):
    def merge(self, data, pairs=PAIRS):
        return sp.merge_env(data, pairs)

    def test_replaces_existing_and_adds_missing_at_the_end(self):
        new, rep = self.merge(b"A=old\nX=keep\n")
        self.assertEqual(new, b"A=1\nX=keep\nB=2\nC=3\n")
        self.assertEqual((rep["replaced"], rep["added"], rep["same"]), (["A"], ["B", "C"], []))

    def test_commented_out_keys_are_not_taken_into_consideration(self):
        data = b"# A=old\n#B=x\n   # C=3\nX=1\n"
        new, rep = self.merge(data)
        self.assertEqual(new, data + b"A=1\nB=2\nC=3\n")
        self.assertEqual(rep["added"], ["A", "B", "C"])
        self.assertEqual(rep["comments"], 3)

    def test_export_prefix_and_spacing_are_kept(self):
        new, _ = self.merge(b"export A=old\nB = old\n  C=old\n")
        self.assertEqual(new, b"export A=1\nB = 2\n  C=3\n")

    def test_key_twice_is_replaced_in_both_places(self):
        new, rep = self.merge(b"A=1\nA=2\n", [("A", "1")])
        self.assertEqual(new, b"A=1\nA=1\n")
        self.assertEqual(rep["replaced"], ["A"])

    def test_similar_key_names_are_not_touched(self):
        data = b"JWT_SECRET_OLD=1\nJWT_SECRETX=2\njwt_secret=3\nJWT_SECRET=old\n"
        new, _ = self.merge(data, [("JWT_SECRET", "s")])
        self.assertEqual(new, b"JWT_SECRET_OLD=1\nJWT_SECRETX=2\njwt_secret=3\nJWT_SECRET=s\n")

    def test_windows_line_endings(self):
        new, _ = self.merge(b"A=old\r\nX=1\r\n")
        self.assertEqual(new, b"A=1\r\nX=1\r\nB=2\r\nC=3\r\n")

    def test_missing_final_newline(self):
        self.assertEqual(self.merge(b"X=1")[0], b"X=1\nA=1\nB=2\nC=3\n")
        self.assertEqual(self.merge(b"A=old", [("A", "1")])[0], b"A=1")  # nothing appended: stays as it was

    def test_byte_order_mark_is_kept(self):
        new, _ = self.merge(b"\xef\xbb\xbfA=old\n")
        self.assertTrue(new.startswith(b"\xef\xbb\xbfA=1\n"))

    def test_empty_file(self):
        self.assertEqual(self.merge(b"")[0], b"A=1\nB=2\nC=3\n")

    def test_multi_line_quoted_value_is_not_looked_into(self):
        data = b'KEY="line1\nA=inside\nline3"\nA=old\n'
        new, rep = self.merge(data)
        self.assertEqual(new, b'KEY="line1\nA=inside\nline3"\nA=1\nB=2\nC=3\n')
        self.assertEqual(rep["untouched"], 3)

    def test_multi_line_value_of_an_instance_key_is_replaced_whole(self):
        new, _ = self.merge(b'A="x\ny"\nX=1\n', [("A", "1")])
        self.assertEqual(new, b"A=1\nX=1\n")

    def test_unclosed_quote_does_not_swallow_the_rest(self):
        new, _ = self.merge(b'X="oops\nA=old\n', [("A", "1")])
        self.assertEqual(new, b'X="oops\nA=1\n')

    def test_values_are_copied_verbatim(self):
        value = "postgresql://postgres:pw@localhost:5441/postgres?a=b#c d"
        new, _ = self.merge(b"", [("DATABASE_URL", value)])
        self.assertEqual(new, f"DATABASE_URL={value}\n".encode())

    def test_empty_instance_value_is_written_empty(self):
        self.assertEqual(self.merge(b"U=real\n", [("U", "")])[0], b"U=\n")

    def test_text_after_the_old_value_goes_with_it(self):
        self.assertEqual(self.merge(b"A=old # note\n", [("A", "1")])[0], b"A=1\n")

    def test_odd_bytes_elsewhere_survive(self):
        data = b"X=caf\xe9\n# \xff\xfe\nA=old\n"
        new, _ = self.merge(data, [("A", "1")])
        self.assertEqual(new, b"X=caf\xe9\n# \xff\xfe\nA=1\n")

    def test_second_run_changes_nothing(self):
        first, _ = self.merge(b"A=old\nX=1\n")
        second, rep = self.merge(first)
        self.assertEqual(second, first)
        self.assertEqual((rep["replaced"], rep["added"], rep["same"]), ([], [], ["A", "B", "C"]))

    def test_utf16_file_is_refused(self):
        with self.assertRaises(sp.TargetError) as cm:
            self.merge("A=1\n".encode("utf-16"))
        self.assertIn("UTF-16", cm.exception.why)

    def test_report_counts(self):
        _, rep = self.merge(b"# c\n\nX=1\nA=old\n")
        self.assertEqual((rep["lines"], rep["untouched"], rep["comments"]), (4, 3, 1))

    def test_instance_env_pairs_skip_comments(self):
        pairs = sp.env_pairs(sp.render_env(CFG, 1, sp.generate_secrets(), MAIL))
        self.assertEqual(len(pairs), 24)
        self.assertEqual(pairs[0], ("SUPABASE_URL", "https://supa1.example.com"))


class CheckMerge(unittest.TestCase):
    BEFORE = b"# note\nX=1\nA=old\n\nY=2\n"

    def test_a_good_merge_passes(self):
        after, _ = sp.merge_env(self.BEFORE, PAIRS)
        self.assertEqual(sp.check_merge(self.BEFORE, after, PAIRS), [])

    def test_a_lost_unrelated_key_is_caught(self):
        after, _ = sp.merge_env(self.BEFORE, PAIRS)
        problems = sp.check_merge(self.BEFORE, after.replace(b"Y=2\n", b""), PAIRS)
        self.assertTrue(problems)

    def test_a_changed_unrelated_value_is_caught(self):
        after, _ = sp.merge_env(self.BEFORE, PAIRS)
        self.assertTrue(sp.check_merge(self.BEFORE, after.replace(b"X=1", b"X=9"), PAIRS))

    def test_a_missing_or_wrong_instance_value_is_caught(self):
        after, _ = sp.merge_env(self.BEFORE, PAIRS)
        self.assertIn("B is missing", sp.check_merge(self.BEFORE, after.replace(b"B=2\n", b""), PAIRS))
        self.assertIn("A does not have the instance's value",
                      sp.check_merge(self.BEFORE, after.replace(b"A=1", b"A=7"), PAIRS))

    def test_random_files_never_lose_an_unrelated_key(self):
        rnd = random.Random(7)
        pool = [b"# A=old", b"#B=c", b"  # C=3", b"", b"X=1", b"Y=two words", b"export Z=3", b"A=old",
                b"B = old", b"export C=old", b'Q="multi', b"A=inside", b'end"', b"JWT=1", b"W='a b'", b"e=1"]
        for _ in range(300):
            lines = [rnd.choice(pool) for _ in range(rnd.randint(0, 12))]
            eol = rnd.choice([b"\n", b"\r\n"])
            data = eol.join(lines) + rnd.choice([b"", eol])
            new, _ = sp.merge_env(data, PAIRS)
            self.assertEqual(sp.check_merge(data, new, PAIRS), [], data)
            again, rep = sp.merge_env(new, PAIRS)
            self.assertEqual(again, new)
            self.assertEqual(rep["replaced"] + rep["added"], [])


class FakeTarget:
    """A file that lives in memory, for testing what push_env does when things go wrong."""

    def __init__(self, data, mode="ok"):
        self.data, self.mode, self.label, self.dead, self.closed = data, mode, "fake:/app/.env", False, False

    def read(self):
        if self.dead:
            raise sp.TargetError("the connection is gone.")
        return self.data

    def write(self, data):
        if self.dead:
            raise sp.TargetError("the connection is gone.")
        if self.mode == "refuse":
            raise sp.TargetError("not allowed to write.")
        if self.mode == "drop-a-line":  # goes wrong once; the write that puts the original back is fine
            data, self.mode = data.replace(b"KEEP=1\n", b""), "ok"
        elif self.mode == "die-half-way":
            self.data, self.dead = data[:5], True
            raise sp.TargetError("the connection dropped while writing.")
        self.data = data

    def close(self):
        self.closed = True


class PushEnv(unittest.TestCase):
    def setUp(self):
        self.dir = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, self.dir, True)
        self.cfg = dict(CFG, storage_root=self.dir)
        os.makedirs(f"{self.dir}/1")
        self.secret = "S3cr3tValue"
        with open(f"{self.dir}/1/.env", "w") as f:
            f.write(f"# instance 1\nSUPABASE_URL=https://supa1.example.com\nJWT_SECRET={self.secret}\n")
        self.tmps = []
        real = sp.private_tmpdir

        def tracked(owner):
            d = real(owner)
            self.tmps.append(d)
            return d
        for patcher in (mock.patch.object(sp, "private_tmpdir", tracked),
                        mock.patch.object(sp, "invoking_user", return_value=None)):
            patcher.start()
            self.addCleanup(patcher.stop)
        self.addCleanup(lambda: [shutil.rmtree(d, True) for d in self.tmps])
        self.out = []

    def run_push(self, spec, dry_run=False):
        return sp.push_env(self.cfg, 1, spec, self.out.append, dry_run)

    def text(self):
        return re.sub(r"\x1b\[[0-9;]*m", "", "\n".join(self.out))

    def local_target(self, content):
        path = f"{self.dir}/target.env"
        with open(path, "wb") as f:
            f.write(content)
        return path

    def fake(self, data, mode="ok"):
        target = FakeTarget(data, mode)
        patcher = mock.patch.object(sp, "open_target", return_value=target)
        patcher.start()
        self.addCleanup(patcher.stop)
        return target

    def test_success_changes_only_the_instances_keys_and_reports_names_not_values(self):
        original = b"KEEP=1\n# JWT_SECRET=commented\nJWT_SECRET=old\nOTHER=x y\n"
        path = self.local_target(original)
        self.assertEqual(self.run_push({"kind": "local", "path": path}), 0)
        with open(path, "rb") as f:
            self.assertEqual(f.read(), b"KEEP=1\n# JWT_SECRET=commented\nJWT_SECRET=" + self.secret.encode() +
                             b"\nOTHER=x y\nSUPABASE_URL=https://supa1.example.com\n")
        text = self.text()
        self.assertIn("SUCCESS  instance 1 (supa1.example.com)", text)
        self.assertIn("Replaced (1): JWT_SECRET", text)
        self.assertIn("Added (1): SUPABASE_URL", text)
        self.assertNotIn(self.secret, text)
        self.assertFalse(any(os.path.exists(d) for d in self.tmps), "temporary copy must be deleted")

    def test_dry_run_writes_nothing(self):
        path = self.local_target(b"KEEP=1\n")
        self.assertEqual(self.run_push({"kind": "local", "path": path}, dry_run=True), 0)
        with open(path, "rb") as f:
            self.assertEqual(f.read(), b"KEEP=1\n")
        self.assertIn("DRY RUN", self.text())
        self.assertIn("Would add (2)", self.text())

    def test_nothing_to_change_does_not_touch_the_file(self):
        path = self.local_target(b"SUPABASE_URL=https://supa1.example.com\nJWT_SECRET=" + self.secret.encode() + b"\n")
        before = os.stat(path).st_mtime_ns
        self.assertEqual(self.run_push({"kind": "local", "path": path}), 0)
        self.assertEqual(os.stat(path).st_mtime_ns, before)
        self.assertIn("Nothing to change", self.text())

    def test_missing_target_file_fails_and_is_not_created(self):
        path = f"{self.dir}/nope.env"
        self.assertEqual(self.run_push({"kind": "local", "path": path}), 1)
        self.assertFalse(os.path.exists(path))
        self.assertIn("FAILED", self.text())
        self.assertIn("does not exist", self.text())
        self.assertIn("Your file was not changed.", self.text())

    def test_instance_without_env_fails(self):
        os.remove(f"{self.dir}/1/.env")
        self.assertEqual(self.run_push({"kind": "local", "path": "/x"}), 1)
        self.assertIn("has no .env file", self.text())

    def test_refused_write_leaves_the_file_alone(self):
        target = self.fake(b"KEEP=1\n", "refuse")
        self.assertEqual(self.run_push({"kind": "ssh"}), 1)
        self.assertEqual(target.data, b"KEEP=1\n")
        self.assertIn("Your file was not changed.", self.text())
        self.assertTrue(target.closed)
        self.assertFalse(any(os.path.exists(d) for d in self.tmps))

    def test_a_write_that_loses_a_line_is_undone(self):
        original = b"KEEP=1\nA=2\n"
        target = self.fake(original, "drop-a-line")
        self.assertEqual(self.run_push({"kind": "ssh"}), 1)
        self.assertEqual(target.data, original)
        self.assertIn("the check after writing failed", self.text())
        self.assertIn("restored to exactly how it was", self.text())
        self.assertFalse(any(os.path.exists(d) for d in self.tmps))

    def test_if_the_file_cannot_be_put_back_the_copy_is_kept(self):
        original = b"KEEP=1\nA=2\n"
        target = self.fake(original, "die-half-way")
        self.assertEqual(self.run_push({"kind": "ssh"}), 1)
        text = self.text()
        self.assertIn("WARNING: the file may be damaged", text)
        self.assertIn("connection dropped while writing", text)
        kept = [d for d in self.tmps if os.path.exists(d)]
        self.assertEqual(len(kept), 1)
        self.assertIn(kept[0] + "/original.env", text)
        with open(kept[0] + "/original.env", "rb") as f:
            self.assertEqual(f.read(), original)
        self.assertNotEqual(target.data, original)

    def test_advice_is_printed_with_the_failure(self):
        with mock.patch.object(sp, "open_target", side_effect=lambda *a: FakeTarget(b"\xff\x00")):
            self.assertEqual(self.run_push({"kind": "ssh"}), 1)
        self.assertIn("save it again as UTF-8", self.text())

    def test_local_write_keeps_mode_and_file_identity(self):
        path = self.local_target(b"KEEP=1\nJWT_SECRET=a long old value that is longer than the new one\n")
        os.chmod(path, 0o640)
        inode = os.stat(path).st_ino
        self.assertEqual(self.run_push({"kind": "local", "path": path}), 0)
        st = os.stat(path)
        self.assertEqual((st.st_ino, st.st_mode & 0o777), (inode, 0o640))


class SshTargetCommands(unittest.TestCase):
    SPEC = {"kind": "ssh", "user": "igor", "host": "pc.lan", "port": 2200, "path": "/C:/app/.env", "key": "/k/id"}

    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.addCleanup(shutil.rmtree, self.tmp, True)
        self.calls = []

    def fake_run(self, content=b"A=1\n", master_fails="", scp_fails=""):
        def run(cmd, **kw):
            self.calls.append(cmd)
            result = (0, "")
            if "scp" in cmd:
                if scp_fails:
                    result = (1, scp_fails)
                elif "download" in cmd[-1]:
                    with open(cmd[-1], "wb") as f:
                        f.write(content)
            elif "-M" in cmd and master_fails:
                kw["stderr"].write(master_fails)  # the master's messages go to a file
                result = (255, "")
            return subprocess_result(*result)
        return mock.patch.object(sp.subprocess, "run", run)

    def scp_call(self):
        return next(c for c in self.calls if "scp" in c)

    def test_one_shared_connection_is_opened_first_and_used_by_every_copy(self):
        target = sp.SshTarget(self.SPEC, self.tmp, None)
        with self.fake_run(b"A=1\n"):
            self.assertEqual(target.read(), b"A=1\n")
            target.write(b"A=2\n")
            target.read()
        masters = [c for c in self.calls if "-M" in c]
        self.assertEqual(len(masters), 1)
        self.assertEqual(masters[0][:4], ["ssh", "-f", "-N", "-M"])
        self.assertEqual(masters[0][masters[0].index("-S") + 1], f"{self.tmp}/c")
        self.assertEqual(masters[0][-1], "igor@pc.lan")
        self.assertIs(self.calls[0], masters[0])
        for scp in (c for c in self.calls if "scp" in c):
            self.assertIn(f"ControlPath={self.tmp}/c", scp)

    def test_read_copies_the_file_with_scp(self):
        target = sp.SshTarget(self.SPEC, self.tmp, None)
        with self.fake_run(b"A=1\n"):
            target.read()
        for cmd in self.calls:
            self.assertIn("Port=2200", cmd)
            self.assertEqual(cmd[cmd.index("-i") + 1], "/k/id")
            self.assertIn("IdentitiesOnly=yes", cmd)
        self.assertEqual(self.scp_call()[-2], "igor@pc.lan:/C:/app/.env")

    def test_write_uploads_and_removes_its_temporary_file(self):
        target = sp.SshTarget(self.SPEC, self.tmp, None)
        with self.fake_run():
            target.write(b"A=2\n")
        cmd = self.scp_call()
        self.assertEqual(cmd[-1], "igor@pc.lan:/C:/app/.env")
        self.assertFalse(os.path.exists(cmd[-2]))

    def test_ssh_runs_as_the_person_who_typed_sudo(self):
        owner = collections.namedtuple("Owner", "pw_name pw_uid pw_gid")("igor", os.getuid(), os.getgid())
        target = sp.SshTarget(self.SPEC, self.tmp, owner)
        with self.fake_run():
            target.read()
        self.assertEqual(len(self.calls), 2)
        for cmd in self.calls:
            self.assertEqual(cmd[:4], ["runuser", "-u", "igor", "--"])

    def test_without_a_terminal_it_never_waits_for_a_password(self):
        with mock.patch("sys.stdin") as stdin:
            stdin.isatty.return_value = False
            self.assertIn("BatchMode=yes", sp.SshTarget(self.SPEC, self.tmp, None).opts)
            stdin.isatty.return_value = True
            self.assertNotIn("BatchMode=yes", sp.SshTarget(self.SPEC, self.tmp, None).opts)

    def test_a_failed_login_becomes_advice(self):
        target = sp.SshTarget(self.SPEC, self.tmp, None)
        with self.fake_run(master_fails="ssh: connect to host pc.lan port 2200: Connection refused"):
            with self.assertRaises(sp.TargetError) as cm:
                target.read()
        self.assertIn("nothing is listening for SSH on port 2200", cm.exception.why)
        self.assertFalse(any("scp" in c for c in self.calls), "no copy is tried without a connection")

    def test_a_failed_copy_becomes_advice(self):
        target = sp.SshTarget(self.SPEC, self.tmp, None)
        with self.fake_run(scp_fails="scp: /C:/app/.env: No such file or directory"):
            with self.assertRaises(sp.TargetError) as cm:
                target.read()
        self.assertIn("does not exist on pc.lan", cm.exception.why)

    def test_close_asks_the_shared_connection_to_quit(self):
        target = sp.SshTarget(self.SPEC, self.tmp, None)
        open(target.sock, "w").close()
        with self.fake_run():
            target.close()
        self.assertEqual(self.calls[0][-3:], ["-O", "exit", "igor@pc.lan"])
        self.assertIn(f"ControlPath={target.sock}", self.calls[0])


def subprocess_result(returncode, stderr):
    return collections.namedtuple("R", "returncode stdout stderr")(returncode, "", stderr)


class SshErrors(unittest.TestCase):
    SPEC = {"user": "igor", "host": "pc.lan", "port": 22, "path": "/app/.env", "key": None}

    def why(self, err, **spec):
        e = sp.ssh_error(err, dict(self.SPEC, **spec), "igor")
        return e.why, "\n".join(e.advice)

    def test_unknown_name(self):
        why, advice = self.why("ssh: Could not resolve hostname pc.lan: Name or service not known")
        self.assertIn("'pc.lan' is not known", why)
        self.assertIn("IP address", advice)

    def test_single_letter_host_hints_at_windows_paths(self):
        _, advice = self.why("ssh: Could not resolve hostname C: Name or service not known", host="C")
        self.assertIn("user@computer:C:/Users/", advice)

    def test_timeout(self):
        why, advice = self.why("ssh: connect to host pc.lan port 22: Connection timed out")
        self.assertIn("did not answer on port 22", why)
        self.assertIn("switched on", advice)
        self.assertIn("ssh -p 22 igor@pc.lan", advice)

    def test_refused_explains_how_to_switch_ssh_on_everywhere(self):
        why, advice = self.why("ssh: connect to host pc.lan port 22: Connection refused")
        self.assertIn("nothing is listening", why)
        for word in ("Remote Login", "openssh-server", "OpenSSH Server"):
            self.assertIn(word, advice)

    def test_login_refused(self):
        why, advice = self.why("igor@pc.lan: Permission denied (publickey,password).")
        self.assertIn("refused the login for the user 'igor'", why)
        self.assertIn("ssh-copy-id -p 22 igor@pc.lan", advice)
        self.assertIn("administrators_authorized_keys", advice)

    def test_unknown_computer_and_changed_computer(self):
        why, advice = self.why("Host key verification failed.")
        self.assertIn("does not trust pc.lan yet", why)
        self.assertIn("answer yes", advice)
        why, advice = self.why("@@@ WARNING: REMOTE HOST IDENTIFICATION HAS CHANGED! @@@\nHost key verification failed.",
                               port=2222)
        self.assertIn("identifies itself differently", why)
        self.assertIn('ssh-keygen -R [pc.lan]:2222', advice)

    def test_missing_file_and_protected_file(self):
        why, _ = self.why("scp: /app/.env: No such file or directory")
        self.assertIn("/app/.env does not exist on pc.lan", why)
        why, advice = self.why('scp: dest open "/app/.env": Permission denied')
        self.assertIn("not allowed to open or change /app/.env", why)
        self.assertIn("write permission", advice)

    def test_sftp_switched_off(self):
        why, advice = self.why("subsystem request failed on channel 0\nscp: Connection closed")
        self.assertIn("does not allow file copying (SFTP)", why)
        self.assertIn("Subsystem sftp", advice)

    def test_bad_key_file(self):
        why, _ = self.why("Warning: Identity file /k/id not accessible: No such file or directory.\n"
                          "igor@pc.lan: Permission denied (publickey).", key="/k/id")
        self.assertIn("SSH key file cannot be used", why)

    def test_anything_else_shows_the_message_and_how_to_try_by_hand(self):
        why, advice = self.why("kex_exchange_identification: read: Connection reset by peer")
        self.assertIn("Connection reset by peer", why)
        self.assertIn("ssh -p 22 igor@pc.lan", advice)


class PushEnvCommand(unittest.TestCase):
    def run_cmd(self, args, numbers=(1, 2), tty=False):
        calls = []
        with mock.patch.object(sp, "existing_numbers", return_value=set(numbers)), \
                mock.patch.object(sp, "invoking_user", return_value=None), \
                mock.patch.object(sp, "push_env", lambda *a: calls.append(a) or 0), \
                mock.patch("sys.stdin") as stdin, \
                contextlib.redirect_stderr(io.StringIO()) as err:
            stdin.isatty.return_value = tty
            code = sp.push_env_command(CFG, args)
        return code, calls, err.getvalue()

    def test_remote_target_with_options(self):
        code, calls, _ = self.run_cmd(["1", "igor@h:/p/.env", "--port", "2222", "--key", "/k", "--dry-run"])
        self.assertEqual(code, 0)
        _, n, spec, _, dry = calls[0]
        self.assertEqual((n, dry), (1, True))
        self.assertEqual(spec, {"kind": "ssh", "user": "igor", "host": "h", "path": "/p/.env", "port": 2222, "key": "/k"})

    def test_local_target(self):
        _, calls, _ = self.run_cmd(["2", "/srv/app/.env"])
        self.assertEqual(calls[0][2], {"kind": "local", "path": "/srv/app/.env"})

    def test_unknown_instance(self):
        code, calls, err = self.run_cmd(["3", "/x"])
        self.assertEqual((code, calls), (1, []))
        self.assertIn("Instance 3 does not exist", err)

    def test_missing_target_without_a_terminal(self):
        code, calls, err = self.run_cmd(["1"])
        self.assertEqual((code, calls), (1, []))
        self.assertIn("Say where the file is", err)

    def test_windows_path_without_computer(self):
        code, calls, err = self.run_cmd(["1", "C:\\Users\\x\\.env"])
        self.assertEqual((code, calls), (1, []))
        self.assertIn("Windows path without the computer's name", err)


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


class Upgrade(unittest.TestCase):
    """`supaspin upgrade` puts the current memory caps on an instance that is already running.
    It must change the caps and nothing else: no volume is resized, no secret is regenerated."""

    STORED = {
        "deployment": {
            "analytics": {"enabled": True, "priorityClassName": "supaspin-low",
                          "resources": {"limits": {"memory": "768Mi"},
                                        "requests": {"memory": "128Mi"}}},
            "vector": {"enabled": True, "priorityClassName": "supaspin-low",
                       "resources": {"limits": {"memory": "512Mi"},
                                     "requests": {"memory": "128Mi"}}},
        },
        "secret": {"jwt": {"secret": "keep-me", "anonKey": "anon"}},
        "persistence": {"db": {"enabled": True, "size": "2048Mi"}},
    }

    def setUp(self):
        self.calls = []
        self.written = {}
        self.log = []

    @contextlib.contextmanager
    def fake_helm(self, stored=None, rc=0):
        stored = self.STORED if stored is None else stored

        def helm(*args, check=True):
            self.calls.append(list(args))
            if args[0] == "get":
                out = yaml.safe_dump(stored) if stored is not None else ""
                return collections.namedtuple("P", "returncode stdout stderr")(rc, out, "boom")
            if args[0] == "upgrade":
                with open(args[args.index("-f") + 1]) as f:
                    self.written = yaml.safe_load(f)
            return collections.namedtuple("P", "returncode stdout stderr")(0, "", "")

        with mock.patch.object(sp, "helm", helm), \
                mock.patch.object(sp, "wait_ready", lambda n, log, **kw: None), \
                mock.patch.object(sp, "setup_log_retention", lambda cfg, n, log: None), \
                mock.patch.object(sp, "STATE_DIR", tempfile.mkdtemp()):
            yield

    def test_raises_the_caps_and_keeps_everything_else(self):
        with self.fake_helm():
            changed = sp.upgrade_instance(CFG, 1, self.log.append)
        self.assertTrue(changed)
        dep = self.written["deployment"]
        self.assertEqual(dep["analytics"]["resources"]["limits"]["memory"],
                         f"{sp.MEMORY['analytics'][1]}Mi")
        self.assertEqual(dep["vector"]["resources"]["limits"]["memory"],
                         f"{sp.MEMORY['vector'][1]}Mi")
        self.assertTrue(dep["analytics"]["enabled"])              # untouched
        self.assertEqual(self.written["secret"]["jwt"]["secret"], "keep-me")
        self.assertEqual(self.written["persistence"]["db"]["size"], "2048Mi")

    def test_every_service_gets_a_cap_even_if_the_stored_values_had_none(self):
        with self.fake_helm():
            sp.upgrade_instance(CFG, 1, self.log.append)
        for svc in sp.SERVICES:
            req, lim = sp.MEMORY.get(svc, sp.DEFAULT_MEMORY)
            got = self.written["deployment"][svc]["resources"]
            self.assertEqual(got["limits"]["memory"], f"{lim}Mi")
            self.assertEqual(got["requests"]["memory"], f"{req}Mi")

    def test_the_values_file_with_the_secrets_is_deleted_again(self):
        with self.fake_helm():
            sp.upgrade_instance(CFG, 1, self.log.append)
        path = [c[c.index("-f") + 1] for c in self.calls if c[0] == "upgrade"][0]
        self.assertFalse(os.path.exists(path))

    def test_helm_gets_the_post_renderer(self):
        with self.fake_helm():
            sp.upgrade_instance(CFG, 2, self.log.append)
        up = [c for c in self.calls if c[0] == "upgrade"][0]
        self.assertIn("--post-renderer", up)
        self.assertEqual(up[up.index("--post-renderer-args") + 1], "post-render")
        self.assertIn("2", up)

    def test_an_instance_already_at_the_caps_is_still_re_applied(self):
        # The caps are not the only thing a new version can change: the Studio rollout, the host
        # ports and the gateway routes are decided while the manifests are rendered, and none of
        # them shows up in the stored values.
        stored = {"deployment": sp.apply_memory({})["deployment"]}
        with self.fake_helm(stored):
            changed = sp.upgrade_instance(CFG, 1, self.log.append)
        self.assertFalse(changed)                                   # no cap moved
        self.assertEqual([c[0] for c in self.calls], ["get", "upgrade"])  # applied anyway

    def test_a_missing_release_is_refused_before_anything_is_written(self):
        with self.fake_helm(rc=1):
            with self.assertRaises(RuntimeError):
                sp.upgrade_instance(CFG, 3, self.log.append)
        self.assertEqual([c[0] for c in self.calls], ["get"])

    def test_empty_stored_values_are_refused(self):
        with self.fake_helm({}):
            with self.assertRaises(RuntimeError):
                sp.upgrade_instance(CFG, 1, self.log.append)
        self.assertEqual([c[0] for c in self.calls], ["get"])

    def test_upgrade_all_does_every_instance_and_reports_the_failures(self):
        done = []

        def one(cfg, n, log):
            if n == 2:
                raise RuntimeError("no release")
            done.append(n)
            return True

        with mock.patch.object(sp, "upgrade_instance", one), \
                mock.patch.object(sp, "existing_numbers", lambda cfg: {1, 2, 3}):
            with self.assertRaises(RuntimeError) as e:
                sp.upgrade_all(CFG, self.log.append)
        self.assertEqual(done, [1, 3])
        self.assertIn("2", str(e.exception))

    def test_upgrade_all_with_a_number_does_only_that_instance(self):
        done = []
        with mock.patch.object(sp, "upgrade_instance", lambda cfg, n, log: done.append(n)), \
                mock.patch.object(sp, "existing_numbers", lambda cfg: {1, 2, 3}):
            sp.upgrade_all(CFG, self.log.append, 2)
        self.assertEqual(done, [2])


class HostPortRollout(unittest.TestCase):
    """Studio and Postgres bind a port on the server itself, and only one pod can hold a port.
    The Studio deployment must therefore stop its old pod before starting the new one, or an
    upgrade leaves the replacement pod Pending for ever."""

    def rendered(self, n=1):
        sec = sp.generate_secrets()
        values = sp.instance_values(CFG, n, sec, MAIL, template())
        with mock.patch.object(sp, "helm"), mock.patch.object(sp, "kubectl"):
            text = yaml.safe_dump_all([
                {"apiVersion": "apps/v1", "kind": "Deployment",
                 "metadata": {"name": f"{sp.fullname(n)}-studio"},
                 "spec": {"strategy": {"type": "RollingUpdate",
                                       "rollingUpdate": {"maxSurge": "25%"}},
                          "template": {"spec": {"containers": [
                              {"name": "studio", "ports": [{"containerPort": 3000}]}]}}}},
                {"apiVersion": "apps/v1", "kind": "StatefulSet",
                 "metadata": {"name": f"{sp.fullname(n)}-db"},
                 "spec": {"template": {"spec": {"containers": [
                     {"name": "db", "ports": [{"containerPort": 5432}]}]}}}},
                {"apiVersion": "apps/v1", "kind": "Deployment",
                 "metadata": {"name": f"{sp.fullname(n)}-rest"},
                 "spec": {"strategy": {"type": "RollingUpdate"},
                          "template": {"spec": {"containers": [
                              {"name": "rest", "ports": [{"containerPort": 3000}]}]}}}},
            ])
        out = list(yaml.safe_load_all(sp.post_render(text, n, CFG)))
        return {d["metadata"]["name"]: d for d in out}

    def test_studio_stops_the_old_pod_before_starting_the_new_one(self):
        d = self.rendered()[f"{sp.fullname(1)}-studio"]
        self.assertEqual(d["spec"]["strategy"], {"type": "Recreate"})
        port = d["spec"]["template"]["spec"]["containers"][0]["ports"][0]
        self.assertEqual(port["hostPort"], sp.studio_port(1))
        self.assertEqual(port["hostIP"], CFG["lan_ip"])

    def test_a_deployment_without_a_host_port_keeps_the_normal_rollout(self):
        d = self.rendered()[f"{sp.fullname(1)}-rest"]
        self.assertEqual(d["spec"]["strategy"]["type"], "RollingUpdate")

    def test_the_database_keeps_its_host_port_and_gains_no_strategy(self):
        d = self.rendered()[f"{sp.fullname(1)}-db"]
        self.assertNotIn("strategy", d["spec"])
        self.assertEqual(d["spec"]["template"]["spec"]["containers"][0]["ports"][0]["hostPort"],
                         sp.pg_port(1))


class SuspendResume(unittest.TestCase):
    """Suspending an instance stops its pods and keeps everything else: the disks, the passwords,
    the certificate and the hostname. Resuming starts it again with the same data."""

    RUNNING = {
        "deployment": {
            "analytics": {"enabled": True,
                          "resources": {"limits": {"memory": "1536Mi"},
                                        "requests": {"memory": "128Mi"}}},
            "db": {"resources": {"limits": {"memory": "2048Mi"},
                                 "requests": {"memory": "512Mi"}}},
        },
        "secret": {"jwt": {"secret": "keep-me"}, "db": {"password": "keep-me-too"}},
        "persistence": {"db": {"enabled": True, "size": "2048Mi"}},
    }

    def setUp(self):
        self.calls = []
        self.written = {}
        self.log = []
        self.waited = []

    @contextlib.contextmanager
    def fake_helm(self, stored):
        P = collections.namedtuple("P", "returncode stdout stderr")

        def helm(*args, check=True):
            self.calls.append(list(args))
            if args[0] == "get":
                return P(0, yaml.safe_dump(stored), "")
            if args[0] == "upgrade":
                with open(args[args.index("-f") + 1]) as f:
                    self.written = yaml.safe_load(f)
            return P(0, "", "")

        with mock.patch.object(sp, "helm", helm), \
                mock.patch.object(sp, "wait_ready",
                                  lambda n, log, **kw: self.waited.append(("ready", n))), \
                mock.patch.object(sp, "wait_gone",
                                  lambda n, log, **kw: self.waited.append(("gone", n))), \
                mock.patch.object(sp, "setup_log_retention",
                                  lambda cfg, n, log: self.waited.append(("retention", n))), \
                mock.patch.object(sp, "STATE_DIR", tempfile.mkdtemp()):
            yield

    def suspended(self):
        return sp.set_replicas(yaml.safe_load(yaml.safe_dump(self.RUNNING)), 0)

    # ---- the setting itself ----

    def test_set_replicas_covers_every_service(self):
        v = sp.set_replicas({}, 0)
        self.assertEqual(sorted(v["deployment"]), sorted(sp.SERVICES))
        self.assertTrue(all(d["replicaCount"] == 0 for d in v["deployment"].values()))
        self.assertTrue(sp.is_suspended(v))

    def test_a_service_missing_the_setting_counts_as_running(self):
        self.assertFalse(sp.is_suspended(self.RUNNING))
        self.assertFalse(sp.is_suspended({"deployment": {}}))

    def test_one_service_left_running_is_not_suspended(self):
        v = sp.set_replicas({}, 0)
        v["deployment"]["kong"]["replicaCount"] = 1
        self.assertFalse(sp.is_suspended(v))

    # ---- suspend ----

    def test_suspend_stops_every_service_and_keeps_the_rest(self):
        with self.fake_helm(self.RUNNING):
            self.assertTrue(sp.suspend_instance(CFG, 1, self.log.append))
        for svc in sp.SERVICES:
            self.assertEqual(self.written["deployment"][svc]["replicaCount"], 0)
        self.assertEqual(self.written["secret"]["jwt"]["secret"], "keep-me")
        self.assertEqual(self.written["secret"]["db"]["password"], "keep-me-too")
        self.assertEqual(self.written["persistence"]["db"]["size"], "2048Mi")
        # the memory caps are kept as they were: they apply again on resume
        self.assertEqual(self.written["deployment"]["analytics"]["resources"]["limits"]["memory"],
                         "1536Mi")
        self.assertEqual(self.waited, [("gone", 1)])

    def test_suspend_waits_for_the_pods_to_be_gone_not_for_them_to_be_ready(self):
        with self.fake_helm(self.RUNNING):
            sp.suspend_instance(CFG, 2, self.log.append)
        self.assertEqual([w[0] for w in self.waited], ["gone"])

    def test_suspending_a_suspended_instance_changes_nothing(self):
        with self.fake_helm(self.suspended()):
            self.assertFalse(sp.suspend_instance(CFG, 1, self.log.append))
        self.assertEqual([c[0] for c in self.calls], ["get"])
        self.assertEqual(self.waited, [])

    def test_the_values_file_with_the_secrets_is_deleted_again(self):
        with self.fake_helm(self.RUNNING):
            sp.suspend_instance(CFG, 1, self.log.append)
        path = [c[c.index("-f") + 1] for c in self.calls if c[0] == "upgrade"][0]
        self.assertFalse(os.path.exists(path))

    # ---- resume ----

    def test_resume_starts_every_service_again(self):
        with self.fake_helm(self.suspended()):
            self.assertTrue(sp.resume_instance(CFG, 3, self.log.append))
        for svc in sp.SERVICES:
            self.assertEqual(self.written["deployment"][svc]["replicaCount"], 1)
        self.assertFalse(sp.is_suspended(self.written))
        self.assertEqual(self.waited, [("ready", 3), ("retention", 3)])

    def test_resuming_a_running_instance_changes_nothing(self):
        with self.fake_helm(self.RUNNING):
            self.assertFalse(sp.resume_instance(CFG, 1, self.log.append))
        self.assertEqual([c[0] for c in self.calls], ["get"])
        self.assertEqual(self.waited, [])

    def test_a_suspended_instance_keeps_its_data_settings(self):
        with self.fake_helm(self.suspended()):
            sp.resume_instance(CFG, 1, self.log.append)
        self.assertEqual(self.written["persistence"]["db"]["size"], "2048Mi")
        self.assertEqual(self.written["secret"]["jwt"]["secret"], "keep-me")

    # ---- upgrade must not trip over a suspended instance ----

    def test_upgrade_leaves_a_suspended_instance_suspended_and_does_not_wait(self):
        with self.fake_helm(self.suspended()):
            sp.upgrade_instance(CFG, 1, self.log.append)
        self.assertTrue(sp.is_suspended(self.written))        # not woken up
        self.assertEqual(self.waited, [])                     # would never finish
        self.assertEqual(self.written["deployment"]["analytics"]["resources"]["limits"]["memory"],
                         f"{sp.MEMORY['analytics'][1]}Mi")    # new caps stored for the resume

    def test_upgrade_still_waits_for_a_running_instance(self):
        with self.fake_helm(self.RUNNING):
            sp.upgrade_instance(CFG, 1, self.log.append)
        self.assertEqual(self.waited, [("ready", 1), ("retention", 1)])


class InstanceState(unittest.TestCase):
    """What `supaspin list` says about an instance, read from what the cluster is asked to run."""

    def state(self, deployments, statefulsets):
        def kjson(*args):
            kind = args[1]
            counts = deployments if kind == "deployments" else statefulsets
            return {"items": [{"spec": {"replicas": c}} for c in counts]}

        with mock.patch.object(sp, "kjson", kjson):
            return sp.instance_state(1)

    def test_running(self):
        self.assertEqual(self.state([1] * 11, [1]), "running")

    def test_suspended(self):
        self.assertEqual(self.state([0] * 11, [0]), "suspended")

    def test_the_database_alone_still_running_is_reported_honestly(self):
        self.assertEqual(self.state([0] * 11, [1]), "partly suspended")

    def test_nothing_there(self):
        self.assertEqual(self.state([], []), "unknown")

    def test_the_list_shows_the_state_and_no_tunnel_for_a_sleeping_instance(self):
        rows = [{"n": 1, "host": "supa1.example.com", "pods": "0/0", "state": "suspended",
                 "memory": "n/a", "disk": "db 1/2M", "env": "/data/supabase/1/.env"}]
        out = io.StringIO()
        with mock.patch.object(sp, "list_rows", lambda cfg: rows), \
                contextlib.redirect_stdout(out):
            sp.print_list(CFG)
        text = out.getvalue()
        self.assertIn("suspended", text)
        self.assertIn("Resume it", text)          # the menu line, not a command to remember
        self.assertNotIn("ssh -N -L", text)


class InstanceMenu(unittest.TestCase):
    """Stopping and starting an instance lives inside the instance you are looking at: one line,
    worded for the state the instance is in, so there is no second command to remember."""

    def run_menu(self, state, choose):
        """Draw the menu for an instance in `state`, pick the item whose text `choose` matches
        (None: leave at once). Returns (the item texts, what was called)."""
        seen, called = [], []

        def menu(title, items):
            seen.append((title, list(items)))
            if choose is None:
                return None
            for i, text in enumerate(items):
                if choose in text:
                    return i
            raise AssertionError(f"no item matching {choose!r} in {items}")

        calls = {"suspend": lambda cfg, n, log: called.append(("suspend", n)),
                 "resume": lambda cfg, n, log: called.append(("resume", n))}
        out = io.StringIO()
        with mock.patch.object(sp, "menu", menu), \
                mock.patch.object(sp, "instance_state", lambda n: state), \
                mock.patch.object(sp, "pause", lambda: (_ for _ in ()).throw(StopIteration)), \
                mock.patch.object(sp, "suspend_instance", calls["suspend"]), \
                mock.patch.object(sp, "resume_instance", calls["resume"]), \
                mock.patch.object(sp, "read_env", lambda cfg, n: "A=b\n"), \
                contextlib.redirect_stdout(out):
            try:
                sp.show_menu(CFG, 2, lambda *a, **k: None)
            except StopIteration:
                pass  # the menu would loop for ever; stop at the first "press Enter"
        return seen[0][1], called, out.getvalue(), seen[0][0]

    def test_a_running_instance_offers_to_suspend_it(self):
        items, called, _, title = self.run_menu("running", None)
        self.assertTrue(any("Suspend it" in t for t in items), items)
        self.assertFalse(any("Resume" in t for t in items), items)
        self.assertIn("running", title)

    def test_a_suspended_instance_offers_to_resume_it(self):
        items, called, _, title = self.run_menu("suspended", None)
        self.assertTrue(any("Resume it" in t for t in items), items)
        self.assertFalse(any("Suspend" in t for t in items), items)
        self.assertIn("suspended", title)

    def test_choosing_it_suspends_this_instance(self):
        _, called, _, _ = self.run_menu("running", "Suspend it")
        self.assertEqual(called, [("suspend", 2)])

    def test_choosing_it_resumes_this_instance(self):
        _, called, _, _ = self.run_menu("suspended", "Resume it")
        self.assertEqual(called, [("resume", 2)])

    def test_the_tunnel_of_a_sleeping_instance_says_why_it_is_pointless(self):
        _, _, text, _ = self.run_menu("suspended", "Tunnel")
        self.assertIn("suspended", text)
        self.assertNotIn("ssh -N -L", text)

    def test_the_picker_names_the_state_of_every_instance(self):
        seen = []

        def menu(title, items):
            seen.append(list(items))
            return None

        with mock.patch.object(sp, "menu", menu), \
                mock.patch.object(sp, "existing_numbers", lambda cfg: {1, 2}), \
                mock.patch.object(sp, "instance_state", lambda n: "suspended" if n == 2 else "running"):
            sp.pick_instance(CFG, "Which instance?")
        self.assertEqual(seen[0], ["1  supa1.example.com  (running)",
                                   "2  supa2.example.com  (suspended)"])