Python: Implement missing features for parity with JVM

python/pytest-selfie/docs/index.md

@@ -1,3 +1,110 @@
 # Welcome to pytest-selfie
 
 A pytest plugin for selfie snapshot testing.
+
+## Quick Start
+
+1. Install pytest-selfie:
+```bash
+pip install pytest-selfie
+```
+
+2. Write a test:
+```python
+def test_my_function():
+    result = my_function()
+    assert_that(result).matches_snapshot()
+```
+
+3. Run the test:
+```bash
+pytest
+```
+
+The first time you run the test, it will fail because there's no snapshot. Add `_TODO` to the test name to create the snapshot:
+
+```python
+def test_my_function_TODO():
+    result = my_function()
+    assert_that(result).matches_snapshot()
+```
+
+Run the test again, and the snapshot will be created. Remove `_TODO` and run again to verify.
+
+## Features
+
+### Zero Setup Snapshot Testing
+- No configuration required
+- Works out of the box with pytest
+- Automatic snapshot management
+- Supports both inline and disk-based snapshots
+
+### Inline and Disk-Based Snapshots
+- Inline snapshots are stored directly in your test file
+- Disk-based snapshots are stored in `.snapshot` directories
+- Use `#selfieonce` or `#SELFIEWRITE` comments to update all snapshots in a file
+- Use `--selfie-overwrite` flag to update all snapshots in the project
+
+### Binary Data Support
+- Handle binary data with built-in facets
+- Support for common binary formats
+- Custom binary facet creation
+```python
+def test_binary_data():
+    data = get_binary_data()
+    assert_that(data).as_binary().matches_snapshot()
+```
+
+### JSON Serialization
+- Built-in JSON serialization support
+- Pretty-printed JSON output
+- Customizable JSON formatting
+```python
+def test_json_data():
+    data = {"key": "value"}
+    assert_that(data).as_json().matches_snapshot()
+```
+
+### Async/Coroutines Support
+- Full support for async/await functions
+- Cache async results for snapshot comparison
+- Easy integration with async test frameworks
+```python
+async def test_async_function():
+    result = await async_function()
+    assert_that(result).matches_snapshot()
+
+# Using cache_selfie_suspend for async operations
+async def test_cached_async():
+    cache = await cache_selfie_suspend(
+        disk,
+        roundtrip,
+        async_value
+    )
+    assert cache.generator() == expected_value
+```
+
+## Advanced Usage
+
+### Test File Annotations
+- `#selfieonce`: Update all snapshots in the file once
+- `#SELFIEWRITE`: Always update snapshots in the file
+- Add `_TODO` to test names for one-time updates
+
+### Command Line Options
+- `--selfie-overwrite`: Update all snapshots in the project
+- `--selfie-mode=readonly`: Prevent snapshot updates
+- `--selfie-mode=interactive`: Default mode, allows updates with annotations
+
+### Error Messages
+Clear error messages guide you through snapshot management:
+- Missing snapshots: Instructions for creating new snapshots
+- Mismatched snapshots: Options for updating existing snapshots
+- File not found: Guidance for creating snapshot files
+
+## Best Practices
+1. Keep snapshots focused and minimal
+2. Use meaningful test names
+3. Review snapshot changes carefully
+4. Use `_TODO` for intentional updates
+5. Commit snapshot files with your code

python/selfie-lib/pyproject.toml

@@ -17,6 +17,7 @@ dependencies = []
 dev = [
     "pyright>=1.1.350",
     "pytest>=8.0.0",
+    "pytest-asyncio>=0.24.0",
     "ruff>=0.5.0",
 ]
 
@@ -34,3 +35,7 @@ ignore = [ "S", "FA",  "PYI", "EM", "PLR",  "FBT", "COM", "RET", "PTH", "PLW", "
 "PGH003", # specific rule codes when ignoring type issues
 "ISC001"
 ]
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]

python/selfie-lib/selfie_lib/RoundtripJson.py

