commit a0710840559ff4c50873b7ef50b4752ce517d99f Author: gilesb Date: Wed Jan 7 01:11:35 2026 +0000 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 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a160f49 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..5eeab9a --- /dev/null +++ b/README.md @@ -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` diff --git a/dog/README.md b/dog/README.md new file mode 100644 index 0000000..c1d7574 --- /dev/null +++ b/dog/README.md @@ -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` diff --git a/dog/effect.py b/dog/effect.py new file mode 100644 index 0000000..ab08625 --- /dev/null +++ b/dog/effect.py @@ -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" diff --git a/dog/requirements.txt b/dog/requirements.txt new file mode 100644 index 0000000..ab90d5e --- /dev/null +++ b/dog/requirements.txt @@ -0,0 +1,2 @@ +# Dog effect requires requests for URL fetching +requests>=2.28.0 diff --git a/dog/test_effect.py b/dog/test_effect.py new file mode 100644 index 0000000..8568100 --- /dev/null +++ b/dog/test_effect.py @@ -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()