bobandClaude Fable 5.1
5549217db5
Walk the grouped index, ask flickr for each new photo, download the best size that actually serves, and write the sidecar only once the file is renamed into place. Photos already complete cost no flickr calls; a change in which pages use one rewrites just the sidecar. Every failure is recorded with the pages affected and the run carries on. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
935 lines
43 KiB
Python
935 lines
43 KiB
Python
import contextlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import urllib.error
|
|
import urllib.parse
|
|
|
|
_here = os.path.dirname(os.path.abspath(__file__))
|
|
_path = os.path.join(_here, "rgl-photo-backup")
|
|
_spec = importlib.util.spec_from_file_location(
|
|
"rgl_photo_backup", _path,
|
|
loader=importlib.machinery.SourceFileLoader("rgl_photo_backup", _path))
|
|
rb = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(rb)
|
|
|
|
|
|
class FakeResponse:
|
|
"""What urlopen would return: a body, a status and headers."""
|
|
|
|
def __init__(self, body=b"", status=200, headers=None):
|
|
self.body = body if isinstance(body, bytes) else json.dumps(body).encode()
|
|
self.status = status
|
|
self.headers = headers or {}
|
|
self._pos = 0
|
|
|
|
def read(self, size=-1):
|
|
if size is None or size < 0:
|
|
data, self._pos = self.body[self._pos:], len(self.body)
|
|
else:
|
|
data = self.body[self._pos:self._pos + size]
|
|
self._pos += len(data)
|
|
return data
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
pass
|
|
|
|
|
|
def http_error(code, body=b"", headers=None):
|
|
return urllib.error.HTTPError("http://test", code, "reason %d" % code,
|
|
headers or {}, io.BytesIO(body))
|
|
|
|
|
|
class FakeOpen:
|
|
"""Replaces rb.OPEN. Serves scripted responses in order and records
|
|
every request so a test can assert on method, url, headers and body."""
|
|
|
|
def __init__(self, *responses):
|
|
self.responses = list(responses)
|
|
self.requests = []
|
|
|
|
def __call__(self, request, timeout=None):
|
|
self.requests.append(request)
|
|
if not self.responses:
|
|
raise AssertionError("unexpected request: %s %s"
|
|
% (request.get_method(), request.full_url))
|
|
answer = self.responses.pop(0)
|
|
if isinstance(answer, Exception):
|
|
raise answer
|
|
return answer
|
|
|
|
|
|
class TestCase(unittest.TestCase):
|
|
"""Silences output and restores the module globals tests move around."""
|
|
|
|
GLOBALS = ("OPEN", "SLEEP", "ATTEMPTS", "BACKOFF_BASE", "PACE", "QUIET",
|
|
"DEBUG", "HTTP_TIMEOUT", "USER_AGENT", "STATS", "DOWNLOAD_CAP")
|
|
|
|
def setUp(self):
|
|
self._output = contextlib.ExitStack()
|
|
if not os.environ.get("RGL_PHOTO_BACKUP_TEST_VERBOSE"):
|
|
self.stdout = self._output.enter_context(
|
|
contextlib.redirect_stdout(io.StringIO()))
|
|
self.stderr = self._output.enter_context(
|
|
contextlib.redirect_stderr(io.StringIO()))
|
|
self.addCleanup(self._output.close)
|
|
self._saved = {k: getattr(rb, k) for k in self.GLOBALS}
|
|
rb.SLEEP = lambda seconds: None
|
|
rb.PACE = 0.0
|
|
rb.STATS = dict.fromkeys(rb.STATS, 0)
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.dir = self.tmp.name
|
|
|
|
def tearDown(self):
|
|
for k, v in self._saved.items():
|
|
setattr(rb, k, v)
|
|
|
|
def write(self, name, content):
|
|
path = os.path.join(self.dir, name)
|
|
mode = "wb" if isinstance(content, bytes) else "w"
|
|
with open(path, mode) as fh:
|
|
fh.write(content)
|
|
return path
|
|
|
|
|
|
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_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), {})
|
|
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")
|