bob/rgl-photo-backup
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2774267425af56e2a83e931ecb3c873a422b99d6
rgl-photo-backup/test_rgl_photo_backup.py
T
bobandClaude Fable 5.1 2774267425 Build the immich description block and tags from a sidecar 
Immich has no copyright or source fields, so the photographer, licence,
flickr page and every wiki page using the photo go into a marked block in
the description. The marker lets a later run find and replace only that
block, leaving anything written by hand alone. Tags mirror licence and
photographer for filtering.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-10 12:07:45 +01:00

739 lines
33 KiB
Python
Raw Blame History

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
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"], 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"))
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_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_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), {})
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))
Reference in new issue View Git Blame Copy Permalink