bobandClaude Fable 5.1
fbd77fc91b
Each run now writes .failures.json in the backup directory: the last reason every photo failed and the wiki pages that use it, dropping any photo that has since arrived. A new --report mode reads the wiki index and that file and lists every page whose photo is not on disk, with the reason or "not yet attempted", calling neither flickr nor immich. That is what tells a wiki editor which photos have gone from flickr and need replacing. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1606 lines
77 KiB
Python
1606 lines
77 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", "PACE_FLOOR",
|
|
"PACEHELD", "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 = rb.PACE_FLOOR = 0.0
|
|
rb.PACEHELD = 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
|
|
|
|
|
|
def index_entry(param, image, page="", copyright="", licence="", lat="51.5", long="-0.1", name=None):
|
|
md = {}
|
|
if image:
|
|
md["node_image"] = [image]
|
|
if page:
|
|
md["node_image_url"] = [page]
|
|
if copyright:
|
|
md["node_image_copyright"] = [copyright]
|
|
if licence:
|
|
md["node_image_licence"] = [licence]
|
|
return {"param": param, "name": name or param.replace("_", " "),
|
|
"wgs84_lat": lat, "wgs84_long": long, "has_geodata": 1,
|
|
"node_data": {"metadata": md}}
|
|
|
|
|
|
class Index(TestCase):
|
|
def test_readindex_tolerates_invalid_utf8(self):
|
|
entry = index_entry("Café,_N1", "https://farm1.static.flickr.com/1/123_abc_m.jpg")
|
|
raw = json.dumps([entry], ensure_ascii=False).encode("utf-8").replace("\xe9".encode(), b"\xc2")
|
|
nodes = rb.readindex(raw)
|
|
self.assertEqual(len(nodes), 1)
|
|
self.assertIn("�", nodes[0]["param"])
|
|
|
|
def test_readindex_flattens_fields(self):
|
|
raw = json.dumps([index_entry("Pub,_E1", "https://farm1.static.flickr.com/1/123_abc_m.jpg",
|
|
page="https://www.flickr.com/photos/x/123/",
|
|
copyright="Someone", licence="https://creativecommons.org/licenses/by-sa/2.0/"),
|
|
index_entry("Nopic,_E2", "")]).encode()
|
|
nodes = rb.readindex(raw)
|
|
self.assertEqual(len(nodes), 2)
|
|
n = nodes[0]
|
|
self.assertEqual(n["param"], "Pub,_E1")
|
|
self.assertEqual(n["name"], "Pub, E1")
|
|
self.assertEqual(n["image"], "https://farm1.static.flickr.com/1/123_abc_m.jpg")
|
|
self.assertEqual(n["page"], "https://www.flickr.com/photos/x/123/")
|
|
self.assertEqual(n["copyright"], "Someone")
|
|
self.assertEqual(n["licence"], "https://creativecommons.org/licenses/by-sa/2.0/")
|
|
self.assertEqual((n["lat"], n["long"]), ("51.5", "-0.1"))
|
|
self.assertEqual(n["url"], "https://london.randomness.org.uk/wiki.cgi?Pub%2C_E1")
|
|
self.assertEqual(nodes[1]["image"], "")
|
|
|
|
def test_zero_coordinates_are_kept(self):
|
|
raw = json.dumps([index_entry("Null,_Island", "https://x/1_a_m.jpg", lat=0, long=0)]).encode()
|
|
node = rb.readindex(raw)[0]
|
|
self.assertEqual((node["lat"], node["long"]), ("0", "0"))
|
|
|
|
def test_missing_coordinates_are_empty(self):
|
|
raw = json.dumps([index_entry("Nowhere", "https://x/1_a_m.jpg", lat=None, long=None)]).encode()
|
|
node = rb.readindex(raw)[0]
|
|
self.assertEqual((node["lat"], node["long"]), ("", ""))
|
|
|
|
def test_loadindex_fetches(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(json.dumps([index_entry("A", "")]).encode()))
|
|
self.assertEqual(len(rb.loadindex("http://t/index")), 1)
|
|
|
|
|
|
class PhotoKeys(TestCase):
|
|
def key(self, image, page=""):
|
|
return rb.photokey({"image": image, "page": page, "param": "X"})
|
|
|
|
def test_id_from_thumbnail_filename(self):
|
|
self.assertEqual(self.key("https://farm4.static.flickr.com/3188/3002966579_97596aea2e_m.jpg"),
|
|
("flickr", "3002966579"))
|
|
self.assertEqual(self.key("https://live.staticflickr.com/65535/52123_abc_m.jpg"),
|
|
("flickr", "52123"))
|
|
|
|
def test_id_from_page_when_thumbnail_is_odd(self):
|
|
self.assertEqual(self.key("https://www.flickr.com/some/other.jpg",
|
|
"https://www.flickr.com/photos/kake/3002966579/in/set-1/"),
|
|
("flickr", "3002966579"))
|
|
self.assertEqual(self.key("https://c1.staticflickr.com/x.jpg",
|
|
"https://secure.flickr.com/photos/12345678@N00/98765"),
|
|
("flickr", "98765"))
|
|
|
|
def test_short_link_is_base58(self):
|
|
self.assertEqual(rb.base58decode("5zmZkR"), 3002966579)
|
|
self.assertEqual(self.key("https://c1.staticflickr.com/x.jpg", "https://flic.kr/p/5zmZkR"),
|
|
("flickr", "3002966579"))
|
|
|
|
def test_non_flickr_hosts_key_on_url_hash(self):
|
|
source, key = self.key("https://www.geograph.org.uk/photos/12/34/1234.jpg",
|
|
"https://www.geograph.org.uk/photo/1234")
|
|
self.assertEqual(source, "url")
|
|
self.assertRegex(key, r"^url-[0-9a-f]{16}$")
|
|
|
|
def test_flickr_cdn_without_id_falls_back_to_url_key(self):
|
|
source, key = self.key("https://c1.staticflickr.com/nodigits.jpg")
|
|
self.assertEqual(source, "url")
|
|
|
|
def test_not_a_url_is_none(self):
|
|
self.assertIsNone(self.key("not a url"))
|
|
self.assertIsNone(self.key(""))
|
|
|
|
|
|
class Grouping(TestCase):
|
|
def test_nodes_sharing_a_photo_group_together(self):
|
|
nodes = rb.readindex(json.dumps([
|
|
index_entry("A,_E1", "https://farm1.static.flickr.com/1/111_a_m.jpg",
|
|
page="https://www.flickr.com/photos/x/111/", copyright="P", licence="L"),
|
|
index_entry("B,_E2", "https://farm1.static.flickr.com/1/111_a_m.jpg",
|
|
page="https://www.flickr.com/photos/x/111/"),
|
|
index_entry("C,_E3", "https://farm1.static.flickr.com/1/222_b_m.jpg"),
|
|
index_entry("D,_E4", ""),
|
|
index_entry("E,_E5", "junk"),
|
|
]).encode())
|
|
groups, rejected = rb.groupphotos(nodes)
|
|
self.assertEqual(sorted(groups), ["111", "222"])
|
|
g = groups["111"]
|
|
self.assertEqual(g["source"], "flickr")
|
|
self.assertEqual(g["id"], "111")
|
|
self.assertEqual([n["param"] for n in g["nodes"]], ["A,_E1", "B,_E2"])
|
|
self.assertEqual(g["copyright"], "P")
|
|
self.assertEqual(g["licence"], "L")
|
|
self.assertEqual(g["page"], "https://www.flickr.com/photos/x/111/")
|
|
self.assertEqual(len(rejected), 1)
|
|
self.assertEqual(rejected[0][0]["param"], "E,_E5")
|
|
|
|
def test_first_non_empty_copyright_wins(self):
|
|
nodes = rb.readindex(json.dumps([
|
|
index_entry("A", "https://farm1.static.flickr.com/1/111_a_m.jpg"),
|
|
index_entry("B", "https://farm1.static.flickr.com/1/111_a_m.jpg", copyright="Later"),
|
|
]).encode())
|
|
groups, _ = rb.groupphotos(nodes)
|
|
self.assertEqual(groups["111"]["copyright"], "Later")
|
|
|
|
|
|
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"], 8)
|
|
self.assertEqual(conf["backoff"], 15.0)
|
|
self.assertEqual(rb.BACKOFF_CAP, 900.0)
|
|
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, 8))
|
|
|
|
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_report_flag(self):
|
|
base = ["--settings", os.path.join(self.dir, "none"), "/d"]
|
|
self.assertFalse(rb.settings(rb.parseargs(base), environ={})["report"])
|
|
self.assertTrue(rb.settings(rb.parseargs(base + ["--report"]), environ={})["report"])
|
|
|
|
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"))
|
|
|
|
|
|
class Retrying(TestCase):
|
|
def test_success_returns_body(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b"ok"))
|
|
self.assertEqual(rb.fetch("http://t/x", "ctx"), b"ok")
|
|
self.assertEqual(rb.OPEN.requests[0].get_header("User-agent"), rb.USER_AGENT)
|
|
|
|
def test_404_is_not_retried(self):
|
|
rb.OPEN = FakeOpen(http_error(404, b"gone"))
|
|
with self.assertRaises(rb.HTTPFailure) as cm:
|
|
rb.fetch("http://t/x", "ctx")
|
|
self.assertEqual(cm.exception.code, 404)
|
|
self.assertIn("gone", str(cm.exception))
|
|
self.assertEqual(len(rb.OPEN.requests), 1)
|
|
|
|
def test_5xx_and_429_retry_then_succeed(self):
|
|
rb.OPEN = FakeOpen(http_error(503), http_error(429), FakeResponse(b"ok"))
|
|
self.assertEqual(rb.fetch("http://t/x", "ctx"), b"ok")
|
|
self.assertEqual(len(rb.OPEN.requests), 3)
|
|
self.assertEqual(rb.STATS["retries"], 2)
|
|
|
|
def test_connection_errors_retry(self):
|
|
rb.OPEN = FakeOpen(OSError("reset"), FakeResponse(b"ok"))
|
|
self.assertEqual(rb.fetch("http://t/x", "ctx"), b"ok")
|
|
|
|
def test_gives_up_after_attempts(self):
|
|
rb.ATTEMPTS = 3
|
|
rb.OPEN = FakeOpen(http_error(500), http_error(500), http_error(500))
|
|
with self.assertRaises(rb.HTTPFailure) as cm:
|
|
rb.fetch("http://t/x", "ctx")
|
|
self.assertIn("3 attempts", str(cm.exception))
|
|
|
|
def test_retry_after_header_is_honoured(self):
|
|
waits = []
|
|
rb.SLEEP = waits.append
|
|
rb.OPEN = FakeOpen(http_error(429, headers={"Retry-After": "12"}), FakeResponse(b"ok"))
|
|
rb.fetch("http://t/x", "ctx")
|
|
self.assertIn(12.0, waits)
|
|
|
|
def test_backoff_grows_and_caps(self):
|
|
rb.BACKOFF_BASE = 5
|
|
self.assertLessEqual(rb.backoff(1), 7.5)
|
|
self.assertGreaterEqual(rb.backoff(1), 2.5)
|
|
self.assertLessEqual(rb.backoff(20), rb.BACKOFF_CAP * 1.5)
|
|
|
|
def test_post_sends_body_and_headers(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b"{}"))
|
|
rb.fetch("http://t/x", "ctx", headers={"x-api-key": "k"}, data=b"body", method="POST")
|
|
req = rb.OPEN.requests[0]
|
|
self.assertEqual(req.get_method(), "POST")
|
|
self.assertEqual(req.data, b"body")
|
|
self.assertEqual(req.get_header("X-api-key"), "k")
|
|
|
|
def test_download_streams_to_path(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b"x" * 1000, headers={"Content-Type": "image/jpeg"}))
|
|
path = os.path.join(self.dir, "p.part")
|
|
self.assertEqual(rb.download("http://t/p.jpg", "ctx", path), ("image/jpeg", 1000))
|
|
self.assertEqual(os.path.getsize(path), 1000)
|
|
|
|
def test_pace_sleeps_before_a_flickr_call(self):
|
|
waits = []
|
|
rb.SLEEP = waits.append
|
|
rb.PACE = 2.0
|
|
rb.OPEN = FakeOpen(FakeResponse({"stat": "ok"}))
|
|
rb.flickrcall("flickr.test.echo", {}, "KEY")
|
|
self.assertEqual(waits, [2.0])
|
|
|
|
def test_429_slows_every_later_flickr_call(self):
|
|
waits = []
|
|
rb.SLEEP = waits.append
|
|
rb.PACE = rb.PACE_FLOOR = 1.0
|
|
rb.OPEN = FakeOpen(http_error(429), FakeResponse({"stat": "ok"}),
|
|
FakeResponse({"stat": "ok"}))
|
|
rb.flickrcall("flickr.test.echo", {}, "KEY")
|
|
self.assertEqual(rb.PACE, 2.0)
|
|
rb.flickrcall("flickr.test.echo", {}, "KEY")
|
|
self.assertEqual(waits[-1], 2.0)
|
|
self.assertEqual(rb.PACEHELD, rb.PACE_HOLD - 2)
|
|
self.assertIn("pacing", self.stderr.getvalue())
|
|
|
|
def test_pace_relaxes_back_to_the_configured_floor(self):
|
|
rb.PACE_FLOOR = 1.0
|
|
rb.PACE = 4.0
|
|
rb.OPEN = FakeOpen(*[FakeResponse({"stat": "ok"}) for _ in range(40)])
|
|
for _ in range(40):
|
|
rb.flickrcall("flickr.test.echo", {}, "KEY")
|
|
self.assertEqual(rb.PACE, 1.0)
|
|
|
|
def test_pace_never_exceeds_the_cap(self):
|
|
rb.PACE = rb.PACE_FLOOR = 8.0
|
|
rb.OPEN = FakeOpen(http_error(429), FakeResponse({"stat": "ok"}))
|
|
rb.flickrcall("flickr.test.echo", {}, "KEY")
|
|
self.assertEqual(rb.PACE, rb.PACE_CAP)
|
|
|
|
def test_429_from_immich_leaves_the_pace_alone(self):
|
|
rb.PACE = rb.PACE_FLOOR = 1.0
|
|
rb.OPEN = FakeOpen(http_error(429), FakeResponse({"id": "u1"}))
|
|
rb.Immich("https://i/api", "k").me()
|
|
self.assertEqual(rb.PACE, 1.0)
|
|
|
|
def test_pace_leaves_immich_and_the_index_alone(self):
|
|
waits = []
|
|
rb.SLEEP = waits.append
|
|
rb.PACE = 2.0
|
|
rb.OPEN = FakeOpen(FakeResponse({"id": "u1"}),
|
|
FakeResponse(json.dumps([index_entry("A", "")]).encode()))
|
|
rb.Immich("https://i/api", "k").me()
|
|
rb.loadindex("http://t/index")
|
|
self.assertEqual(waits, [])
|
|
|
|
def test_download_refuses_an_oversized_body(self):
|
|
rb.DOWNLOAD_CAP = 10
|
|
rb.OPEN = FakeOpen(FakeResponse(b"x" * 100))
|
|
path = os.path.join(self.dir, "p.part")
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
rb.download("http://t/p.jpg", "ctx", path)
|
|
self.assertIn("10", str(cm.exception))
|
|
self.assertEqual(len(rb.OPEN.requests), 1) # not a retryable failure
|
|
|
|
def test_debug_redacts_the_flickr_key(self):
|
|
rb.DEBUG = True
|
|
rb.OPEN = FakeOpen(FakeResponse({"stat": "ok"}))
|
|
rb.flickrcall("flickr.test.echo", {}, "SECRETKEY")
|
|
logged = self.stderr.getvalue()
|
|
self.assertIn("api_key=<redacted>", logged)
|
|
self.assertNotIn("SECRETKEY", logged)
|
|
|
|
|
|
def flickr_info(photoid="111", licence=5, realname="Real Name"):
|
|
return {"stat": "ok", "photo": {
|
|
"id": photoid, "license": str(licence), "originalformat": "jpg",
|
|
"owner": {"nsid": "12345678@N00", "username": "handle", "realname": realname},
|
|
"title": {"_content": "A title"}, "description": {"_content": "Words"},
|
|
"dates": {"taken": "2009-07-22 14:03:11", "posted": "1248345600"},
|
|
"urls": {"url": [{"type": "photopage",
|
|
"_content": "https://www.flickr.com/photos/12345678@N00/%s/" % photoid}]},
|
|
}}
|
|
|
|
def flickr_sizes(*sizes):
|
|
return {"stat": "ok", "sizes": {"size": [
|
|
{"label": l, "width": w, "height": h, "source": s, "media": "photo"}
|
|
for (l, w, h, s) in sizes]}}
|
|
|
|
|
|
class Flickr(TestCase):
|
|
def test_call_builds_unsigned_json_request(self):
|
|
rb.OPEN = FakeOpen(FakeResponse({"stat": "ok", "x": 1}))
|
|
self.assertEqual(rb.flickrcall("flickr.test.echo", {"a": "b"}, "KEY")["x"], 1)
|
|
url = rb.OPEN.requests[0].full_url
|
|
q = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
|
|
self.assertEqual(q["method"], ["flickr.test.echo"])
|
|
self.assertEqual(q["api_key"], ["KEY"])
|
|
self.assertEqual(q["format"], ["json"])
|
|
self.assertEqual(q["nojsoncallback"], ["1"])
|
|
self.assertEqual(q["a"], ["b"])
|
|
|
|
def test_api_failure_raises(self):
|
|
rb.OPEN = FakeOpen(FakeResponse({"stat": "fail", "code": 100, "message": "Invalid API Key"}))
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
rb.flickrcall("flickr.photos.getInfo", {"photo_id": "1"}, "k")
|
|
self.assertIn("Invalid API Key", str(cm.exception))
|
|
|
|
def test_transient_service_error_is_retried(self):
|
|
busy = {"stat": "fail", "code": 105, "message": "Service currently unavailable"}
|
|
rb.OPEN = FakeOpen(FakeResponse(busy), FakeResponse(busy),
|
|
FakeResponse({"stat": "ok", "x": 1}))
|
|
self.assertEqual(rb.flickrcall("flickr.test.echo", {}, "k")["x"], 1)
|
|
self.assertEqual(len(rb.OPEN.requests), 3)
|
|
self.assertEqual(rb.STATS["retries"], 2)
|
|
|
|
def test_transient_service_error_gives_up_after_attempts(self):
|
|
rb.ATTEMPTS = 2
|
|
busy = {"stat": "fail", "code": "106", "message": "Write operation failed"}
|
|
rb.OPEN = FakeOpen(FakeResponse(busy), FakeResponse(busy))
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
rb.flickrcall("flickr.test.echo", {}, "k")
|
|
self.assertIn("Write operation failed", str(cm.exception))
|
|
self.assertEqual(len(rb.OPEN.requests), 2)
|
|
|
|
def test_missing_or_private_photo_is_gone(self):
|
|
for code in (1, 2):
|
|
rb.OPEN = FakeOpen(FakeResponse({"stat": "fail", "code": code, "message": "nope"}))
|
|
with self.assertRaises(rb.PhotoGone):
|
|
rb.getinfo("1", "k")
|
|
|
|
def test_getinfo_flattens(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(flickr_info()))
|
|
info = rb.getinfo("111", "k")
|
|
self.assertEqual(info["id"], "111")
|
|
self.assertEqual(info["owner"], "12345678@N00")
|
|
self.assertEqual(info["owner_name"], "handle")
|
|
self.assertEqual(info["owner_realname"], "Real Name")
|
|
self.assertEqual(info["title"], "A title")
|
|
self.assertEqual(info["taken"], "2009-07-22 14:03:11")
|
|
self.assertEqual(info["posted"], "2009-07-23T10:40:00Z")
|
|
self.assertEqual(info["licence_id"], 5)
|
|
self.assertEqual(info["page"], "https://www.flickr.com/photos/12345678@N00/111/")
|
|
self.assertEqual(info["original_format"], "jpg")
|
|
|
|
def test_getsizes_normalises_numbers(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(flickr_sizes(("Small", "240", "180", "https://x/s.jpg"))))
|
|
sizes = rb.getsizes("111", "k")
|
|
self.assertEqual(sizes[0]["width"], 240)
|
|
self.assertEqual(sizes[0]["label"], "Small")
|
|
|
|
def test_picksizes_prefers_original_then_area(self):
|
|
sizes = [
|
|
{"label": "Small", "width": 240, "height": 180, "source": "s", "media": "photo"},
|
|
{"label": "Large", "width": 1024, "height": 768, "source": "l", "media": "photo"},
|
|
{"label": "Original", "width": 800, "height": 600, "source": "o", "media": "photo"},
|
|
{"label": "Video Player", "width": 640, "height": 480, "source": "v", "media": "video"},
|
|
]
|
|
self.assertEqual([s["label"] for s in rb.picksizes(sizes)], ["Original", "Large", "Small"])
|
|
|
|
def test_picksizes_empty(self):
|
|
self.assertEqual(rb.picksizes([]), [])
|
|
|
|
def test_extensionfor(self):
|
|
self.assertEqual(rb.extensionfor("https://x/1_o.png", ""), ".png")
|
|
self.assertEqual(rb.extensionfor("https://x/1_o.jpeg", ""), ".jpg")
|
|
self.assertEqual(rb.extensionfor("https://x/1_o", "image/gif"), ".gif")
|
|
self.assertEqual(rb.extensionfor("https://x/1_o.weird", "text/html"), ".jpg")
|
|
|
|
|
|
def sidecar(key="111", file="111.jpg", nodes=None, copyright="P", licence="https://creativecommons.org/licenses/by-sa/2.0/", flickr=True):
|
|
data = {"key": key, "file": file, "source": "flickr" if flickr else "url",
|
|
"rgl": {"copyright": copyright, "licence": licence,
|
|
"nodes": nodes or [{"param": "A,_E1", "name": "A, E1",
|
|
"url": "https://london.randomness.org.uk/wiki.cgi?A%2C_E1",
|
|
"lat": "51.5", "long": "-0.1"}]},
|
|
"fetched": "2026-09-09T21:14:00Z"}
|
|
if flickr:
|
|
data["flickr"] = {"id": key, "owner": "12345678@N00", "owner_name": "handle",
|
|
"owner_realname": "Real Name", "title": "A title",
|
|
"taken": "2009-07-22 14:03:11", "posted": "2009-07-23T10:40:00Z",
|
|
"licence_id": 5, "page": "https://www.flickr.com/photos/12345678@N00/%s/" % key,
|
|
"size_label": "Original", "width": 800, "height": 600}
|
|
else:
|
|
data["page"] = "https://www.geograph.org.uk/photos/1.jpg"
|
|
return data
|
|
|
|
|
|
class Sidecars(TestCase):
|
|
def test_round_trip_is_atomic(self):
|
|
path = rb.sidecarpath(self.dir, "111")
|
|
self.assertEqual(path, os.path.join(self.dir, "111.json"))
|
|
rb.writejson(path, sidecar())
|
|
self.assertEqual(rb.readsidecar(path)["key"], "111")
|
|
self.assertFalse(os.path.exists(path + ".part"))
|
|
|
|
def test_unreadable_sidecar_is_none(self):
|
|
self.assertIsNone(rb.readsidecar(os.path.join(self.dir, "none.json")))
|
|
self.write("bad.json", "{not json")
|
|
self.assertIsNone(rb.readsidecar(os.path.join(self.dir, "bad.json")))
|
|
|
|
def test_mediapath_finds_the_photo_not_the_sidecar(self):
|
|
self.write("111.json", "{}")
|
|
self.write("111.part", b"x")
|
|
self.assertIsNone(rb.mediapath(self.dir, "111"))
|
|
self.write("111.png", b"x")
|
|
self.assertEqual(rb.mediapath(self.dir, "111"), os.path.join(self.dir, "111.png"))
|
|
|
|
def test_iscomplete(self):
|
|
self.assertFalse(rb.iscomplete(self.dir, "111"))
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
|
|
self.assertFalse(rb.iscomplete(self.dir, "111")) # no file yet
|
|
self.write("111.jpg", b"x")
|
|
self.assertTrue(rb.iscomplete(self.dir, "111"))
|
|
self.write("111.part", b"x")
|
|
self.assertFalse(rb.iscomplete(self.dir, "111")) # interrupted refetch
|
|
|
|
def test_sidenodes_survives_any_shape(self):
|
|
self.assertEqual(rb.sidenodes({}), [])
|
|
self.assertEqual(rb.sidenodes({"rgl": "nonsense"}), [])
|
|
self.assertEqual(rb.sidenodes({"rgl": {"nodes": "nonsense"}}), [])
|
|
self.assertEqual(rb.sidenodes({"rgl": {"nodes": ["nonsense"]}}), [])
|
|
self.assertEqual(rb.sidenodes(sidecar())[0]["name"], "A, E1")
|
|
|
|
def test_listsidecars_unreadable_directory_is_an_error(self):
|
|
with self.assertRaises(rb.BackupError):
|
|
rb.listsidecars(os.path.join(self.dir, "nowhere"))
|
|
|
|
def test_listsidecars_sorted_and_skips_hash_cache(self):
|
|
rb.writejson(rb.sidecarpath(self.dir, "222"), sidecar(key="222", file="222.jpg"))
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
|
|
rb.writejson(os.path.join(self.dir, rb.HASHES), {})
|
|
rb.writejson(os.path.join(self.dir, rb.FAILURES), {})
|
|
self.assertEqual([k for k, _ in rb.listsidecars(self.dir)], ["111", "222"])
|
|
|
|
def test_utcnow_shape(self):
|
|
self.assertRegex(rb.utcnow(), r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ$")
|
|
|
|
|
|
class Licences(TestCase):
|
|
def test_label_from_url(self):
|
|
self.assertEqual(rb.licencelabel("https://creativecommons.org/licenses/by-nc-sa/2.0/deed.en_GB"), "CC BY-NC-SA 2.0")
|
|
self.assertEqual(rb.licencelabel("https://creativecommons.org/licenses/by/2.0/"), "CC BY 2.0")
|
|
self.assertEqual(rb.licencelabel("http://creativecommons.org/publicdomain/zero/1.0/"), "CC0 1.0")
|
|
self.assertIsNone(rb.licencelabel(""))
|
|
self.assertIsNone(rb.licencelabel("https://example.com/terms"))
|
|
|
|
def test_slug(self):
|
|
self.assertEqual(rb.licenceslug("CC BY-NC-SA 2.0"), "cc-by-nc-sa-2.0")
|
|
self.assertEqual(rb.licenceslug("All rights reserved"), "all-rights-reserved")
|
|
|
|
def test_rgl_licence_wins(self):
|
|
side = sidecar()
|
|
self.assertEqual(rb.licencefor(side),
|
|
("CC BY-SA 2.0", "https://creativecommons.org/licenses/by-sa/2.0/", "rgl"))
|
|
|
|
def test_flickr_fills_a_gap(self):
|
|
side = sidecar(licence="")
|
|
label, url, origin = rb.licencefor(side)
|
|
self.assertEqual((label, origin), ("CC BY-SA 2.0", "flickr"))
|
|
self.assertIn("by-sa/2.0", url)
|
|
|
|
def test_a_licence_url_we_cannot_name_beats_flickr(self):
|
|
side = sidecar(licence="https://example.com/our-own-terms")
|
|
self.assertEqual(rb.licencefor(side),
|
|
(None, "https://example.com/our-own-terms", "wiki"))
|
|
self.assertIn("Licence: https://example.com/our-own-terms (from the wiki)",
|
|
rb.describeblock(side))
|
|
self.assertNotIn("from Flickr", rb.describeblock(side))
|
|
self.assertIn("rgl/licence/unknown", rb.tagsfor(side))
|
|
|
|
def test_unknown_when_neither(self):
|
|
side = sidecar(licence="")
|
|
side["flickr"]["licence_id"] = None
|
|
self.assertEqual(rb.licencefor(side), (None, None, None))
|
|
|
|
def test_flickr_table_covers_published_ids(self):
|
|
self.assertEqual(rb.FLICKR_LICENCES[0][0], "All rights reserved")
|
|
self.assertEqual(rb.FLICKR_LICENCES[4][0], "CC BY 2.0")
|
|
self.assertEqual(rb.FLICKR_LICENCES[9][0], "CC0 1.0")
|
|
self.assertEqual(rb.FLICKR_LICENCES[10][0], "Public Domain Mark 1.0")
|
|
|
|
|
|
class Describing(TestCase):
|
|
def test_block_shape(self):
|
|
block = rb.describeblock(sidecar())
|
|
self.assertEqual(block.splitlines(), [
|
|
rb.MARKER,
|
|
"Photo: A title",
|
|
"Source: https://www.flickr.com/photos/12345678@N00/111/",
|
|
"Photographer: P",
|
|
"Licence: CC BY-SA 2.0 https://creativecommons.org/licenses/by-sa/2.0/",
|
|
"Used on:",
|
|
" A, E1 https://london.randomness.org.uk/wiki.cgi?A%2C_E1",
|
|
])
|
|
|
|
def test_photographer_falls_back_to_flickr(self):
|
|
side = sidecar(copyright="")
|
|
self.assertEqual(rb.photographer(side), "Real Name")
|
|
side["flickr"]["owner_realname"] = ""
|
|
self.assertEqual(rb.photographer(side), "handle")
|
|
|
|
def test_licence_from_flickr_is_marked_and_disagreement_noted(self):
|
|
side = sidecar(licence="")
|
|
self.assertIn("Licence: CC BY-SA 2.0 https://creativecommons.org/licenses/by-sa/2.0/ (from Flickr)",
|
|
rb.describeblock(side))
|
|
side = sidecar(licence="https://creativecommons.org/licenses/by-nc-sa/2.0/")
|
|
block = rb.describeblock(side)
|
|
self.assertIn("Licence: CC BY-NC-SA 2.0 https://creativecommons.org/licenses/by-nc-sa/2.0/", block)
|
|
self.assertIn("Note: Flickr lists this photo as CC BY-SA 2.0", block)
|
|
|
|
def test_no_licence_says_unknown(self):
|
|
side = sidecar(licence="")
|
|
side["flickr"]["licence_id"] = None
|
|
self.assertIn("Licence: unknown", rb.describeblock(side))
|
|
|
|
def test_non_flickr_photo(self):
|
|
side = sidecar(flickr=False)
|
|
block = rb.describeblock(side)
|
|
self.assertNotIn("Photo:", block)
|
|
self.assertIn("Source: https://www.geograph.org.uk/photos/1.jpg", block)
|
|
|
|
def test_several_pages(self):
|
|
side = sidecar(nodes=[
|
|
{"param": "A", "name": "A", "url": "https://x/A", "lat": "", "long": ""},
|
|
{"param": "B", "name": "B", "url": "https://x/B", "lat": "", "long": ""}])
|
|
self.assertTrue(rb.describeblock(side).endswith("Used on:\n A https://x/A\n B https://x/B"))
|
|
|
|
|
|
class Merging(TestCase):
|
|
block = "%s\nPhoto: x\nUsed on:\n A https://x/A" % rb.MARKER
|
|
|
|
def test_empty_or_boilerplate_is_written(self):
|
|
self.assertEqual(rb.mergedescription("", self.block), (self.block, "write"))
|
|
self.assertEqual(rb.mergedescription("OLYMPUS DIGITAL CAMERA", self.block), (self.block, "write"))
|
|
|
|
def test_existing_text_is_kept_above(self):
|
|
text, action = rb.mergedescription("My caption", self.block)
|
|
self.assertEqual(action, "append")
|
|
self.assertEqual(text, "My caption\n\n" + self.block)
|
|
|
|
def test_same_block_is_quiet(self):
|
|
held = "My caption\n\n" + self.block
|
|
self.assertEqual(rb.mergedescription(held, self.block), (held, "same"))
|
|
|
|
def test_changed_block_is_replaced_in_place(self):
|
|
held = "My caption\n\n" + self.block + "\n\nTrailing note"
|
|
newblock = self.block + "\n B https://x/B"
|
|
text, action = rb.mergedescription(held, newblock)
|
|
self.assertEqual(action, "replace")
|
|
self.assertEqual(text, "My caption\n\n" + newblock + "\n\nTrailing note")
|
|
|
|
def test_isboilerplate(self):
|
|
self.assertTrue(rb.isboilerplate("SONY DSC"))
|
|
self.assertFalse(rb.isboilerplate("The 100 Club"))
|
|
|
|
|
|
class Tags(TestCase):
|
|
def test_tags(self):
|
|
self.assertEqual(rb.tagsfor(sidecar()),
|
|
["rgl", "rgl/licence/cc-by-sa-2.0", "rgl/photographer/P"])
|
|
|
|
def test_unknown_licence_and_blank_photographer(self):
|
|
side = sidecar(licence="", copyright="")
|
|
side["flickr"]["licence_id"] = None
|
|
side["flickr"]["owner_realname"] = ""
|
|
side["flickr"]["owner_name"] = ""
|
|
self.assertEqual(rb.tagsfor(side), ["rgl", "rgl/licence/unknown"])
|
|
|
|
def test_photographer_whitespace_and_slashes(self):
|
|
side = sidecar(copyright=" Some / One ")
|
|
self.assertIn("rgl/photographer/Some - One", rb.tagsfor(side))
|
|
|
|
|
|
class FetchPhoto(TestCase):
|
|
def group(self, key="111", nodes=None):
|
|
nodes = nodes or [{"param": "A,_E1", "name": "A, E1",
|
|
"url": "https://london.randomness.org.uk/wiki.cgi?A%2C_E1",
|
|
"image": "https://farm1.static.flickr.com/1/111_a_m.jpg",
|
|
"page": "https://www.flickr.com/photos/x/111/",
|
|
"copyright": "P", "licence": "https://creativecommons.org/licenses/by-sa/2.0/",
|
|
"lat": "51.5", "long": "-0.1"}]
|
|
return {"key": key, "source": "flickr", "id": key, "image": nodes[0]["image"],
|
|
"page": nodes[0]["page"], "copyright": "P",
|
|
"licence": "https://creativecommons.org/licenses/by-sa/2.0/", "nodes": nodes}
|
|
|
|
def test_fetches_largest_and_writes_sidecar(self):
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse(flickr_info("111")),
|
|
FakeResponse(flickr_sizes(("Small", 240, 180, "https://x/111_s.jpg"),
|
|
("Original", 800, 600, "https://x/111_o.png"))),
|
|
FakeResponse(b"PNGDATA", headers={"Content-Type": "image/png"}))
|
|
side, action = rb.fetchphoto(self.dir, self.group(), "k")
|
|
self.assertEqual(action, "fetched")
|
|
self.assertEqual(side["file"], "111.png")
|
|
self.assertEqual(side["flickr"]["size_label"], "Original")
|
|
self.assertEqual(side["flickr"]["width"], 800)
|
|
self.assertEqual(side["rgl"]["nodes"][0]["param"], "A,_E1")
|
|
self.assertNotIn("image", side["rgl"]["nodes"][0])
|
|
self.assertTrue(rb.iscomplete(self.dir, "111"))
|
|
self.assertEqual(open(os.path.join(self.dir, "111.png"), "rb").read(), b"PNGDATA")
|
|
self.assertFalse(os.path.exists(os.path.join(self.dir, "111.part")))
|
|
|
|
def test_falls_back_when_largest_404s(self):
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse(flickr_info("111")),
|
|
FakeResponse(flickr_sizes(("Large", 1024, 768, "https://x/111_b.jpg"),
|
|
("Original", 800, 600, "https://x/111_o.jpg"))),
|
|
http_error(404),
|
|
FakeResponse(b"JPG", headers={"Content-Type": "image/jpeg"}))
|
|
side, action = rb.fetchphoto(self.dir, self.group(), "k")
|
|
self.assertEqual(side["flickr"]["size_label"], "Large")
|
|
self.assertIn("111_o.jpg", rb.OPEN.requests[2].full_url)
|
|
|
|
def test_gone_photo_raises_and_writes_nothing(self):
|
|
rb.OPEN = FakeOpen(FakeResponse({"stat": "fail", "code": 1, "message": "Photo not found"}))
|
|
with self.assertRaises(rb.PhotoGone):
|
|
rb.fetchphoto(self.dir, self.group(), "k")
|
|
self.assertEqual(os.listdir(self.dir), [])
|
|
|
|
def test_complete_photo_is_skipped_without_calls(self):
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
|
|
self.write("111.jpg", b"x")
|
|
rb.OPEN = FakeOpen()
|
|
side, action = rb.fetchphoto(self.dir, self.group(), "k")
|
|
self.assertEqual(action, "skipped")
|
|
|
|
def test_changed_nodes_rewrite_sidecar_only(self):
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
|
|
self.write("111.jpg", b"x")
|
|
rb.OPEN = FakeOpen()
|
|
g = self.group()
|
|
g["nodes"].append(dict(g["nodes"][0], param="B,_E2", name="B, E2", url="https://x/B"))
|
|
side, action = rb.fetchphoto(self.dir, g, "k")
|
|
self.assertEqual(action, "updated")
|
|
self.assertEqual(len(rb.readsidecar(rb.sidecarpath(self.dir, "111"))["rgl"]["nodes"]), 2)
|
|
self.assertEqual(rb.STATS["rewritten"], 1)
|
|
|
|
def test_nodes_are_sorted_by_param(self):
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse(flickr_info("111")),
|
|
FakeResponse(flickr_sizes(("Original", 800, 600, "https://x/111_o.jpg"))),
|
|
FakeResponse(b"J", headers={"Content-Type": "image/jpeg"}))
|
|
g = self.group()
|
|
g["nodes"].insert(0, dict(g["nodes"][0], param="Z,_E9", name="Z, E9"))
|
|
side, action = rb.fetchphoto(self.dir, g, "k")
|
|
self.assertEqual([n["param"] for n in side["rgl"]["nodes"]], ["A,_E1", "Z,_E9"])
|
|
# and the same order the comparison uses, so a re-run rewrites nothing
|
|
self.assertEqual(rb.fetchphoto(self.dir, g, "k")[1], "skipped")
|
|
|
|
def test_downloadany_skips_an_empty_body(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b""),
|
|
FakeResponse(b"JPG", headers={"Content-Type": "image/jpeg"}))
|
|
candidates = [{"label": "Original", "width": 8, "height": 6, "source": "https://x/o.jpg",
|
|
"media": "photo"},
|
|
{"label": "Large", "width": 4, "height": 3, "source": "https://x/l.jpg",
|
|
"media": "photo"}]
|
|
partial = os.path.join(self.dir, "111.part")
|
|
pick, ext = rb.downloadany(candidates, partial, "photo 111")
|
|
self.assertEqual((pick["label"], ext), ("Large", ".jpg"))
|
|
|
|
def test_downloadany_with_nothing_but_empty_bodies_fails(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b""))
|
|
candidates = [{"label": "Original", "width": 8, "height": 6, "source": "https://x/o.jpg",
|
|
"media": "photo"}]
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
rb.downloadany(candidates, os.path.join(self.dir, "111.part"), "photo 111")
|
|
self.assertIn("empty", str(cm.exception))
|
|
|
|
def test_partial_file_is_refetched(self):
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
|
|
self.write("111.jpg", b"x")
|
|
self.write("111.part", b"half")
|
|
rb.OPEN = FakeOpen(FakeResponse(flickr_info("111")),
|
|
FakeResponse(flickr_sizes(("Original", 800, 600, "https://x/111_o.jpg"))),
|
|
FakeResponse(b"FULL", headers={"Content-Type": "image/jpeg"}))
|
|
side, action = rb.fetchphoto(self.dir, self.group(), "k")
|
|
self.assertEqual(action, "fetched")
|
|
self.assertEqual(open(os.path.join(self.dir, "111.jpg"), "rb").read(), b"FULL")
|
|
|
|
def test_sidecar_without_an_rgl_block_is_rewritten_not_fatal(self):
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), {"key": "111", "file": "111.jpg"})
|
|
self.write("111.jpg", b"x")
|
|
rb.OPEN = FakeOpen()
|
|
side, action = rb.fetchphoto(self.dir, self.group(), "k")
|
|
self.assertEqual(action, "updated")
|
|
self.assertEqual(len(rb.readsidecar(rb.sidecarpath(self.dir, "111"))["rgl"]["nodes"]), 1)
|
|
|
|
def test_dry_run_calls_nothing(self):
|
|
rb.OPEN = FakeOpen()
|
|
side, action = rb.fetchphoto(self.dir, self.group(), "k", dryrun=True)
|
|
self.assertEqual(action, "would fetch")
|
|
self.assertEqual(os.listdir(self.dir), [])
|
|
|
|
def test_non_flickr_photo_downloads_image_url(self):
|
|
g = self.group(key="url-0123456789abcdef")
|
|
g["source"] = "url"
|
|
del g["id"]
|
|
g["image"] = g["nodes"][0]["image"] = "https://www.geograph.org.uk/p/1.jpg"
|
|
g["page"] = "https://www.geograph.org.uk/photo/1"
|
|
rb.OPEN = FakeOpen(FakeResponse(b"JPG", headers={"Content-Type": "image/jpeg"}))
|
|
side, action = rb.fetchphoto(self.dir, g, "k")
|
|
self.assertEqual(side["source"], "url")
|
|
self.assertEqual(side["page"], "https://www.geograph.org.uk/photo/1")
|
|
self.assertNotIn("flickr", side)
|
|
self.assertEqual(side["file"], "url-0123456789abcdef.jpg")
|
|
|
|
|
|
class FetchPhase(TestCase):
|
|
def conf(self, **over):
|
|
conf = {"directory": self.dir, "index_url": "http://t/index", "limit": None,
|
|
"dry_run": False, "quiet": False}
|
|
conf.update(over)
|
|
return conf
|
|
|
|
def test_runs_over_index_and_collects_failures(self):
|
|
index = [index_entry("A,_E1", "https://farm1.static.flickr.com/1/111_a_m.jpg",
|
|
page="https://www.flickr.com/photos/x/111/"),
|
|
index_entry("B,_E2", "https://farm1.static.flickr.com/1/222_b_m.jpg",
|
|
page="https://www.flickr.com/photos/x/222/"),
|
|
index_entry("C,_E3", "junk")]
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse(json.dumps(index).encode()),
|
|
FakeResponse(flickr_info("111")),
|
|
FakeResponse(flickr_sizes(("Original", 8, 6, "https://x/111_o.jpg"))),
|
|
FakeResponse(b"J", headers={"Content-Type": "image/jpeg"}),
|
|
FakeResponse({"stat": "fail", "code": 1, "message": "Photo not found"}))
|
|
failures = rb.fetchphase(self.conf(), "k")
|
|
self.assertTrue(rb.iscomplete(self.dir, "111"))
|
|
self.assertEqual(rb.STATS["fetched"], 1)
|
|
self.assertEqual(len(failures), 2)
|
|
reasons = [f[2] for f in failures.items]
|
|
self.assertTrue(any("Photo not found" in r for r in reasons), reasons)
|
|
self.assertTrue(any("not a url" in r for r in reasons), reasons)
|
|
names = [n for f in failures.items for n in f[1]]
|
|
self.assertIn("B, E2", names)
|
|
|
|
def test_limit_stops_early(self):
|
|
index = [index_entry("A", "https://farm1.static.flickr.com/1/111_a_m.jpg"),
|
|
index_entry("B", "https://farm1.static.flickr.com/1/222_b_m.jpg")]
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse(json.dumps(index).encode()),
|
|
FakeResponse(flickr_info("111")),
|
|
FakeResponse(flickr_sizes(("Original", 8, 6, "https://x/111_o.jpg"))),
|
|
FakeResponse(b"J", headers={"Content-Type": "image/jpeg"}))
|
|
rb.fetchphase(self.conf(limit=1), "k")
|
|
self.assertEqual(rb.STATS["fetched"], 1)
|
|
self.assertFalse(os.path.exists(rb.sidecarpath(self.dir, "222")))
|
|
|
|
def test_dry_run_creates_no_directory(self):
|
|
room = os.path.join(self.dir, "not-yet")
|
|
rb.OPEN = FakeOpen(FakeResponse(json.dumps(
|
|
[index_entry("A", "https://farm1.static.flickr.com/1/111_a_m.jpg")]).encode()))
|
|
rb.fetchphase(self.conf(directory=room, dry_run=True), "k")
|
|
self.assertFalse(os.path.exists(room))
|
|
|
|
def test_index_failure_is_fatal(self):
|
|
rb.OPEN = FakeOpen(http_error(404))
|
|
with self.assertRaises(rb.BackupError):
|
|
rb.fetchphase(self.conf(), "k")
|
|
|
|
|
|
def multipart_fields(request):
|
|
"""Parse the multipart body FakeOpen captured into {name: (filename, bytes)}."""
|
|
ctype = request.get_header("Content-type")
|
|
boundary = ctype.split("boundary=")[1].encode()
|
|
fields = {}
|
|
for part in request.data.split(b"--" + boundary)[1:-1]:
|
|
head, _, body = part.partition(b"\r\n\r\n")
|
|
name = re.search(rb'name="([^"]+)"', head).group(1).decode()
|
|
fname = re.search(rb'filename="([^"]+)"', head)
|
|
fields[name] = (fname.group(1).decode() if fname else None, body.rstrip(b"\r\n"))
|
|
return fields
|
|
|
|
|
|
class ImmichClient(TestCase):
|
|
def client(self, *responses):
|
|
rb.OPEN = FakeOpen(*responses)
|
|
return rb.Immich("https://immich.example/api", "KEY")
|
|
|
|
def test_call_sets_headers_and_parses(self):
|
|
im = self.client(FakeResponse({"id": "u1"}))
|
|
self.assertEqual(im.me(), {"id": "u1"})
|
|
req = rb.OPEN.requests[0]
|
|
self.assertEqual(req.full_url, "https://immich.example/api/users/me")
|
|
self.assertEqual(req.get_header("X-api-key"), "KEY")
|
|
|
|
def test_http_error_becomes_backuperror(self):
|
|
im = self.client(http_error(401, b"unauthorized"))
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
im.me()
|
|
self.assertIn("401", str(cm.exception))
|
|
|
|
def test_a_body_that_is_not_json_becomes_backuperror(self):
|
|
im = self.client(FakeResponse(b"<html>proxy says no</html>"))
|
|
with self.assertRaises(rb.BackupError) as cm:
|
|
im.me()
|
|
self.assertIn("/users/me", str(cm.exception))
|
|
|
|
def test_me_without_an_id_raises(self):
|
|
im = self.client(FakeResponse({"email": "someone@example.invalid"}))
|
|
with self.assertRaises(rb.BackupError):
|
|
im.me()
|
|
|
|
def test_createalbum_without_an_id_raises(self):
|
|
im = self.client(FakeResponse({"albumName": "RGL"}))
|
|
with self.assertRaises(rb.BackupError):
|
|
im.createalbum("RGL")
|
|
|
|
def test_assets_pages_and_filters_owner(self):
|
|
page1 = {"assets": {"items": [
|
|
{"id": "a1", "ownerId": "u1", "checksum": "c1", "originalFileName": "1.jpg",
|
|
"exifInfo": {"description": "d", "latitude": 51.5, "longitude": -0.1,
|
|
"dateTimeOriginal": "2009-07-22T14:03:11.000Z"}},
|
|
{"id": "a2", "ownerId": "other", "checksum": "c2"}], "nextPage": "2"}}
|
|
page2 = {"assets": {"items": [
|
|
{"id": "a3", "ownerId": "u1", "checksum": "c3", "exifInfo": None}], "nextPage": None}}
|
|
im = self.client(FakeResponse(page1), FakeResponse(page2))
|
|
assets = im.assets("u1")
|
|
self.assertEqual(sorted(assets), ["c1", "c3"])
|
|
self.assertEqual(assets["c1"]["latitude"], 51.5)
|
|
self.assertEqual(assets["c1"]["taken"], "2009-07-22T14:03:11.000Z")
|
|
self.assertEqual(assets["c3"]["description"], "")
|
|
self.assertIsNone(assets["c3"]["latitude"])
|
|
body = json.loads(rb.OPEN.requests[1].data)
|
|
self.assertEqual(body["page"], 2)
|
|
self.assertTrue(body["withExif"])
|
|
self.assertTrue(body["withArchived"])
|
|
|
|
def test_asset_reads_one_in_the_same_shape(self):
|
|
im = self.client(FakeResponse({
|
|
"id": "a1", "originalFileName": "111.jpg",
|
|
"exifInfo": {"description": "Mine", "latitude": 51.5, "longitude": -0.1,
|
|
"dateTimeOriginal": "2009-07-22T14:03:11.000Z"}}))
|
|
self.assertEqual(im.asset("a1"), {
|
|
"id": "a1", "name": "111.jpg", "description": "Mine",
|
|
"latitude": 51.5, "longitude": -0.1,
|
|
"taken": "2009-07-22T14:03:11.000Z"})
|
|
req = rb.OPEN.requests[0]
|
|
self.assertEqual(req.get_method(), "GET")
|
|
self.assertTrue(req.full_url.endswith("/assets/a1"))
|
|
|
|
def test_asset_without_exif_is_blank_not_missing(self):
|
|
im = self.client(FakeResponse({"id": "a1", "exifInfo": None}))
|
|
asset = im.asset("a1")
|
|
self.assertEqual((asset["description"], asset["taken"]), ("", ""))
|
|
self.assertIsNone(asset["latitude"])
|
|
|
|
def test_upload_is_multipart_with_checksum_header(self):
|
|
path = self.write("111.jpg", b"JPEGDATA")
|
|
im = self.client(FakeResponse({"id": "a9", "status": "created"}, status=201))
|
|
result = im.upload(path, "Q0hFQ0s=", "111", "2009-07-22T14:03:11Z", "2026-09-09T00:00:00Z")
|
|
self.assertEqual(result, ("a9", "created"))
|
|
req = rb.OPEN.requests[0]
|
|
self.assertEqual(req.get_method(), "POST")
|
|
self.assertTrue(req.full_url.endswith("/assets"))
|
|
self.assertEqual(req.get_header("X-immich-checksum"), "Q0hFQ0s=")
|
|
fields = multipart_fields(req)
|
|
self.assertEqual(fields["assetData"], ("111.jpg", b"JPEGDATA"))
|
|
self.assertEqual(fields["deviceAssetId"][1], b"111")
|
|
self.assertEqual(fields["deviceId"][1], rb.DEVICE_ID.encode())
|
|
self.assertEqual(fields["fileCreatedAt"][1], b"2009-07-22T14:03:11Z")
|
|
self.assertEqual(fields["fileModifiedAt"][1], b"2026-09-09T00:00:00Z")
|
|
|
|
def test_upload_duplicate(self):
|
|
path = self.write("111.jpg", b"J")
|
|
im = self.client(FakeResponse({"id": "a1", "status": "duplicate"}))
|
|
self.assertEqual(im.upload(path, "x", "111", "t", "t"), ("a1", "duplicate"))
|
|
|
|
def test_upload_empty_response_raises(self):
|
|
path = self.write("111.jpg", b"J")
|
|
im = self.client(FakeResponse(b""))
|
|
with self.assertRaises(rb.BackupError):
|
|
im.upload(path, "x", "111", "t", "t")
|
|
|
|
def test_update_sends_only_given_fields(self):
|
|
im = self.client(FakeResponse(b""))
|
|
im.update("a1", description="d", latitude=51.5, longitude=-0.1)
|
|
req = rb.OPEN.requests[0]
|
|
self.assertEqual(req.get_method(), "PUT")
|
|
self.assertTrue(req.full_url.endswith("/assets/a1"))
|
|
self.assertEqual(json.loads(req.data), {"description": "d", "latitude": 51.5, "longitude": -0.1})
|
|
|
|
def test_albums(self):
|
|
im = self.client(FakeResponse([{"albumName": "RGL", "id": "al1", "ownerId": "u1"},
|
|
{"albumName": "RGL", "id": "shared", "ownerId": "u2"},
|
|
{"albumName": "Nameless"}]),
|
|
FakeResponse({"id": "al2"}),
|
|
FakeResponse({"assets": [{"id": "a1"}, {"id": "a2"}]}),
|
|
FakeResponse([{"id": "a3", "success": True}, {"id": "a1", "success": False, "error": "duplicate"}]))
|
|
self.assertEqual(im.albums("u1"), {"RGL": "al1"})
|
|
self.assertEqual(im.createalbum("X"), "al2")
|
|
self.assertEqual(im.albumassets("al1"), {"a1", "a2"})
|
|
self.assertEqual(im.addtoalbum("al1", ["a3", "a1"]), 1)
|
|
|
|
def test_tags(self):
|
|
im = self.client(FakeResponse([{"value": "rgl", "id": "t1"}, {"value": "rgl/licence/x", "id": "t2"}]),
|
|
FakeResponse([{"success": True}, {"success": False, "error": "duplicate"}]))
|
|
self.assertEqual(im.tagids(["rgl", "rgl/licence/x"]), {"rgl": "t1", "rgl/licence/x": "t2"})
|
|
self.assertEqual(im.tagassets("t1", ["a1", "a2"]), (1, 1))
|
|
|
|
def test_tagids_skips_malformed_entries(self):
|
|
im = self.client(FakeResponse([{"value": "rgl", "id": "t1"}, {"value": "rgl/broken"}]))
|
|
self.assertEqual(im.tagids(["rgl", "rgl/broken"]), {"rgl": "t1"})
|
|
|
|
def test_batches_of_500(self):
|
|
ids = ["a%d" % i for i in range(1200)]
|
|
im = self.client(FakeResponse([{"success": True}] * 500),
|
|
FakeResponse([{"success": True}] * 500),
|
|
FakeResponse([{"success": True}] * 200))
|
|
self.assertEqual(im.addtoalbum("al", ids), 1200)
|
|
self.assertEqual(len(rb.OPEN.requests), 3)
|
|
|
|
|
|
class Hashing(TestCase):
|
|
def test_checksum_is_base64_sha1(self):
|
|
path = self.write("f", b"hello")
|
|
self.assertEqual(rb.checksum(path), "qvTGHdzF6KLavt4PO0gs2a6pQ00=")
|
|
|
|
def test_cache_reuses_until_file_changes(self):
|
|
path = self.write("f", b"hello")
|
|
cache = rb.Hashes(os.path.join(self.dir, rb.HASHES))
|
|
first = cache.of(path, "f", os.stat(path))
|
|
cache.save()
|
|
again = rb.Hashes(os.path.join(self.dir, rb.HASHES))
|
|
self.assertEqual(again.of(path, "f", os.stat(path)), first)
|
|
self.assertFalse(again.dirty)
|
|
self.write("f", b"changed")
|
|
os.utime(path, (1, 1))
|
|
self.assertNotEqual(again.of(path, "f", os.stat(path)), first)
|
|
self.assertTrue(again.dirty)
|
|
|
|
|
|
class PlanAsset(TestCase):
|
|
def asset(self, **over):
|
|
a = {"id": "a1", "name": "111.jpg", "description": "", "latitude": None,
|
|
"longitude": None, "taken": ""}
|
|
a.update(over)
|
|
return a
|
|
|
|
def test_fresh_asset_gets_everything(self):
|
|
fields = rb.planasset(sidecar(), self.asset())
|
|
self.assertEqual(fields["description"], rb.describeblock(sidecar()))
|
|
self.assertEqual((fields["latitude"], fields["longitude"]), (51.5, -0.1))
|
|
self.assertEqual(fields["dateTimeOriginal"], "2009-07-22T14:03:11Z")
|
|
|
|
def test_existing_values_are_kept(self):
|
|
a = self.asset(description="Mine\n\n" + rb.describeblock(sidecar()),
|
|
latitude=1.0, longitude=2.0, taken="2001-01-01T00:00:00.000Z")
|
|
self.assertEqual(rb.planasset(sidecar(), a), {})
|
|
|
|
def test_hand_written_description_is_appended_to(self):
|
|
fields = rb.planasset(sidecar(), self.asset(description="Mine", latitude=1.0, longitude=2.0, taken="t"))
|
|
self.assertEqual(list(fields), ["description"])
|
|
self.assertTrue(fields["description"].startswith("Mine\n\n"))
|
|
|
|
def test_bad_coordinates_are_skipped(self):
|
|
side = sidecar(nodes=[{"param": "A", "name": "A", "url": "u", "lat": "", "long": ""}])
|
|
fields = rb.planasset(side, self.asset())
|
|
self.assertNotIn("latitude", fields)
|
|
|
|
def test_flickrtaken(self):
|
|
self.assertEqual(rb.flickrtaken(sidecar()), "2009-07-22T14:03:11Z")
|
|
self.assertIsNone(rb.flickrtaken(sidecar(flickr=False)))
|
|
side = sidecar()
|
|
side["flickr"]["taken"] = "garbage"
|
|
self.assertIsNone(rb.flickrtaken(side))
|
|
|
|
def test_takenat(self):
|
|
path = self.write("111.jpg", b"x")
|
|
os.utime(path, (1000000000, 1000000000))
|
|
self.assertEqual(rb.takenat(sidecar(), path), "2009-07-22T14:03:11Z")
|
|
self.assertEqual(rb.takenat(sidecar(flickr=False), path), "2001-09-09T01:46:40Z")
|
|
side = sidecar()
|
|
side["flickr"]["taken"] = "garbage"
|
|
self.assertEqual(rb.takenat(side, path), "2001-09-09T01:46:40Z")
|
|
|
|
|
|
class SyncPhase(TestCase):
|
|
def setUp(self):
|
|
super().setUp()
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
|
|
self.write("111.jpg", b"NEW")
|
|
rb.writejson(rb.sidecarpath(self.dir, "222"), sidecar(key="222", file="222.jpg", copyright="Q"))
|
|
self.write("222.jpg", b"OLD")
|
|
self.old = rb.checksum(os.path.join(self.dir, "222.jpg"))
|
|
|
|
def conf(self, **over):
|
|
conf = {"directory": self.dir, "dry_run": False, "quiet": False, "limit": None}
|
|
conf.update(over)
|
|
return conf
|
|
|
|
def responses(self):
|
|
return [
|
|
FakeResponse({"id": "u1"}), # me
|
|
FakeResponse({"assets": {"items": [
|
|
{"id": "a2", "ownerId": "u1", "checksum": self.old,
|
|
"exifInfo": {"description": "Mine", "latitude": 1, "longitude": 2,
|
|
"dateTimeOriginal": "t"}}], "nextPage": None}}), # assets
|
|
FakeResponse({"id": "a1", "status": "created"}, status=201), # upload 111
|
|
FakeResponse(b""), # update a2 (append)
|
|
FakeResponse([]), # albums
|
|
FakeResponse({"id": "al1"}), # create album
|
|
FakeResponse({"assets": []}), # album assets
|
|
FakeResponse([{"success": True}, {"success": True}]), # add to album
|
|
FakeResponse([{"value": "rgl", "id": "t1"},
|
|
{"value": "rgl/licence/cc-by-sa-2.0", "id": "t2"},
|
|
{"value": "rgl/photographer/P", "id": "t3"},
|
|
{"value": "rgl/photographer/Q", "id": "t4"}]), # tags
|
|
FakeResponse([{"success": True}, {"success": True}]), # t1
|
|
FakeResponse([{"success": True}, {"success": True}]), # t2
|
|
FakeResponse([{"success": True}]), # t3
|
|
FakeResponse([{"success": True}]), # t4
|
|
FakeResponse({"id": "a1", "originalFileName": "111.jpg",
|
|
"exifInfo": {}}), # re-read a1
|
|
FakeResponse(b""), # update a1
|
|
]
|
|
|
|
def test_uploads_new_reuses_existing_and_annotates(self):
|
|
rb.OPEN = FakeOpen(*self.responses())
|
|
failures = rb.syncphase(self.conf(), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual(len(failures), 0)
|
|
self.assertEqual(rb.STATS["uploaded"], 1)
|
|
self.assertEqual(rb.STATS["existing"], 1)
|
|
reqs = rb.OPEN.requests
|
|
self.assertTrue(reqs[2].full_url.endswith("/assets"))
|
|
upd2 = json.loads(reqs[3].data)
|
|
self.assertEqual(list(upd2), ["description"])
|
|
self.assertTrue(upd2["description"].startswith("Mine\n\n"))
|
|
self.assertEqual(json.loads(reqs[5].data)["albumName"], "RGL")
|
|
self.assertEqual(sorted(json.loads(reqs[7].data)["ids"]), ["a1", "a2"])
|
|
# the uploaded asset is annotated last, against what immich now holds
|
|
self.assertEqual((reqs[13].get_method(), reqs[13].full_url),
|
|
("GET", "https://i/api/assets/a1"))
|
|
upd1 = json.loads(reqs[14].data)
|
|
self.assertIn("latitude", upd1)
|
|
self.assertEqual(upd1["description"], rb.describeblock(sidecar()))
|
|
self.assertTrue(os.path.exists(os.path.join(self.dir, rb.HASHES)))
|
|
|
|
def test_duplicate_upload_appends_to_what_immich_already_holds(self):
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [], "nextPage": None}}),
|
|
FakeResponse({"id": "a1", "status": "duplicate"}), # upload 111
|
|
FakeResponse([{"albumName": "RGL", "id": "al1", "ownerId": "u1"}]),
|
|
FakeResponse({"assets": []}),
|
|
FakeResponse([{"success": True}]),
|
|
FakeResponse([{"value": "rgl", "id": "t1"},
|
|
{"value": "rgl/licence/cc-by-sa-2.0", "id": "t2"},
|
|
{"value": "rgl/photographer/P", "id": "t3"}]),
|
|
FakeResponse([{"success": True}]),
|
|
FakeResponse([{"success": True}]),
|
|
FakeResponse([{"success": True}]),
|
|
FakeResponse({"id": "a1", "originalFileName": "mine.jpg",
|
|
"exifInfo": {"description": "My own words",
|
|
"latitude": 1, "longitude": 2,
|
|
"dateTimeOriginal": "2001-01-01T00:00:00.000Z"}}),
|
|
FakeResponse(b""))
|
|
failures = rb.syncphase(self.conf(limit=1), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual(len(failures), 0)
|
|
self.assertEqual(rb.STATS["uploaded"], 0)
|
|
self.assertEqual(rb.STATS["existing"], 1)
|
|
last = rb.OPEN.requests[-1]
|
|
self.assertEqual((last.get_method(), last.full_url),
|
|
("PUT", "https://i/api/assets/a1"))
|
|
body = json.loads(last.data)
|
|
self.assertEqual(list(body), ["description"])
|
|
self.assertTrue(body["description"].startswith("My own words\n\n"))
|
|
|
|
def test_a_repeated_checksum_is_uploaded_once(self):
|
|
self.write("222.jpg", b"NEW") # byte-identical to 111.jpg
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [], "nextPage": None}}),
|
|
FakeResponse({"id": "a1", "status": "created"}, status=201), # upload 111
|
|
FakeResponse([{"albumName": "RGL", "id": "al1", "ownerId": "u1"}]),
|
|
FakeResponse({"assets": [{"id": "a1"}]}),
|
|
FakeResponse([{"value": "rgl", "id": "t1"},
|
|
{"value": "rgl/licence/cc-by-sa-2.0", "id": "t2"},
|
|
{"value": "rgl/photographer/Q", "id": "t3"}]),
|
|
FakeResponse([{"success": True}]),
|
|
FakeResponse([{"success": True}]),
|
|
FakeResponse([{"success": True}]),
|
|
FakeResponse({"id": "a1", "exifInfo": {}}), # re-read a1
|
|
FakeResponse(b""))
|
|
failures = rb.syncphase(self.conf(), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual(len(failures), 0)
|
|
uploads = [r for r in rb.OPEN.requests if r.full_url.endswith("/api/assets")]
|
|
self.assertEqual(len(uploads), 1)
|
|
self.assertEqual(rb.STATS["uploaded"], 1)
|
|
self.assertEqual(rb.STATS["existing"], 1)
|
|
|
|
def test_second_run_is_quiet(self):
|
|
block1 = rb.describeblock(sidecar())
|
|
block2 = rb.describeblock(sidecar(key="222", file="222.jpg", copyright="Q"))
|
|
new = rb.checksum(os.path.join(self.dir, "111.jpg"))
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [
|
|
{"id": "a1", "ownerId": "u1", "checksum": new,
|
|
"exifInfo": {"description": block1, "latitude": 51.5, "longitude": -0.1, "dateTimeOriginal": "t"}},
|
|
{"id": "a2", "ownerId": "u1", "checksum": self.old,
|
|
"exifInfo": {"description": "Mine\n\n" + block2, "latitude": 1, "longitude": 2, "dateTimeOriginal": "t"}}],
|
|
"nextPage": None}}),
|
|
FakeResponse([{"albumName": "RGL", "id": "al1", "ownerId": "u1"}]),
|
|
FakeResponse({"assets": [{"id": "a1"}, {"id": "a2"}]}),
|
|
FakeResponse([{"value": "rgl", "id": "t1"}, {"value": "rgl/licence/cc-by-sa-2.0", "id": "t2"},
|
|
{"value": "rgl/photographer/P", "id": "t3"}, {"value": "rgl/photographer/Q", "id": "t4"}]),
|
|
FakeResponse([{"error": "duplicate"}, {"error": "duplicate"}]),
|
|
FakeResponse([{"error": "duplicate"}, {"error": "duplicate"}]),
|
|
FakeResponse([{"error": "duplicate"}]),
|
|
FakeResponse([{"error": "duplicate"}]))
|
|
failures = rb.syncphase(self.conf(), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual(len(failures), 0)
|
|
self.assertEqual(rb.STATS["uploaded"], 0)
|
|
self.assertEqual(rb.STATS["updated"], 0)
|
|
|
|
def test_dry_run_only_reads(self):
|
|
rb.OPEN = FakeOpen(FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [], "nextPage": None}}),
|
|
FakeResponse([]))
|
|
rb.syncphase(self.conf(dry_run=True), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual([r.get_method() for r in rb.OPEN.requests], ["GET", "POST", "GET"])
|
|
self.assertIn("would upload", self.stdout.getvalue())
|
|
self.assertFalse(os.path.exists(os.path.join(self.dir, rb.HASHES)))
|
|
|
|
def test_dry_run_limit_counts_what_it_looked_at(self):
|
|
rb.OPEN = FakeOpen(FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [], "nextPage": None}}),
|
|
FakeResponse([]))
|
|
rb.syncphase(self.conf(dry_run=True, limit=1), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual(self.stdout.getvalue().count("would upload"), 1)
|
|
|
|
def test_unsupported_file_is_a_failure_not_a_stop(self):
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [], "nextPage": None}}),
|
|
http_error(400, b'{"message":"unsupported file type"}'), # upload 111
|
|
FakeResponse({"id": "a2", "status": "created"}, status=201), # upload 222
|
|
FakeResponse([]), FakeResponse({"id": "al1"}), FakeResponse({"assets": []}),
|
|
FakeResponse([{"success": True}]),
|
|
FakeResponse([{"value": "rgl", "id": "t1"}, {"value": "rgl/licence/cc-by-sa-2.0", "id": "t2"},
|
|
{"value": "rgl/photographer/Q", "id": "t4"}]),
|
|
FakeResponse([{"success": True}]), FakeResponse([{"success": True}]), FakeResponse([{"success": True}]),
|
|
FakeResponse({"id": "a2", "exifInfo": {}}), # re-read a2
|
|
FakeResponse(b"")) # update a2
|
|
failures = rb.syncphase(self.conf(), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual(len(failures), 1)
|
|
self.assertIn("unsupported", failures.items[0][2])
|
|
self.assertEqual(rb.STATS["uploaded"], 1)
|
|
|
|
def test_malformed_sidecar_is_a_failure_not_a_stop(self):
|
|
room = os.path.join(self.dir, "other")
|
|
os.mkdir(room)
|
|
rb.writejson(os.path.join(room, "333.json"), {"key": "333"})
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [], "nextPage": None}}),
|
|
FakeResponse([]), FakeResponse({"id": "al1"}), FakeResponse({"assets": []}))
|
|
failures = rb.syncphase(self.conf(directory=room), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual(len(failures), 1)
|
|
self.assertIn("no file", failures.items[0][2])
|
|
|
|
def test_bad_key_is_fatal_before_writes(self):
|
|
rb.OPEN = FakeOpen(http_error(401))
|
|
with self.assertRaises(rb.BackupError):
|
|
rb.syncphase(self.conf(), rb.Immich("https://i/api", "k"))
|
|
|
|
def test_album_stage_failure_is_recorded(self):
|
|
rb.ATTEMPTS = 1
|
|
rb.OPEN = FakeOpen(
|
|
FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [], "nextPage": None}}),
|
|
FakeResponse({"id": "a1", "status": "created"}, status=201), # upload 111
|
|
http_error(500), # albums
|
|
FakeResponse({"id": "a1", "exifInfo": {}}), # re-read a1
|
|
FakeResponse(b"")) # update a1
|
|
failures = rb.syncphase(self.conf(limit=1), rb.Immich("https://i/api", "k"))
|
|
self.assertEqual(len(failures), 1)
|
|
reason = failures.items[0][2]
|
|
self.assertTrue("albums" in reason or "HTTP 500" in reason)
|
|
|
|
|
|
class Run(TestCase):
|
|
def conf(self, **over):
|
|
conf = {"directory": self.dir, "flickr_auth": self.write("f.yml", "key: FK\n"),
|
|
"immich_auth": self.write("i.yml", "url: https://i/api\nkey: IK\n"),
|
|
"index_url": "http://t/index", "fetch": True, "sync": True,
|
|
"dry_run": False, "report": False, "limit": None, "pace": 0, "quiet": False,
|
|
"attempts": 2, "backoff": 1, "timeout": 5, "user_agent": "ua", "debug": False}
|
|
conf.update(over)
|
|
return conf
|
|
|
|
def test_applysettings(self):
|
|
rb.applysettings(self.conf(pace=3, quiet=True, debug=True))
|
|
self.assertEqual(rb.PACE, 3)
|
|
self.assertTrue(rb.QUIET)
|
|
self.assertTrue(rb.DEBUG)
|
|
self.assertEqual(rb.ATTEMPTS, 2)
|
|
|
|
def test_fetch_only_never_reads_immich_config(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b"[]"))
|
|
conf = self.conf(sync=False, immich_auth=os.path.join(self.dir, "missing.yml"))
|
|
self.assertEqual(rb.run(conf), 0)
|
|
|
|
def test_sync_only_never_reads_flickr_config(self):
|
|
rb.OPEN = FakeOpen(FakeResponse({"id": "u1"}),
|
|
FakeResponse({"assets": {"items": [], "nextPage": None}}),
|
|
FakeResponse([]), FakeResponse({"id": "al1"}), FakeResponse({"assets": []}),
|
|
FakeResponse([]))
|
|
conf = self.conf(fetch=False, flickr_auth=os.path.join(self.dir, "missing.yml"))
|
|
self.assertEqual(rb.run(conf), 0)
|
|
|
|
def test_summary_counts_rewritten_sidecars(self):
|
|
rb.STATS["rewritten"] = 3
|
|
rb.summary(rb.Failures())
|
|
self.assertIn("rewritten 3", self.stdout.getvalue())
|
|
|
|
def test_failures_give_exit_1_and_are_summarised(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(json.dumps([index_entry("A,_E1", "junk")]).encode()))
|
|
self.assertEqual(rb.run(self.conf(sync=False)), 1)
|
|
self.assertIn("failed 1", self.stdout.getvalue())
|
|
self.assertIn("A, E1", self.stderr.getvalue())
|
|
|
|
def test_quiet_failures_are_printed_once_with_a_count(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(json.dumps([index_entry("A,_E1", "junk")]).encode()))
|
|
self.assertEqual(rb.run(self.conf(sync=False, quiet=True)), 1)
|
|
self.assertEqual(self.stdout.getvalue(), "")
|
|
self.assertEqual(self.stderr.getvalue().count("A, E1"), 1)
|
|
self.assertTrue(self.stderr.getvalue().rstrip("\n").endswith("1 failed"))
|
|
|
|
def test_config_error_gives_exit_2(self):
|
|
conf = self.conf(sync=False, flickr_auth=os.path.join(self.dir, "missing.yml"))
|
|
self.assertEqual(rb.run(conf), 2)
|
|
self.assertIn("key", self.stderr.getvalue())
|
|
|
|
def test_quiet_clean_run_prints_nothing(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b"[]"))
|
|
self.assertEqual(rb.run(self.conf(sync=False, quiet=True)), 0)
|
|
self.assertEqual(self.stdout.getvalue(), "")
|
|
self.assertEqual(self.stderr.getvalue(), "")
|
|
|
|
def test_main_wires_args(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b"[]"))
|
|
settingsfile = self.write("s.yml", "flickr_auth: %s\n" % self.write("f.yml", "key: FK\n"))
|
|
self.assertEqual(rb.main(["--settings", settingsfile, "--fetch-only", self.dir]), 0)
|
|
|
|
def test_main_reports_bad_settings(self):
|
|
self.assertEqual(rb.main(["--settings", os.path.join(self.dir, "none")]), 2)
|
|
self.assertIn("directory", self.stderr.getvalue())
|
|
|
|
def instead_of_run(self, boom):
|
|
original = rb.run
|
|
self.addCleanup(setattr, rb, "run", original)
|
|
def raiser(conf):
|
|
raise boom
|
|
rb.run = raiser
|
|
|
|
def test_unexpected_error_is_reported_not_traced(self):
|
|
self.instead_of_run(RuntimeError("something odd"))
|
|
self.assertEqual(rb.main(["--settings", os.path.join(self.dir, "none"), self.dir]), 1)
|
|
self.assertIn("something odd", self.stderr.getvalue())
|
|
|
|
def test_interrupt_exits_130(self):
|
|
self.instead_of_run(KeyboardInterrupt())
|
|
self.assertEqual(rb.main(["--settings", os.path.join(self.dir, "none"), self.dir]), 130)
|
|
|
|
def test_unreadable_settings_file_gives_exit_2(self):
|
|
self.assertEqual(rb.main(["--settings", self.dir, self.dir]), 2)
|
|
self.assertIn(self.dir, self.stderr.getvalue())
|
|
|
|
|
|
class FailureRecords(TestCase):
|
|
def failures(self, *items):
|
|
f = rb.Failures()
|
|
for key, names, reason in items:
|
|
f.add(key, [{"name": n} for n in names], reason)
|
|
return f
|
|
|
|
def test_missing_file_is_empty(self):
|
|
self.assertEqual(rb.loadfailures(self.dir), {})
|
|
|
|
def test_records_merge_and_drop_completed(self):
|
|
rb.recordfailures(self.dir, self.failures(("111", ["A, E1"], "gone"),
|
|
("222", ["B, E2"], "HTTP 404")))
|
|
held = rb.loadfailures(self.dir)
|
|
self.assertEqual(sorted(held), ["111", "222"])
|
|
self.assertEqual(held["111"]["reason"], "gone")
|
|
self.assertEqual(held["111"]["pages"], ["A, E1"])
|
|
self.assertRegex(held["111"]["when"], r"^\d{4}-")
|
|
# 111 later succeeds; 333 newly fails; 222 untouched this run but still missing
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
|
|
self.write("111.jpg", b"x")
|
|
rb.recordfailures(self.dir, self.failures(("333", ["C, E3"], "timeout")))
|
|
held = rb.loadfailures(self.dir)
|
|
self.assertEqual(sorted(held), ["222", "333"])
|
|
|
|
def test_unreadable_file_is_treated_as_empty(self):
|
|
self.write(rb.FAILURES, "{not json")
|
|
self.assertEqual(rb.loadfailures(self.dir), {})
|
|
|
|
|
|
class Report(TestCase):
|
|
def conf(self, **over):
|
|
conf = {"directory": self.dir, "index_url": "http://t/index", "report": True,
|
|
"fetch": True, "sync": True, "dry_run": False, "limit": None, "quiet": False,
|
|
"flickr_auth": os.path.join(self.dir, "missing.yml"),
|
|
"immich_auth": os.path.join(self.dir, "missing.yml"),
|
|
"pace": 0, "attempts": 2, "backoff": 1, "timeout": 5, "user_agent": "ua",
|
|
"debug": False}
|
|
conf.update(over)
|
|
return conf
|
|
|
|
def index(self):
|
|
return json.dumps([
|
|
index_entry("A,_E1", "https://farm1.static.flickr.com/1/111_a_m.jpg",
|
|
page="https://www.flickr.com/photos/x/111/"),
|
|
index_entry("B,_E2", "https://farm1.static.flickr.com/1/222_b_m.jpg",
|
|
page="https://www.flickr.com/photos/x/222/"),
|
|
index_entry("C,_E3", "https://farm1.static.flickr.com/1/333_c_m.jpg"),
|
|
index_entry("D,_E4", "junk"),
|
|
]).encode()
|
|
|
|
def test_lists_missing_photos_with_pages_and_reasons(self):
|
|
rb.writejson(rb.sidecarpath(self.dir, "111"), sidecar())
|
|
self.write("111.jpg", b"x")
|
|
rb.writejson(os.path.join(self.dir, rb.FAILURES),
|
|
{"222": {"reason": "photo 222: HTTP 404 Not Found", "pages": ["B, E2"],
|
|
"when": "2026-09-13T10:00:00Z"}})
|
|
rb.OPEN = FakeOpen(FakeResponse(self.index()))
|
|
self.assertEqual(rb.run(self.conf()), 0)
|
|
out = self.stdout.getvalue()
|
|
self.assertNotIn("111", out)
|
|
self.assertIn("222: photo 222: HTTP 404 Not Found (2026-09-13)", out)
|
|
self.assertIn(" B, E2 https://london.randomness.org.uk/wiki.cgi?B%2C_E2", out)
|
|
self.assertIn("333: not yet attempted", out)
|
|
self.assertIn(" C, E3 ", out)
|
|
self.assertIn("D,_E4: node_image is not a url", out)
|
|
self.assertIn("2 of 3 photos missing, 1 node with an unusable image", out)
|
|
self.assertEqual(len(rb.OPEN.requests), 1) # the index only
|
|
|
|
def test_report_prints_even_when_quiet(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(self.index()))
|
|
rb.run(self.conf(quiet=True))
|
|
self.assertIn("3 of 3 photos missing", self.stdout.getvalue())
|
|
|
|
def test_report_needs_no_credentials(self):
|
|
rb.OPEN = FakeOpen(FakeResponse(b"[]"))
|
|
self.assertEqual(rb.run(self.conf()), 0)
|
|
self.assertIn("0 of 0 photos missing", self.stdout.getvalue())
|
|
|
|
|
|
class RunRecordsFailures(TestCase):
|
|
def conf(self, **over):
|
|
conf = {"directory": self.dir, "flickr_auth": self.write("f.yml", "key: FK\n"),
|
|
"immich_auth": self.write("i.yml", "url: https://i/api\nkey: IK\n"),
|
|
"index_url": "http://t/index", "fetch": True, "sync": False, "report": False,
|
|
"dry_run": False, "limit": None, "pace": 0, "quiet": False,
|
|
"attempts": 2, "backoff": 1, "timeout": 5, "user_agent": "ua", "debug": False}
|
|
conf.update(over)
|
|
return conf
|
|
|
|
def test_failed_run_writes_the_failures_file(self):
|
|
index = [index_entry("A,_E1", "https://farm1.static.flickr.com/1/111_a_m.jpg")]
|
|
rb.OPEN = FakeOpen(FakeResponse(json.dumps(index).encode()),
|
|
FakeResponse({"stat": "fail", "code": 1, "message": "Photo not found"}))
|
|
self.assertEqual(rb.run(self.conf()), 1)
|
|
held = rb.loadfailures(self.dir)
|
|
self.assertIn("111", held)
|
|
self.assertIn("Photo not found", held["111"]["reason"])
|
|
self.assertEqual(held["111"]["pages"], ["A, E1"])
|
|
|
|
def test_dry_run_writes_nothing(self):
|
|
index = [index_entry("A,_E1", "https://farm1.static.flickr.com/1/111_a_m.jpg")]
|
|
rb.OPEN = FakeOpen(FakeResponse(json.dumps(index).encode()))
|
|
rb.run(self.conf(dry_run=True))
|
|
self.assertFalse(os.path.exists(os.path.join(self.dir, rb.FAILURES)))
|