Scaffold rgl-photo-backup with output helpers and test harness
Start the script that will back up the Randomness Guide to London's photos to disk and Immich. This commit holds the constants, error classes and quiet-aware output helpers every later section relies on, plus the test loader for an extensionless script. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
4 files changed
+258
No files matched your search
@@ -0,0 +1,134 @@
|
||||
import contextlib
|
||||
import importlib.machinery
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
|
||||
_here = os.path.dirname(os.path.abspath(__file__))
|
||||
_path = os.path.join(_here, "rgl-photo-backup")
|
||||
_spec = importlib.util.spec_from_file_location(
|
||||
"rgl_photo_backup", _path,
|
||||
loader=importlib.machinery.SourceFileLoader("rgl_photo_backup", _path))
|
||||
rb = importlib.util.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(rb)
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
"""What urlopen would return: a body, a status and headers."""
|
||||
|
||||
def __init__(self, body=b"", status=200, headers=None):
|
||||
self.body = body if isinstance(body, bytes) else json.dumps(body).encode()
|
||||
self.status = status
|
||||
self.headers = headers or {}
|
||||
self._pos = 0
|
||||
|
||||
def read(self, size=-1):
|
||||
if size is None or size < 0:
|
||||
data, self._pos = self.body[self._pos:], len(self.body)
|
||||
else:
|
||||
data = self.body[self._pos:self._pos + size]
|
||||
self._pos += len(data)
|
||||
return data
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc):
|
||||
pass
|
||||
|
||||
|
||||
def http_error(code, body=b"", headers=None):
|
||||
return urllib.error.HTTPError("http://test", code, "reason %d" % code,
|
||||
headers or {}, io.BytesIO(body))
|
||||
|
||||
|
||||
class FakeOpen:
|
||||
"""Replaces rb.OPEN. Serves scripted responses in order and records
|
||||
every request so a test can assert on method, url, headers and body."""
|
||||
|
||||
def __init__(self, *responses):
|
||||
self.responses = list(responses)
|
||||
self.requests = []
|
||||
|
||||
def __call__(self, request, timeout=None):
|
||||
self.requests.append(request)
|
||||
if not self.responses:
|
||||
raise AssertionError("unexpected request: %s %s"
|
||||
% (request.get_method(), request.full_url))
|
||||
answer = self.responses.pop(0)
|
||||
if isinstance(answer, Exception):
|
||||
raise answer
|
||||
return answer
|
||||
|
||||
|
||||
class TestCase(unittest.TestCase):
|
||||
"""Silences output and restores the module globals tests move around."""
|
||||
|
||||
GLOBALS = ("OPEN", "SLEEP", "ATTEMPTS", "BACKOFF_BASE", "PACE", "QUIET",
|
||||
"DEBUG", "HTTP_TIMEOUT", "USER_AGENT", "STATS", "DOWNLOAD_CAP")
|
||||
|
||||
def setUp(self):
|
||||
self._output = contextlib.ExitStack()
|
||||
if not os.environ.get("RGL_PHOTO_BACKUP_TEST_VERBOSE"):
|
||||
self.stdout = self._output.enter_context(
|
||||
contextlib.redirect_stdout(io.StringIO()))
|
||||
self.stderr = self._output.enter_context(
|
||||
contextlib.redirect_stderr(io.StringIO()))
|
||||
self.addCleanup(self._output.close)
|
||||
self._saved = {k: getattr(rb, k) for k in self.GLOBALS}
|
||||
rb.SLEEP = lambda seconds: None
|
||||
rb.PACE = 0.0
|
||||
rb.STATS = dict.fromkeys(rb.STATS, 0)
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.dir = self.tmp.name
|
||||
|
||||
def tearDown(self):
|
||||
for k, v in self._saved.items():
|
||||
setattr(rb, k, v)
|
||||
|
||||
def write(self, name, content):
|
||||
path = os.path.join(self.dir, name)
|
||||
mode = "wb" if isinstance(content, bytes) else "w"
|
||||
with open(path, mode) as fh:
|
||||
fh.write(content)
|
||||
return path
|
||||
|
||||
|
||||
class Output(TestCase):
|
||||
def test_say_prints_unless_quiet(self):
|
||||
rb.QUIET = False
|
||||
rb.say("hello")
|
||||
self.assertEqual(self.stdout.getvalue(), "hello\n")
|
||||
rb.QUIET = True
|
||||
rb.say("again")
|
||||
self.assertEqual(self.stdout.getvalue(), "hello\n")
|
||||
|
||||
def test_warn_always_reaches_stderr(self):
|
||||
rb.QUIET = True
|
||||
rb.warn("careful")
|
||||
self.assertEqual(self.stderr.getvalue(), "careful\n")
|
||||
|
||||
def test_debug_only_when_enabled(self):
|
||||
rb.DEBUG = False
|
||||
rb.debug("x")
|
||||
self.assertEqual(self.stderr.getvalue(), "")
|
||||
rb.DEBUG = True
|
||||
rb.debug("y")
|
||||
self.assertEqual(self.stderr.getvalue(), "debug: y\n")
|
||||
|
||||
def test_errors_share_a_base(self):
|
||||
self.assertTrue(issubclass(rb.PhotoGone, rb.BackupError))
|
||||
e = rb.HTTPFailure("ctx: HTTP 404", code=404)
|
||||
self.assertEqual(e.code, 404)
|
||||
self.assertIsInstance(e, rb.BackupError)
|
||||
Reference in new issue
Block a user