@@ -0,0 +1,16 @@
+import json
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+class RoundtripJson(Generic[T]):
+    @staticmethod
+    def of() -> "RoundtripJson[T]":
+        return RoundtripJson()
+
+    def to_string(self, value: T) -> str:
+        return json.dumps(value, indent=2)
+
+    def from_string(self, str_value: str) -> T:
+        return json.loads(str_value)

python/selfie-lib/selfie_lib/SelfieImplementations.py

@@ -187,16 +187,45 @@ def __init__(self, actual: Snapshot, disk: DiskStorage, only_facet: str):
             )
 
     def to_be_base64(self, expected: str) -> bytes:
-        raise NotImplementedError
+        actual_bytes = self.actual.subject_or_facet(self.only_facet).value_binary()
+        actual_base64 = base64.b64encode(actual_bytes).decode().replace("\r", "")
+        if actual_base64 == expected:
+            return actual_bytes
+        else:
+            _toBeDidntMatch(expected, actual_base64, LiteralString())
+            return actual_bytes
 
     def to_be_base64_TODO(self, _: Any = None) -> bytes:
-        raise NotImplementedError
+        actual_bytes = self.actual.subject_or_facet(self.only_facet).value_binary()
+        actual_base64 = base64.b64encode(actual_bytes).decode().replace("\r", "")
+        _toBeDidntMatch(None, actual_base64, LiteralString())
+        return actual_bytes
 
     def to_be_file(self, subpath: str) -> bytes:
-        raise NotImplementedError
+        actual_bytes = self.actual.subject_or_facet(self.only_facet).value_binary()
+        call = recordCall(False)
+        if _selfieSystem().mode.can_write(False, call, _selfieSystem()):
+            self.disk.write_disk(Snapshot.of(actual_bytes), subpath, call)
+        else:
+            expected = self.disk.read_disk(subpath, call)
+            if expected is None:
+                raise _selfieSystem().fs.assert_failed(
+                    _selfieSystem().mode.msg_snapshot_not_found()
+                )
+            _assertEqual(expected, Snapshot.of(actual_bytes), _selfieSystem())
+        return actual_bytes
 
     def to_be_file_TODO(self, subpath: str) -> bytes:
-        raise NotImplementedError
+        actual_bytes = self.actual.subject_or_facet(self.only_facet).value_binary()
+        call = recordCall(False)
+        if _selfieSystem().mode.can_write(True, call, _selfieSystem()):
+            self.disk.write_disk(Snapshot.of(actual_bytes), subpath, call)
+            _selfieSystem().write_inline(TodoStub.to_be_file.create_literal(), call)
+            return actual_bytes
+        else:
+            raise _selfieSystem().fs.assert_failed(
+                f"Can't call `toBeFile_TODO` in {Mode.readonly} mode!"
+            )
 
 
 def _checkSrc(value: T) -> T:

python/selfie-lib/selfie_lib/SnapshotSystem.py

@@ -119,13 +119,28 @@ def can_write(self, is_todo: bool, call: CallStack, system: SnapshotSystem) -> b
             raise ValueError(f"Unknown mode: {self}")
 
     def msg_snapshot_not_found(self) -> str:
-        return self.msg("Snapshot not found")
+        return self.msg(
+            "Snapshot not found. To create the snapshot:\n"
+            "1. Add '_TODO' to the function name\n"
+            "2. Or add '#selfieonce' or '#SELFIEWRITE' to the file"
+        )
 
     def msg_snapshot_not_found_no_such_file(self, file) -> str:
-        return self.msg(f"Snapshot not found: no such file {file}")
+        return self.msg(
+            f"Snapshot not found: file '{file}' does not exist.\n"
+            "To create the snapshot file:\n"
+            "1. Add '_TODO' to the function name\n"
+            "2. Or add '#selfieonce' or '#SELFIEWRITE' to the test file\n"
+            "3. Or run with --selfie-overwrite to create all missing snapshots"
+        )
 
     def msg_snapshot_mismatch(self) -> str:
