import contextlib import io import json import pathlib import shutil import subprocess import sys import tempfile import unittest from datetime import datetime, timedelta, timezone from unittest import mock from scripts import skill_update class DailyUpdateTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = pathlib.Path(self.temp.name) self.now = datetime(2026, 9, 30, 9, tzinfo=timezone(timedelta(hours=8))) def state(self, root=None): return json.loads(((root or self.root) / "scripts/update-state.json").read_text(encoding="utf-8")) def git(self, root, *args): result = subprocess.run( ["git", "-C", str(root), *args], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) return result.stdout.decode().strip() def commit(self, root, message): self.git(root, "add", "--all") self.git(root, "-c", "user.name=Update Test", "-c", "user.email=update@example.test", "commit", "-m", message) def repository(self): remote, source, installed = [self.root / name for name in ("remote.git", "source", "installed")] self.git(self.root, "init", "--bare", str(remote)) self.git(self.root, "init", "-b", "main", str(source)) (source / "config.json").write_bytes(b'{"local":"keep"}\n') (source / "VERSION").write_text("1.0.0\n") (source / ".gitignore").write_text("scripts/update-state*\nscripts/config.json\n") self.commit(source, "initial") self.git(source, "remote", "add", "origin", str(remote)) self.git(source, "push", "origin", "main") self.git(self.root, "clone", "-b", "main", str(remote), str(installed)) (installed / "scripts").mkdir() (installed / "scripts/config.json").write_bytes(b'{"access_token":"local-only"}') return source, installed def publish(self, source, config=False): (source / "VERSION").write_text("1.0.1\n") if config: (source / "config.json").write_bytes(b'{"new":"template"}\n') self.commit(source, "update") self.git(source, "push", "origin", "main") def test_once_per_local_day_and_every_usage_timestamp(self): with mock.patch.object(skill_update, "_check_remote", return_value={"status": "up_to_date"}) as check: skill_update.on_use(self.root, self.now) skill_update.on_use(self.root, self.now + timedelta(hours=2)) self.assertEqual(check.call_count, 1) self.assertEqual(self.state()["last_used_at"], "2026-09-30T11:00:00+08:00") skill_update.on_use(self.root, self.now + timedelta(days=1)) self.assertEqual(check.call_count, 2) def test_failed_check_does_not_block_or_repeat_today_or_leak_output(self): with mock.patch.object(skill_update, "_check_remote", side_effect=RuntimeError("secret-url")) as check: output, error = io.StringIO(), io.StringIO() with contextlib.redirect_stdout(output), contextlib.redirect_stderr(error): self.assertFalse(skill_update.on_use(self.root, self.now)) skill_update.on_use(self.root, self.now) self.assertEqual(check.call_count, 1) self.assertEqual(output.getvalue(), "") self.assertNotIn("secret-url", error.getvalue()) self.assertNotIn("secret-url", json.dumps(self.state())) self.assertEqual(self.state()["status"], "check_failed") def test_fast_forward_keeps_modified_local_login_configuration(self): source, installed = self.repository() (installed / "config.json").write_bytes(b'{"local":"modified"}') original = (installed / "config.json").read_bytes() self.publish(source) self.assertTrue(skill_update.on_use(installed, self.now)) self.assertEqual(self.git(installed, "rev-parse", "HEAD"), self.git(source, "rev-parse", "HEAD")) self.assertEqual((installed / "config.json").read_bytes(), original) self.assertEqual((installed / "scripts/config.json").read_bytes(), b'{"access_token":"local-only"}') self.assertEqual(self.state(installed)["status"], "updated") def test_clean_config_preserved_when_upstream_template_changes(self): source, installed = self.repository() original = (installed / "config.json").read_bytes() self.publish(source, config=True) self.assertTrue(skill_update.on_use(installed, self.now)) self.assertEqual((installed / "config.json").read_bytes(), original) self.assertEqual((installed / "VERSION").read_text(), "1.0.1\n") def test_source_edits_block_install_but_do_not_discard_changes(self): source, installed = self.repository() self.publish(source) (installed / "VERSION").write_text("local edit\n") head = self.git(installed, "rev-parse", "HEAD") self.assertFalse(skill_update.on_use(installed, self.now)) self.assertEqual(self.git(installed, "rev-parse", "HEAD"), head) self.assertEqual((installed / "VERSION").read_text(), "local edit\n") self.assertEqual(self.state(installed)["status"], "update_blocked") def test_conflicting_remote_config_keeps_dirty_local_config(self): source, installed = self.repository() (installed / "config.json").write_bytes(b'{"local":"changed"}') self.publish(source, config=True) head = self.git(installed, "rev-parse", "HEAD") self.assertFalse(skill_update.on_use(installed, self.now)) self.assertEqual(self.git(installed, "rev-parse", "HEAD"), head) self.assertEqual((installed / "config.json").read_bytes(), b'{"local":"changed"}') self.assertEqual(self.state(installed)["status"], "update_blocked") def test_cli_restarts_into_new_version_and_keeps_stdout_json(self): source, installed = self.repository() (source / "scripts").mkdir() scripts = pathlib.Path(skill_update.__file__).parent for name in ("oracle_skill.py", "skill_update.py", "runtime_state.py"): shutil.copyfile(str(scripts / name), str(source / "scripts" / name)) self.commit(source, "add cli") self.git(source, "push", "origin", "main") self.git(installed, "pull", "--ff-only", "origin", "main") self.publish(source) result = subprocess.run( [sys.executable, str(installed / "scripts/oracle_skill.py"), "capabilities", "--json"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True, timeout=30, ) self.assertEqual(json.loads(result.stdout.decode("utf-8"))["version"], "1.0.1") self.assertEqual(self.state(installed)["status"], "updated") self.assertEqual(self.git(installed, "rev-parse", "HEAD"), self.git(source, "rev-parse", "HEAD")) def test_diverged_branch_never_resets_local_commit(self): source, installed = self.repository() (installed / "VERSION").write_text("local branch\n") self.commit(installed, "local") self.publish(source) head = self.git(installed, "rev-parse", "HEAD") self.assertFalse(skill_update.on_use(installed, self.now)) self.assertEqual(self.git(installed, "rev-parse", "HEAD"), head) self.assertEqual(self.state(installed)["status"], "update_blocked") def test_download_without_git_does_not_fetch_parent_repository(self): with mock.patch.object(skill_update, "_git") as git: self.assertFalse(skill_update.on_use(self.root, self.now)) git.assert_not_called() self.assertEqual(self.state()["status"], "not_git") def test_concurrent_invocation_does_not_fetch_twice_or_lose_latest_usage(self): def checking(root): skill_update.on_use(root, self.now + timedelta(minutes=1)) return {"status": "up_to_date"} with mock.patch.object(skill_update, "_check_remote", side_effect=checking) as check: skill_update.on_use(self.root, self.now) self.assertEqual(check.call_count, 1) self.assertEqual(self.state()["last_used_at"], "2026-09-30T09:01:00+08:00")