10c6af3736
- Recipe schema using artdag primitives (SOURCE, EFFECT) - identity-cat: applies identity effect to cat image - Test verifying identity property: output_hash == input_hash - Demonstrates content-addressed DAG execution Owner: @giles@artdag.rose-ash.com 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
112 lines
3.0 KiB
Python
112 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test the identity-cat recipe.
|
|
|
|
Verifies that identity(cat) produces the same content hash as cat.
|
|
"""
|
|
|
|
import hashlib
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
# Add artdag to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "artdag"))
|
|
|
|
from artdag.dag import DAG, Node, NodeType
|
|
from artdag.engine import Engine
|
|
from artdag import nodes # Register executors
|
|
|
|
|
|
def file_hash(path: Path) -> str:
|
|
"""Compute SHA3-256 hash of a file."""
|
|
hasher = hashlib.sha3_256()
|
|
# Resolve symlinks to get actual content
|
|
actual_path = path.resolve() if path.is_symlink() else path
|
|
with open(actual_path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(65536), b""):
|
|
hasher.update(chunk)
|
|
return hasher.hexdigest()
|
|
|
|
|
|
def main():
|
|
# Cat image location (from registry)
|
|
cat_url = "https://rose-ash.com/content/images/2026/01/cat.jpg"
|
|
cat_hash = "33268b6e167deaf018cc538de12dbe562612b33e89a749391cef855b320a269b"
|
|
|
|
# Try to find cat locally
|
|
cat_paths = [
|
|
Path("/home/giles/Pictures/cat.jpg"),
|
|
Path.home() / "Pictures" / "cat.jpg",
|
|
]
|
|
|
|
cat_path = None
|
|
for p in cat_paths:
|
|
if p.exists():
|
|
cat_path = p
|
|
break
|
|
|
|
if not cat_path:
|
|
print(f"Cat image not found locally. Download from: {cat_url}")
|
|
sys.exit(1)
|
|
|
|
# Verify input hash
|
|
input_hash = file_hash(cat_path)
|
|
print(f"Input: {cat_path}")
|
|
print(f"Input hash: {input_hash}")
|
|
|
|
if input_hash != cat_hash:
|
|
print(f"WARNING: Input hash doesn't match registry!")
|
|
print(f"Expected: {cat_hash}")
|
|
|
|
# Build the DAG: SOURCE -> EFFECT(identity) -> output
|
|
dag = DAG()
|
|
|
|
# Node 1: SOURCE(cat)
|
|
source_node = Node(
|
|
node_type=NodeType.SOURCE,
|
|
config={"path": str(cat_path)},
|
|
inputs=[],
|
|
name="source_cat",
|
|
)
|
|
source_id = dag.add_node(source_node)
|
|
|
|
# Node 2: EFFECT(identity)
|
|
effect_node = Node(
|
|
node_type="EFFECT",
|
|
config={"effect": "identity"},
|
|
inputs=[source_id],
|
|
name="apply_identity",
|
|
)
|
|
effect_id = dag.add_node(effect_node)
|
|
|
|
dag.set_output(effect_id)
|
|
|
|
# Execute
|
|
with tempfile.TemporaryDirectory() as cache_dir:
|
|
engine = Engine(cache_dir)
|
|
result = engine.execute(dag)
|
|
|
|
if not result.success:
|
|
print(f"FAILED: {result.error}")
|
|
sys.exit(1)
|
|
|
|
# Verify output hash
|
|
output_hash = file_hash(result.output_path)
|
|
print(f"\nOutput: {result.output_path}")
|
|
print(f"Output hash: {output_hash}")
|
|
|
|
# The identity property: output hash == input hash
|
|
if output_hash == input_hash:
|
|
print("\n✓ PASS: identity(cat) == cat")
|
|
print(" Output hash equals input hash - identity property verified!")
|
|
else:
|
|
print("\n✗ FAIL: identity(cat) != cat")
|
|
print(f" Expected: {input_hash}")
|
|
print(f" Got: {output_hash}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|