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>
This commit was merged in pull request #2.
This commit is contained in:
Igor K.
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
cfbc4f1612
commit
20704e128f
@@ -4,9 +4,18 @@ 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
|
||||
|
||||
@@ -323,5 +332,599 @@ class Manifests(unittest.TestCase):
|
||||
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
Block a user