public/supaspin
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity
Files
45b02a85af518fb0e962e5b451e34ea9fdd6e188
supaspin/tests/test_supaspin.py
T
Igor K.andClaude Sonnet 5 20704e128f feat: tunnel guide in list, push-env, VERSION file 
- list and tunnel print plain-language tunnel instructions for every instance: keep it open, run it in the
  background, close it, with a notice that background/close work on Mac and Linux only, not Windows
- push-env N TARGET puts an instance's .env values into another .env file, on this server or over SSH (Mac, Linux,
  Windows): existing keys are replaced, missing keys added, commented-out lines ignored, every other line kept.
  A temporary copy of the target is compared after writing and put back on any mismatch. --dry-run shows the
  changes without writing. SSH failures are explained with how to set SSH up.
- VERSION file (1.1.0), version line in the README, `--version` prints "supaspin 1.1.0"; a test keeps the file, the
  code constant and the README in step
- remove an unused parameter of run() and an unreachable second help branch in main()

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-19 12:22:13 +03:00

931 lines
43 KiB
Python
Raw Blame History

"""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_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 2048, analytics 768, kong 512, realtime 384, the other eight 256 each
self.assertEqual(lim, 2048 + 768 + 512 + 384 + 8 * 256)
self.assertEqual(lim / 1024, 5.625)
self.assertEqual(req, 512 + 11 * 128) # 1.875 GiB
# analytics and realtime were killed for running out of memory at half of production
self.assertGreaterEqual(sp.MEMORY["analytics"][1], 768)
self.assertGreaterEqual(sp.MEMORY["realtime"][1], 384)
self.assertEqual(req / 1024, 1.875)
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"], "2048Mi")
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_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 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, "192.168.10.30")))
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_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, "10.0.0.1")))
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, "10.0.0.1"))
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"], "2048Mi")
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()
Reference in New Issue View Git Blame Copy Permalink