161 lines
8.1 KiB
Python
161 lines
8.1 KiB
Python
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")
|