-        return self.msg("Snapshot mismatch")
+        return self.msg(
+            "Snapshot does not match expected value. To update:\n"
+            "1. Add '_TODO' to the function name to update this snapshot\n"
+            "2. Or add '#selfieonce' or '#SELFIEWRITE' to update all snapshots in this file\n"
+            "3. Or run with --selfie-overwrite to update all snapshots in the project"
+        )
 
     def msg(self, headline: str) -> str:
         if self == Mode.interactive:

python/selfie-lib/selfie_lib/TodoStub.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class TodoStubMethod:
+    name: str
+
+    def create_literal(self) -> str:
+        return f"_{self.name}()"
+
+
+@dataclass
+class TodoStub:
+    to_be_file = TodoStubMethod("toBeFile")
+    to_be_base64 = TodoStubMethod("toBeBase64")

python/selfie-lib/selfie_lib/coroutines/CacheSelfie.py

@@ -0,0 +1,17 @@
+from collections.abc import Awaitable
+from typing import Callable, TypeVar
+
+from selfie_lib.CacheSelfie import CacheSelfie
+from selfie_lib.Roundtrip import Roundtrip
+from selfie_lib.SnapshotSystem import DiskStorage
+
+T = TypeVar("T")
+
+
+async def cache_selfie_suspend(
+    disk: DiskStorage,
+    roundtrip: Roundtrip[T, str],
+    to_cache: Callable[[], Awaitable[T]],
+) -> CacheSelfie[T]:
+    result = await to_cache()
+    return CacheSelfie(disk, roundtrip, lambda: result)

python/selfie-lib/selfie_lib/coroutines/__init__.py

@@ -0,0 +1,3 @@
+from .CacheSelfie import cache_selfie_suspend
+
+__all__ = ["cache_selfie_suspend"]

python/selfie-lib/tests/RoundtripJson_test.py

@@ -0,0 +1,22 @@
+from selfie_lib.RoundtripJson import RoundtripJson
+
+
+def test_roundtrip_json_simple():
+    roundtrip = RoundtripJson.of()
+    value = {"key": "value", "number": 42}
+    json_str = roundtrip.to_string(value)
+    assert roundtrip.from_string(json_str) == value
+
+
+def test_roundtrip_json_nested():
+    roundtrip = RoundtripJson.of()
+    value = {"nested": {"array": [1, 2, 3], "object": {"key": "value"}}}
+    json_str = roundtrip.to_string(value)
+    assert roundtrip.from_string(json_str) == value
+
+
+def test_roundtrip_json_special_chars():
+    roundtrip = RoundtripJson.of()
+    value = {"special": 'line\nbreak\ttab"quote\\backslash', "unicode": "🐍Python"}
+    json_str = roundtrip.to_string(value)
+    assert roundtrip.from_string(json_str) == value

python/selfie-lib/tests/coroutines/__init__.py

@@ -0,0 +1 @@
+# Test package initialization

python/selfie-lib/tests/coroutines/test_CacheSelfie.py

@@ -0,0 +1,39 @@
+import asyncio
+from typing import Optional
+
+import pytest
+
+from selfie_lib.coroutines.CacheSelfie import cache_selfie_suspend
+from selfie_lib.Roundtrip import Roundtrip
+from selfie_lib.SnapshotSystem import DiskStorage, Snapshot
+from selfie_lib.WriteTracker import CallStack
+
+
+class TestDiskStorage(DiskStorage):
+    def read_disk(
+        self,
+        sub: str,  # noqa: ARG002
+        call: CallStack,  # noqa: ARG002
+    ) -> Optional[Snapshot]:
+        return None
+
+    def write_disk(self, actual: Snapshot, sub: str, call: CallStack):
+        pass
+
+    def keep(self, sub_or_keep_all: Optional[str]):
+        pass
+
+
+async def async_value() -> str:
+    await asyncio.sleep(0.1)
+    return "test_value"
+
+
+@pytest.mark.asyncio
+async def test_cache_selfie_suspend():
+    disk = TestDiskStorage()
+    roundtrip = Roundtrip[str, str]()
+
+    cache = await cache_selfie_suspend(disk, roundtrip, async_value)
+
+    assert cache.generator() == "test_value"