feat: add dog effect

- Ignores input, returns dog.mkv from immutable URL
- Downloads from art-source with hash verification
- Caches downloaded file locally
- 5 automated tests (all passing)

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>
This commit is contained in:
gilesb
2026-01-07 01:11:35 +00:00
commit a071084055
6 changed files with 274 additions and 0 deletions

3
.gitignore vendored Normal file
View File

@@ -0,0 +1,3 @@
__pycache__/
*.py[cod]
.pytest_cache/

35
README.md Normal file
View File

@@ -0,0 +1,35 @@
# Art DAG - Effects
Custom effects for the Art DAG engine.
## Effects
| Effect | Description | Output Hash |
|--------|-------------|-------------|
| dog | Returns dog.mkv regardless of input | `772f26f9...` |
## Structure
```
dog/
├── README.md # Documentation
├── requirements.txt # Dependencies
├── effect.py # Implementation
└── test_effect.py # Automated tests
```
## Running Tests
```bash
cd dog && python -m pytest test_effect.py -v
```
## Related
- **Engine**: https://github.com/gilesbradshaw/art-dag
- **Registry**: https://git.rose-ash.com/art-dag/registry
- **Recipes**: https://git.rose-ash.com/art-dag/recipes
## Owner
`@giles@artdag.rose-ash.com`

33
dog/README.md Normal file
View File

@@ -0,0 +1,33 @@
# Dog Effect
The dog effect ignores its input and returns the dog video.
## Signature
```
dog(input) → dog.mkv
```
For any input X: `dog(X) = dog.mkv`
## Source
Fetches from immutable URL:
```
https://git.rose-ash.com/art-dag/art-source/raw/3cb70f9aab6c5e9473e5ce34c59fb1d7dca4bdaa/dog.mkv
```
## Expected Output Hash
```
772f26f9b4e80984788bc48f7c6eee0a1974966b2d4ee56a72d7c6586b3ac9d8
```
## Properties
- **Constant**: Output is independent of input
- **Idempotent**: `dog(dog(x)) = dog(x)`
## Owner
`@giles@artdag.rose-ash.com`

84
dog/effect.py Normal file
View File

@@ -0,0 +1,84 @@
"""
Dog effect - returns dog.mkv regardless of input.
This is a constant effect that fetches from an immutable URL.
"""
import hashlib
import logging
import shutil
from pathlib import Path
from typing import Any, Dict
import requests
logger = logging.getLogger(__name__)
# Immutable source URL (commit-pinned)
DOG_URL = "https://git.rose-ash.com/art-dag/art-source/raw/3cb70f9aab6c5e9473e5ce34c59fb1d7dca4bdaa/dog.mkv"
DOG_HASH = "772f26f9b4e80984788bc48f7c6eee0a1974966b2d4ee56a72d7c6586b3ac9d8"
def file_hash(path: Path) -> str:
"""Compute SHA3-256 hash of a file."""
hasher = hashlib.sha3_256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
hasher.update(chunk)
return hasher.hexdigest()
def effect_dog(input_path: Path, output_path: Path, config: Dict[str, Any]) -> Path:
"""
Dog effect - ignores input, returns dog.mkv.
Downloads from immutable URL and verifies hash.
"""
output_path.parent.mkdir(parents=True, exist_ok=True)
# Output with correct extension
actual_output = output_path.with_suffix(".mkv")
if actual_output.exists():
actual_output.unlink()
# Check cache first
cache_dir = Path.home() / ".artdag" / "effect_cache"
cache_dir.mkdir(parents=True, exist_ok=True)
cached_file = cache_dir / f"{DOG_HASH}.mkv"
if cached_file.exists():
# Verify cached file
if file_hash(cached_file) == DOG_HASH:
logger.debug(f"EFFECT dog: using cached {cached_file}")
shutil.copy2(cached_file, actual_output)
return actual_output
else:
# Cache corrupted, remove it
cached_file.unlink()
# Download from immutable URL
logger.info(f"EFFECT dog: downloading from {DOG_URL}")
response = requests.get(DOG_URL, stream=True)
response.raise_for_status()
# Write to cache
with open(cached_file, "wb") as f:
for chunk in response.iter_content(chunk_size=65536):
f.write(chunk)
# Verify hash
downloaded_hash = file_hash(cached_file)
if downloaded_hash != DOG_HASH:
cached_file.unlink()
raise ValueError(f"Hash mismatch! Expected {DOG_HASH}, got {downloaded_hash}")
# Copy to output
shutil.copy2(cached_file, actual_output)
logger.debug(f"EFFECT dog: {input_path.name} -> {actual_output} (input ignored)")
return actual_output
# Export for registration
effect = effect_dog
name = "dog"

