Wire the phases into a command with a summary and exit status

Load only the credentials the requested phases need, run fetch then sync,
and exit 1 when any photo failed so cron mails. A quiet clean run prints
nothing; a quiet failing run prints the failure list at the end.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
bobandClaude Fable 5.1 committed 2026-09-10 12:07:46 +01:00
1 parent 534d68aac0
commit 4e0bafe497
2 files changed
+159 -1

No files matched your search

+89
View File
@@ -1364,3 +1364,92 @@ class SyncPhase(TestCase):
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, "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())