M9.2-M9.5: Add improvement pipeline, community workflow, transitive integrity, and supply chain attestation
This commit is contained in:
@@ -0,0 +1,45 @@
|
||||
"""Tests for M9.5 supply chain attestation."""
|
||||
|
||||
from cmdforge.attestation import (
|
||||
Attestation,
|
||||
sign_tool,
|
||||
verify_attestation,
|
||||
verify_content_hash,
|
||||
)
|
||||
from cmdforge.tool import Tool
|
||||
|
||||
|
||||
class TestSignTool:
|
||||
def test_creates_attestation(self):
|
||||
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
|
||||
assert att.tool_name == "mytool"
|
||||
assert att.version == "1.0.0"
|
||||
assert att.content_hash == "abc123"
|
||||
assert att.signer == "alice"
|
||||
assert len(att.signature) == 64 # SHA256 hex
|
||||
assert att.algorithm == "hmac-sha256"
|
||||
|
||||
def test_different_keys_different_signatures(self):
|
||||
att1 = sign_tool("tool", "1.0.0", "hash", "alice", "key1")
|
||||
att2 = sign_tool("tool", "1.0.0", "hash", "alice", "key2")
|
||||
assert att1.signature != att2.signature
|
||||
|
||||
|
||||
class TestVerifyAttestation:
|
||||
def test_valid_signature(self):
|
||||
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
|
||||
assert verify_attestation(att, "secret-key")
|
||||
|
||||
def test_wrong_key_fails(self):
|
||||
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
|
||||
assert not verify_attestation(att, "wrong-key")
|
||||
|
||||
def test_tampered_content_fails(self):
|
||||
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
|
||||
att.content_hash = "tampered"
|
||||
assert not verify_attestation(att, "secret-key")
|
||||
|
||||
def test_tampered_signer_fails(self):
|
||||
att = sign_tool("mytool", "1.0.0", "abc123", "alice", "secret-key")
|
||||
att.signer = "eve"
|
||||
assert not verify_attestation(att, "secret-key")
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Tests for M9.3 community improvement workflow."""
|
||||
|
||||
from cmdforge.community import (
|
||||
ImprovementSubmission,
|
||||
create_submission,
|
||||
run_submission_tests,
|
||||
review_submission,
|
||||
)
|
||||
from cmdforge.tool import Tool, PromptStep, CodeStep
|
||||
|
||||
|
||||
class TestCreateSubmission:
|
||||
def test_create_for_prompt_step(self):
|
||||
tool = Tool(
|
||||
name="test",
|
||||
version="1.0.0",
|
||||
steps=[PromptStep(prompt="Summarize", provider="mock", output_var="x")],
|
||||
output="{x}",
|
||||
)
|
||||
sub = create_submission(tool, 0, "Summarize the text thoroughly", "alice", "More detail")
|
||||
assert sub.tool_name == "test"
|
||||
assert sub.step_type == "prompt"
|
||||
assert sub.original == "Summarize"
|
||||
assert sub.proposed == "Summarize the text thoroughly"
|
||||
assert sub.submitter == "alice"
|
||||
assert sub.status == "pending"
|
||||
|
||||
def test_create_for_code_step(self):
|
||||
tool = Tool(
|
||||
name="test",
|
||||
steps=[CodeStep(code="x = 1", output_var="x")],
|
||||
output="{x}",
|
||||
)
|
||||
sub = create_submission(tool, 0, "x = 42", "bob")
|
||||
assert sub.step_type == "code"
|
||||
assert sub.original == "x = 1"
|
||||
|
||||
def test_invalid_step_index(self):
|
||||
tool = Tool(name="test")
|
||||
try:
|
||||
create_submission(tool, 5, "new", "alice")
|
||||
assert False, "Should have raised"
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
|
||||
class TestReviewSubmission:
|
||||
def test_review_approved(self):
|
||||
sub = ImprovementSubmission(
|
||||
tool_name="test", tool_version="1.0.0", submitter="alice",
|
||||
step_index=0, step_type="prompt", original="a", proposed="b",
|
||||
status="tested",
|
||||
)
|
||||
review = review_submission(sub, "approve", "admin", "Good improvement")
|
||||
assert review.decision == "approve"
|
||||
assert sub.status == "approve"
|
||||
|
||||
def test_review_pending_rejected(self):
|
||||
sub = ImprovementSubmission(
|
||||
tool_name="test", tool_version="1.0.0", submitter="alice",
|
||||
step_index=0, step_type="prompt", original="a", proposed="b",
|
||||
status="pending",
|
||||
)
|
||||
try:
|
||||
review_submission(sub, "approve", "admin")
|
||||
assert False, "Should have raised"
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
def test_invalid_decision(self):
|
||||
sub = ImprovementSubmission(
|
||||
tool_name="test", tool_version="1.0.0", submitter="alice",
|
||||
step_index=0, step_type="prompt", original="a", proposed="b",
|
||||
status="tested",
|
||||
)
|
||||
try:
|
||||
review_submission(sub, "maybe", "admin")
|
||||
assert False, "Should have raised"
|
||||
except ValueError:
|
||||
pass
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Tests for scrutiny-driven improvement pipeline (M9.2)."""
|
||||
|
||||
from cmdforge.improvement import (
|
||||
ImprovementSuggestion,
|
||||
ImprovementReport,
|
||||
generate_improvements,
|
||||
)
|
||||
from cmdforge.tool import Tool, PromptStep, CodeStep
|
||||
|
||||
|
||||
class TestImprovementReport:
|
||||
def test_empty_report(self):
|
||||
report = ImprovementReport(tool_name="test")
|
||||
assert not report.has_high_severity
|
||||
assert report.by_category == {}
|
||||
|
||||
def test_has_high_severity(self):
|
||||
report = ImprovementReport(
|
||||
tool_name="test",
|
||||
suggestions=[
|
||||
ImprovementSuggestion("honesty", "high", "Bad", "desc"),
|
||||
ImprovementSuggestion("efficiency", "low", "Minor", "desc"),
|
||||
],
|
||||
)
|
||||
assert report.has_high_severity
|
||||
|
||||
def test_by_category(self):
|
||||
report = ImprovementReport(
|
||||
tool_name="test",
|
||||
suggestions=[
|
||||
ImprovementSuggestion("honesty", "high", "A", "d"),
|
||||
ImprovementSuggestion("honesty", "low", "B", "d"),
|
||||
ImprovementSuggestion("efficiency", "medium", "C", "d"),
|
||||
],
|
||||
)
|
||||
assert len(report.by_category["honesty"]) == 2
|
||||
assert len(report.by_category["efficiency"]) == 1
|
||||
|
||||
|
||||
class TestGenerateImprovements:
|
||||
def test_missing_description(self):
|
||||
tool = Tool(name="nodesc")
|
||||
report = generate_improvements(tool)
|
||||
honesty = [s for s in report.suggestions if s.category == "honesty"]
|
||||
assert any("Missing description" in s.title for s in honesty)
|
||||
|
||||
def test_short_description(self):
|
||||
tool = Tool(name="short", description="Does stuff")
|
||||
report = generate_improvements(tool)
|
||||
honesty = [s for s in report.suggestions if s.category == "honesty"]
|
||||
assert any("short" in s.title.lower() for s in honesty)
|
||||
|
||||
def test_echo_prompt_detected(self):
|
||||
tool = Tool(
|
||||
name="echo",
|
||||
steps=[PromptStep(prompt="{input}", provider="mock", output_var="out")],
|
||||
output="{out}",
|
||||
)
|
||||
report = generate_improvements(tool)
|
||||
efficiency = [s for s in report.suggestions if s.category == "efficiency"]
|
||||
assert any("echo" in s.title.lower() for s in efficiency)
|
||||
|
||||
def test_multiple_same_provider_prompts(self):
|
||||
tool = Tool(
|
||||
name="multi",
|
||||
steps=[
|
||||
PromptStep(prompt="Step 1: {input}", provider="claude", output_var="a"),
|
||||
PromptStep(prompt="Step 2: {a}", provider="claude", output_var="b"),
|
||||
],
|
||||
output="{b}",
|
||||
)
|
||||
report = generate_improvements(tool)
|
||||
efficiency = [s for s in report.suggestions if s.category == "efficiency"]
|
||||
assert any("same provider" in s.title.lower() for s in efficiency)
|
||||
|
||||
def test_exec_detected(self):
|
||||
tool = Tool(
|
||||
name="dangerous",
|
||||
steps=[CodeStep(code="exec('print(1)')", output_var="x")],
|
||||
output="{x}",
|
||||
)
|
||||
report = generate_improvements(tool)
|
||||
transparency = [s for s in report.suggestions if s.category == "transparency"]
|
||||
assert any("exec" in s.title.lower() for s in transparency)
|
||||
|
||||
def test_short_prompt_detected(self):
|
||||
tool = Tool(
|
||||
name="vague",
|
||||
steps=[PromptStep(prompt="Do it", provider="mock", output_var="x")],
|
||||
output="{x}",
|
||||
)
|
||||
report = generate_improvements(tool)
|
||||
transparency = [s for s in report.suggestions if s.category == "transparency"]
|
||||
assert any("short" in s.title.lower() for s in transparency)
|
||||
|
||||
def test_clean_tool_no_suggestions(self):
|
||||
tool = Tool(
|
||||
name="clean",
|
||||
description="A well-documented tool that summarizes text input concisely",
|
||||
steps=[PromptStep(
|
||||
prompt="Summarize the following text in 2-3 sentences. Focus on key points.\n\nText: {input}",
|
||||
provider="mock",
|
||||
output_var="summary",
|
||||
)],
|
||||
output="{summary}",
|
||||
)
|
||||
report = generate_improvements(tool)
|
||||
high = [s for s in report.suggestions if s.severity == "high"]
|
||||
assert len(high) == 0
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Tests for M9.4 transitive integrity verification."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from cmdforge.integrity import (
|
||||
IntegrityNode,
|
||||
IntegrityChain,
|
||||
compute_tool_hash,
|
||||
build_integrity_chain,
|
||||
verify_integrity,
|
||||
)
|
||||
from cmdforge.tool import Tool, PromptStep, ToolStep
|
||||
|
||||
|
||||
class TestComputeToolHash:
|
||||
def test_same_tool_same_hash(self):
|
||||
tool = Tool(name="test", version="1.0.0")
|
||||
assert compute_tool_hash(tool) == compute_tool_hash(tool)
|
||||
|
||||
def test_different_tools_different_hash(self):
|
||||
a = Tool(name="a", version="1.0.0")
|
||||
b = Tool(name="b", version="1.0.0")
|
||||
assert compute_tool_hash(a) != compute_tool_hash(b)
|
||||
|
||||
def test_version_change_changes_hash(self):
|
||||
v1 = Tool(name="test", version="1.0.0")
|
||||
v2 = Tool(name="test", version="2.0.0")
|
||||
assert compute_tool_hash(v1) != compute_tool_hash(v2)
|
||||
|
||||
|
||||
class TestIntegrityChain:
|
||||
def test_single_tool_chain(self):
|
||||
tool = Tool(name="solo", version="1.0.0")
|
||||
chain = build_integrity_chain(tool)
|
||||
assert chain.root.name == "solo"
|
||||
assert chain.is_valid
|
||||
|
||||
def test_chain_with_unresolved_dep(self):
|
||||
tool = Tool(
|
||||
name="parent",
|
||||
version="1.0.0",
|
||||
steps=[ToolStep(tool="missing-dep", output_var="x")],
|
||||
output="{x}",
|
||||
)
|
||||
chain = build_integrity_chain(tool)
|
||||
assert "missing-dep" not in chain.nodes
|
||||
# Chain is still valid because the dependency hash is "unresolved"
|
||||
# and there's no node to compare against
|
||||
assert chain.is_valid
|
||||
|
||||
def test_chain_valid_with_resolved_dep(self, tmp_path):
|
||||
with patch("cmdforge.tool.TOOLS_DIR", tmp_path / ".cmdforge"):
|
||||
from cmdforge.tool import save_tool
|
||||
|
||||
child = Tool(name="child", version="1.0.0")
|
||||
save_tool(child)
|
||||
|
||||
parent = Tool(
|
||||
name="parent",
|
||||
version="1.0.0",
|
||||
steps=[ToolStep(tool="child", output_var="x")],
|
||||
output="{x}",
|
||||
)
|
||||
save_tool(parent)
|
||||
|
||||
chain = build_integrity_chain(parent)
|
||||
assert "child" in chain.nodes
|
||||
assert chain.is_valid
|
||||
|
||||
|
||||
class TestVerifyIntegrity:
|
||||
def test_valid_chain(self):
|
||||
chain = IntegrityChain(
|
||||
root=IntegrityNode(name="root", content_hash="abc"),
|
||||
nodes={"root": IntegrityNode(name="root", content_hash="abc")},
|
||||
)
|
||||
assert verify_integrity(chain)
|
||||
|
||||
def test_tampered_chain(self):
|
||||
chain = IntegrityChain(
|
||||
root=IntegrityNode(
|
||||
name="root",
|
||||
content_hash="abc",
|
||||
dependencies=["dep"],
|
||||
dependency_hashes={"dep": "expected_hash"},
|
||||
),
|
||||
nodes={
|
||||
"root": IntegrityNode(
|
||||
name="root",
|
||||
content_hash="abc",
|
||||
dependencies=["dep"],
|
||||
dependency_hashes={"dep": "expected_hash"},
|
||||
),
|
||||
"dep": IntegrityNode(name="dep", content_hash="different_hash"),
|
||||
},
|
||||
)
|
||||
assert not chain.is_valid
|
||||
Reference in New Issue
Block a user