鲨鱼辣椒
694c4317a3
Complete the selective V2 checkpoint with its minimal AgentBus, object-storage, replay persistence, and validated-workbench shared dependency closure.
99 lines
4.3 KiB
Python
99 lines
4.3 KiB
Python
"""Contract, safety and determinism tests for the field Recovery Skill package."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from pathlib import Path
|
|
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
|
|
PACKAGER_PATH = REPOSITORY_ROOT / "scripts" / "package_fixed_channel_field_recovery_skill.py"
|
|
SKILL_SOURCE = REPOSITORY_ROOT / "agent-skills" / "fixed-channel-field-recovery"
|
|
|
|
|
|
def load_packager():
|
|
spec = importlib.util.spec_from_file_location("field_recovery_skill_packager", PACKAGER_PATH)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError("Field Recovery Skill packager module cannot be loaded")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class FixedChannelFieldRecoverySkillPackageTest(unittest.TestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.packager = load_packager()
|
|
|
|
def test_source_is_complete_and_contains_no_external_payload(self) -> None:
|
|
files = self.packager.validate_source(SKILL_SOURCE)
|
|
|
|
self.assertEqual(6, len(files))
|
|
self.assertNotIn("rate-room", "\n".join(path.name for path in files).lower())
|
|
|
|
def test_archive_and_manifest_are_deterministic_and_internal_only(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="field-recovery-skill-test-") as temporary_directory:
|
|
temporary_root = Path(temporary_directory)
|
|
first = temporary_root / "first" / "fixed-channel-field-recovery.skill"
|
|
second = temporary_root / "second" / "fixed-channel-field-recovery.skill"
|
|
first_members = self.packager.write_archive(SKILL_SOURCE, first)
|
|
second_members = self.packager.write_archive(SKILL_SOURCE, second)
|
|
|
|
self.assertEqual(first.read_bytes(), second.read_bytes())
|
|
self.assertEqual(first_members, second_members)
|
|
manifest = self.packager.build_manifest(SKILL_SOURCE, first, first_members)
|
|
self.assertEqual("INTERNAL_ONLY", manifest["publication_status"])
|
|
self.assertEqual(6, manifest["member_count"])
|
|
self.assertNotIn("rate_room_directory", manifest)
|
|
|
|
def test_validation_rejects_placeholder_or_email_leak(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="field-recovery-skill-invalid-") as temporary_directory:
|
|
copied = Path(temporary_directory) / "skill"
|
|
shutil.copytree(SKILL_SOURCE, copied)
|
|
skill_path = copied / "SKILL.md"
|
|
skill_path.write_text(skill_path.read_text(encoding="utf-8") + "\nTODO real@example.com\n", encoding="utf-8")
|
|
|
|
with self.assertRaisesRegex(ValueError, "unfinished placeholder"):
|
|
self.packager.validate_source(copied)
|
|
|
|
def test_archive_validation_rejects_unexpected_member(self) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="field-recovery-skill-tamper-") as temporary_directory:
|
|
archive_path = Path(temporary_directory) / "fixed-channel-field-recovery.skill"
|
|
self.packager.write_archive(SKILL_SOURCE, archive_path)
|
|
with zipfile.ZipFile(archive_path, "a") as archive:
|
|
archive.writestr("fixed-channel-field-recovery/extra.txt", "unexpected")
|
|
|
|
with self.assertRaisesRegex(ValueError, "member list"):
|
|
self.packager.validate_archive(SKILL_SOURCE, archive_path)
|
|
|
|
def test_schema_keeps_java_value_constraints_and_patch_statuses_in_sync(self) -> None:
|
|
request_schema = json.loads(
|
|
(SKILL_SOURCE / "references" / "recovery-request-set-v1.schema.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
enum_values = request_schema["$defs"]["constraints"]["properties"]["enum_values"]
|
|
self.assertEqual("array", enum_values["type"])
|
|
self.assertNotIn("minItems", enum_values)
|
|
|
|
patch_schema = json.loads(
|
|
(SKILL_SOURCE / "references" / "recovery-patch-set-v1.schema.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
status_branches = {
|
|
branch["if"]["properties"]["patch_status"]["const"]
|
|
for branch in patch_schema["$defs"]["patch"]["allOf"]
|
|
}
|
|
self.assertEqual({"PROPOSED", "NO_CANDIDATE", "AMBIGUOUS", "FAILED"}, status_branches)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|