bobandClaude Fable 5.1
b916165cab
Reuse the flickr key flickrtouchr already holds and the immich url and key immich-sync uses, so nothing secret is duplicated. Everything else comes from a typed settings file with flag > environment > file > default precedence, so a cron line can be the bare script. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
239 lines
9.4 KiB
Python
239 lines
9.4 KiB
Python
import contextlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.parse
|
|
|
|
_here = os.path.dirname(os.path.abspath(__file__))
|
|
_path = os.path.join(_here, "rgl-photo-backup")
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"rgl_photo_backup", _path,
|
|
loader=importlib.machinery.SourceFileLoader("rgl_photo_backup", _path))
|
|
rb = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(rb)
|
|
|
|
|
|
class FakeResponse:
|
|
"""What urlopen would return: a body, a status and headers."""
|
|
|
|
def __init__(self, body=b"", status=200, headers=None):
|
|
self.body = body if isinstance(body, bytes) else json.dumps(body).encode()
|
|
self.status = status
|
|
self.headers = headers or {}
|
|
self._pos = 0
|
|
|
|
def read(self, size=-1):
|
|
if size is None or size < 0:
|
|
data, self._pos = self.body[self._pos:], len(self.body)
|
|
else:
|
|
data = self.body[self._pos:self._pos + size]
|
|
self._pos += len(data)
|
|
return data
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
pass
|
|
|
|
|
|
def http_error(code, body=b"", headers=None):
|
|
return urllib.error.HTTPError("http://test", code, "reason %d" % code,
|
|
headers or {}, io.BytesIO(body))
|
|
|
|
|
|
class FakeOpen:
|
|
"""Replaces rb.OPEN. Serves scripted responses in order and records
|
|
every request so a test can assert on method, url, headers and body."""
|
|
|
|
def __init__(self, *responses):
|
|
self.responses = list(responses)
|
|
self.requests = []
|
|
|
|
def __call__(self, request, timeout=None):
|
|
self.requests.append(request)
|
|
if not self.responses:
|
|
raise AssertionError("unexpected request: %s %s"
|
|
% (request.get_method(), request.full_url))
|
|
answer = self.responses.pop(0)
|
|
if isinstance(answer, Exception):
|
|
raise answer
|
|
return answer
|
|
|
|
|
|
class TestCase(unittest.TestCase):
|
|
"""Silences output and restores the module globals tests move around."""
|
|
|
|
GLOBALS = ("OPEN", "SLEEP", "ATTEMPTS", "BACKOFF_BASE", "PACE", "QUIET",
|
|
"DEBUG", "HTTP_TIMEOUT", "USER_AGENT", "STATS", "DOWNLOAD_CAP")
|
|
|
|
def setUp(self):
|
|
self._output = contextlib.ExitStack()
|
|
if not os.environ.get("RGL_PHOTO_BACKUP_TEST_VERBOSE"):
|
|
self.stdout = self._output.enter_context(
|
|
contextlib.redirect_stdout(io.StringIO()))
|
|
self.stderr = self._output.enter_context(
|
|
contextlib.redirect_stderr(io.StringIO()))
|
|
self.addCleanup(self._output.close)
|
|
self._saved = {k: getattr(rb, k) for k in self.GLOBALS}
|
|
rb.SLEEP = lambda seconds: None
|
|
rb.PACE = 0.0
|
|
rb.STATS = dict.fromkeys(rb.STATS, 0)
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.dir = self.tmp.name
|
|
|
|
def tearDown(self):
|
|
for k, v in self._saved.items():
|
|
setattr(rb, k, v)
|
|
|
|
def write(self, name, content):
|
|
path = os.path.join(self.dir, name)
|
|
mode = "wb" if isinstance(content, bytes) else "w"
|
|
with open(path, mode) as fh:
|
|
fh.write(content)
|
|
return path
|
|
|
|
|
|
class Output(TestCase):
|
|
def test_say_prints_unless_quiet(self):
|
|
rb.QUIET = False
|
|
rb.say("hello")
|
|
self.assertEqual(self.stdout.getvalue(), "hello\n")
|
|
rb.QUIET = True
|
|
rb.say("again")
|
|
self.assertEqual(self.stdout.getvalue(), "hello\n")
|
|
|
|
def test_warn_always_reaches_stderr(self):
|
|
rb.QUIET = True
|
|
rb.warn("careful")
|
|
self.assertEqual(self.stderr.getvalue(), "careful\n")
|
|
|
|
def test_debug_only_when_enabled(self):
|
|
rb.DEBUG = False
|
|
rb.debug("x")
|
|
self.assertEqual(self.stderr.getvalue(), "")
|
|
rb.DEBUG = True
|
|
rb.debug("y")
|
|
self.assertEqual(self.stderr.getvalue(), "debug: y\n")
|
|
|
|
def test_errors_share_a_base(self):
|
|
self.assertTrue(issubclass(rb.PhotoGone, rb.BackupError))
|
|
e = rb.HTTPFailure("ctx: HTTP 404", code=404)
|
|
self.assertEqual(e.code, 404)
|
|
self.assertIsInstance(e, rb.BackupError)
|
|
|
|
|
|
class Config(TestCase):
|
|
def test_readyaml_reads_flat_pairs_and_ignores_comments(self):
|
|
path = self.write("a.yml", "# note\nkey: 'abc'\nurl: https://immich.example/api \n")
|
|
self.assertEqual(rb.readyaml(path), {"key": "abc", "url": "https://immich.example/api"})
|
|
|
|
def test_readyaml_missing_file_is_empty(self):
|
|
self.assertEqual(rb.readyaml(os.path.join(self.dir, "none.yml")), {})
|
|
|
|
def test_readyaml_unreadable_path_names_it(self):
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
rb.readyaml(self.dir) # a directory, not a file
|
|
self.assertIn(self.dir, str(cm.exception))
|
|
|
|
def test_loadflickr_needs_key(self):
|
|
path = self.write("f.yml", "secret: s\n")
|
|
with self.assertRaises(rb.BackupError):
|
|
rb.loadflickr(path)
|
|
path = self.write("f2.yml", "key: k\nsecret: s\n")
|
|
self.assertEqual(rb.loadflickr(path), "k")
|
|
|
|
def test_loadimmich_needs_url_and_key_and_strips_slash(self):
|
|
path = self.write("i.yml", "url: https://immich.example/api/\nkey: abc\n")
|
|
self.assertEqual(rb.loadimmich(path), ("https://immich.example/api", "abc"))
|
|
with self.assertRaises(rb.BackupError):
|
|
rb.loadimmich(self.write("i2.yml", "url: x\n"))
|
|
|
|
|
|
class Settings(TestCase):
|
|
def test_flag_beats_env_beats_file_beats_default(self):
|
|
path = self.write("s.yml", "directory: /from/file\npace: 3\nquiet: true\n")
|
|
args = rb.parseargs(["--settings", path])
|
|
conf = rb.settings(args, environ={})
|
|
self.assertEqual(conf["directory"], "/from/file")
|
|
self.assertEqual(conf["pace"], 3.0)
|
|
self.assertTrue(conf["quiet"])
|
|
conf = rb.settings(args, environ={rb.ENV + "PACE": "7"})
|
|
self.assertEqual(conf["pace"], 7.0)
|
|
args = rb.parseargs(["--settings", path, "--pace", "9", "/cli"])
|
|
conf = rb.settings(args, environ={rb.ENV + "PACE": "7"})
|
|
self.assertEqual(conf["pace"], 9.0)
|
|
self.assertEqual(conf["directory"], "/cli")
|
|
|
|
def test_defaults(self):
|
|
args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "/d"])
|
|
conf = rb.settings(args, environ={})
|
|
self.assertEqual(conf["index_url"], rb.INDEX_URL)
|
|
self.assertTrue(conf["fetch"] and conf["sync"])
|
|
self.assertFalse(conf["dry_run"])
|
|
self.assertIsNone(conf["limit"])
|
|
self.assertEqual(conf["attempts"], 6)
|
|
self.assertEqual(conf["flickr_auth"], os.path.expanduser(rb.FLICKR_CONFIG))
|
|
|
|
def test_defaults_are_literals_not_moving_constants(self):
|
|
rb.PACE, rb.ATTEMPTS, rb.USER_AGENT = 99.0, 99, "mangled"
|
|
args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "/d"])
|
|
conf = rb.settings(args, environ={})
|
|
self.assertEqual(conf["pace"], rb.DEFAULTS["pace"])
|
|
self.assertEqual(conf["attempts"], rb.DEFAULTS["attempts"])
|
|
self.assertEqual(conf["user_agent"], rb.DEFAULTS["user_agent"])
|
|
self.assertEqual((rb.DEFAULTS["pace"], rb.DEFAULTS["attempts"]), (1.0, 6))
|
|
|
|
def test_debug_is_a_setting(self):
|
|
path = self.write("s.yml", "directory: /d\ndebug: true\n")
|
|
self.assertEqual(self.stderr.getvalue(), "") # no "unknown setting"
|
|
args = rb.parseargs(["--settings", path])
|
|
self.assertTrue(rb.settings(args, environ={})["debug"])
|
|
args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "/d"])
|
|
self.assertFalse(rb.settings(args, environ={})["debug"])
|
|
self.assertTrue(rb.settings(args, environ={rb.ENV + "DEBUG": "yes"})["debug"])
|
|
args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "--debug", "/d"])
|
|
self.assertTrue(rb.settings(args, environ={})["debug"])
|
|
|
|
def test_a_bad_environment_value_is_a_configuration_error(self):
|
|
args = rb.parseargs(["--settings", os.path.join(self.dir, "none"), "/d"])
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
rb.settings(args, environ={rb.ENV + "ATTEMPTS": "lots"})
|
|
self.assertIn("ATTEMPTS", str(cm.exception))
|
|
|
|
def test_phase_flags(self):
|
|
base = ["--settings", os.path.join(self.dir, "none"), "/d"]
|
|
self.assertFalse(rb.settings(rb.parseargs(base + ["--sync-only"]), environ={})["fetch"])
|
|
self.assertFalse(rb.settings(rb.parseargs(base + ["--fetch-only"]), environ={})["sync"])
|
|
with self.assertRaises(rb.BackupError):
|
|
rb.settings(rb.parseargs(base + ["--fetch-only", "--sync-only"]), environ={})
|
|
|
|
def test_missing_directory_is_an_error(self):
|
|
args = rb.parseargs(["--settings", os.path.join(self.dir, "none")])
|
|
with self.assertRaises(rb.BackupError):
|
|
rb.settings(args, environ={})
|
|
|
|
def test_unknown_key_warns_and_wrong_type_fails(self):
|
|
path = self.write("s.yml", "directory: /d\nbogus: 1\n")
|
|
rb.loadsettings(path)
|
|
self.assertIn("bogus", self.stderr.getvalue())
|
|
path = self.write("s2.yml", "directory: /d\nlimit: many\n")
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
rb.loadsettings(path)
|
|
self.assertIn("limit", str(cm.exception))
|
|
|
|
def test_tilde_expanded_in_paths(self):
|
|
path = self.write("s.yml", "directory: ~/photos\n")
|
|
self.assertEqual(rb.loadsettings(path)["directory"], os.path.expanduser("~/photos"))
|