2
dog/requirements.txt Normal file
View File

@@ -0,0 +1,2 @@
# Dog effect requires requests for URL fetching
requests>=2.28.0

117
dog/test_effect.py Normal file
View File

@@ -0,0 +1,117 @@
#!/usr/bin/env python3
"""
Automated tests for the dog effect.
Test plan:
1. Test that effect ignores input and returns dog.mkv
2. Test that output hash matches expected DOG_HASH
3. Test idempotence: dog(dog(x)) == dog(x)
4. Test with different inputs produce same output
"""
import hashlib
import tempfile
import unittest
from pathlib import Path
from effect import effect_dog, DOG_HASH, file_hash
class TestDogEffect(unittest.TestCase):
"""Tests for the dog effect."""
def setUp(self):
"""Create temp directory for tests."""
self.temp_dir = tempfile.mkdtemp()
self.temp_path = Path(self.temp_dir)
def tearDown(self):
"""Clean up temp directory."""
import shutil
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_returns_dog_video(self):
"""Test that effect returns dog.mkv with correct hash."""
# Create a dummy input file
input_path = self.temp_path / "dummy_input.txt"
input_path.write_text("This is a dummy input that should be ignored")
output_path = self.temp_path / "output"
result = effect_dog(input_path, output_path, {})
# Verify output exists
self.assertTrue(result.exists(), "Output file should exist")
# Verify hash
output_hash = file_hash(result)
self.assertEqual(output_hash, DOG_HASH, "Output hash should match DOG_HASH")
def test_ignores_input(self):
"""Test that different inputs produce the same output."""
# Input 1: text file
input1 = self.temp_path / "input1.txt"
input1.write_text("Input 1")
output1 = self.temp_path / "output1"
result1 = effect_dog(input1, output1, {})
# Input 2: different text file
input2 = self.temp_path / "input2.txt"
input2.write_text("Completely different input 2")
output2 = self.temp_path / "output2"
result2 = effect_dog(input2, output2, {})
# Both outputs should have same hash
self.assertEqual(file_hash(result1), file_hash(result2),
"Different inputs should produce same output")
def test_idempotence(self):
"""Test that dog(dog(x)) == dog(x)."""
# First application
input_path = self.temp_path / "input.txt"
input_path.write_text("Original input")
output1 = self.temp_path / "output1"
result1 = effect_dog(input_path, output1, {})
hash1 = file_hash(result1)
# Second application (using first output as input)
output2 = self.temp_path / "output2"
result2 = effect_dog(result1, output2, {})
hash2 = file_hash(result2)
self.assertEqual(hash1, hash2, "dog(dog(x)) should equal dog(x)")
def test_output_extension(self):
"""Test that output has .mkv extension."""
input_path = self.temp_path / "input.txt"
input_path.write_text("test")
output_path = self.temp_path / "output"
result = effect_dog(input_path, output_path, {})
self.assertEqual(result.suffix, ".mkv", "Output should have .mkv extension")
def test_caching(self):
"""Test that second call uses cache (faster)."""
import time
input_path = self.temp_path / "input.txt"
input_path.write_text("test")
# First call (may download)
output1 = self.temp_path / "output1"
start1 = time.time()
effect_dog(input_path, output1, {})
time1 = time.time() - start1
# Second call (should use cache)
output2 = self.temp_path / "output2"
start2 = time.time()
effect_dog(input_path, output2, {})
time2 = time.time() - start2
# Second call should be faster (or at least not much slower)
# We're not strictly asserting this as network conditions vary
print(f"First call: {time1:.3f}s, Second call: {time2:.3f}s")
if __name__ == "__main__":
unittest.main()