SnapshotReader passing all the test

python/selfie-lib/selfie_lib/ArrayMap.py

@@ -9,10 +9,12 @@
 
 class ListBackedSet(Set[T], ABC):
     @abstractmethod
-    def __len__(self) -> int: ...
+    def __len__(self) -> int:
+        ...
 
     @abstractmethod
-    def __getitem__(self, index: Union[int, slice]) -> Union[T, List[T]]: ...
+    def __getitem__(self, index: Union[int, slice]) -> Union[T, List[T]]:
+        ...
 
     def __contains__(self, item: object) -> bool:
         for i in range(len(self)):

python/selfie-lib/selfie_lib/ConvertToWindowsNewlines.py

@@ -5,10 +5,10 @@ def __init__(self, sink):
     def append(self, value, start_index=None, end_index=None):
         # If value is a single character
         if isinstance(value, str) and len(value) == 1:
-            if value != '\n':
+            if value != "\n":
                 self.sink.write(value)
             else:
-                self.sink.write('\r\n')
+                self.sink.write("\r\n")
         # If value is a CharSequence (in Python, a str)
         elif isinstance(value, str):
             # If start_index and end_index are provided, use the slice of the string

python/selfie-lib/selfie_lib/Snapshot.py

@@ -1,5 +1,10 @@
 from .SnapshotValue import SnapshotValue
 from collections import OrderedDict
+import logging
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
 
 class Snapshot:
     def __init__(self, subject, facet_data):
@@ -10,6 +15,14 @@ def __init__(self, subject, facet_data):
     def facets(self):
         return OrderedDict(sorted(self._facet_data.items()))
 
+    def __eq__(self, other):
+        if not isinstance(other, Snapshot):
+            return NotImplemented
+        return self._subject == other._subject and self._facet_data == other._facet_data
+
+    def __hash__(self):
+        return hash((self._subject, frozenset(self._facet_data.items())))
+
     def plus_facet(self, key, value):
         if isinstance(value, bytes):
             value = SnapshotValue.of(value)
@@ -47,7 +60,7 @@ def all_entries(self):
         entries.extend(self._facet_data.items())
         return entries
 
-    def __str__(self):
+    def __bytes__(self):
         return f"[{self._subject} {self._facet_data}]"
 
     @staticmethod
@@ -58,8 +71,10 @@ def of(data):
         elif isinstance(data, str):
             # Handling string data
             return Snapshot(SnapshotValue.of(data), {})
+        elif isinstance(data, SnapshotValue):
+            return Snapshot(data, {})
         else:
-            raise TypeError("Data must be either binary or string")
+            raise TypeError("Data must be either binary or string" + data)
 
     @staticmethod
     def of_entries(entries):
@@ -68,7 +83,9 @@ def of_entries(entries):
         for key, value in entries:
             if not key:
                 if subject is not None:
-                    raise ValueError(f"Duplicate root snapshot.\n first: {subject}\nsecond: {value}")
+                    raise ValueError(
+                        f"Duplicate root snapshot.\n first: {subject}\nsecond: {value}"
+                    )
                 subject = value
             else:
                 facet_data[key] = value
@@ -78,4 +95,4 @@ def of_entries(entries):
 
     @staticmethod
     def _unix_newlines(string):
-        return string.replace("\\r\\n", "\\n")
+        return string.replace("\\r\\n", "\\n")

python/selfie-lib/selfie_lib/SnapshotFile.py

@@ -1,121 +1,134 @@
-import threading
-from typing import List
-import base64
-
-from .SnapshotValue import SnapshotValue
-from .ConvertToWindowsNewlines import ConvertToWindowsNewlines
-from .ParseException import ParseException
-from .SnapshotReader import SnapshotReader
-from .SnapshotValueReader import SnapshotValueReader
-
-class SnapshotFile:
-    HEADER_PREFIX = "📷 "
-    END_OF_FILE = "[end of file]"
-
-    def __init__(self):
-        self.unix_newlines = True
-        self.metadata = None
-        self._snapshots = {}
-        self._lock = threading.Lock()
-        self._was_set_at_test_time = False
-
-    @property
-    def snapshots(self):
-        return self._snapshots
-
-    @snapshots.setter
-    def snapshots(self, value):
-        with self._lock:
-            self._snapshots = value
-
-    @property
-    def was_set_at_test_time(self):
-        return self._was_set_at_test_time
-
-    def set_at_test_time(self, key, snapshot):
-        with self._lock:
-            old_snapshots = self._snapshots.copy()
-            self._snapshots[key] = snapshot
-            self._was_set_at_test_time = True if self._snapshots != old_snapshots else self._was_set_at_test_time
-
-    def serialize(self, value_writer_raw):
-        value_writer = value_writer_raw if self.unix_newlines else ConvertToWindowsNewlines(value_writer_raw)
-        if self.metadata:
-            self.write_entry(value_writer, f"📷 {self.metadata[0]}", None, SnapshotValue.of(self.metadata[1]))
-        for key, snapshot in self._snapshots.items():
-            self.write_entry(value_writer, key, None, snapshot.subject)
-            for facet_key, facet_value in snapshot.facets.items():
-                self.write_entry(value_writer, key, facet_key, facet_value)
-        self.write_entry(value_writer, "", "end of file", SnapshotValue.of(""))
-
-    def write_entry(value_writer, key, facet, value):
-        value_writer.write("╔═ ")
-        value_writer.write(SnapshotValueReader.nameEsc.escape(key))
-        if facet is not None:
-            value_writer.write("[")
-            value_writer.write(SnapshotValueReader.nameEsc.escape(facet))
-            value_writer.write("]")
-        value_writer.write(" ═╗")
-        if value.is_binary:
-            binary_length = len(value.value_binary())
-            value_writer.write(f" base64 length {binary_length} bytes")
-        value_writer.write("\n")
-
-        if key == "" and facet == "end of file":
-            return
-
-        if value.is_binary:
-            # Base64 encoding and replacing \r with an empty string
-            binary_data = value.value_binary()
-            encoded = base64.b64encode(binary_data).decode('utf-8')
-            # Assuming efficientReplace is a more efficient method for replacing characters
-            # Here, we just use the regular replace method for simplicity
-            escaped = encoded.replace("\r", "")
-            value_writer.write(escaped)
-        else:
-            # For string values, applying specific escape logic and then replacing "\n╔" with a special sequence
-            text_data = value.value_string()
-            escaped = SnapshotValueReader.bodyEsc(text_data).replace("\n╔", "\n\uDF41")
-            value_writer.write(escaped)
-        value_writer.write("\n")
-
-    @staticmethod
-    def parse(value_reader):
-        try:
-            result = SnapshotFile()
-            result.unix_newlines = value_reader.unix_newlines
-            reader = SnapshotReader(value_reader)
-
-            # Check if the first value starts with 📷
-            if reader.peek_key() and reader.peek_key().startswith(SnapshotFile.HEADER_PREFIX):
-                metadata_name = reader.peek_key()[len(SnapshotFile.HEADER_PREFIX):]
-                metadata_value = reader.value_reader.next_value().value_string()
-                # Assuming 'entry' function creates a dictionary entry in Python
-                result.metadata = (metadata_name, metadata_value)
-
-            while reader.peek_key() is not None:
-                key = reader.peek_key()
-                snapshot = reader.next_snapshot()
-                # Update snapshots dictionary with new key-value pair
-                result.snapshots.update({key: snapshot})
-
-            return result
-
-        except ValueError as e:
-            if isinstance(e, ParseException):
-                raise e
-            else:
-                raise ParseException(value_reader.line_reader, e) from None
-
-
-    @staticmethod
-    def create_empty_with_unix_newlines(unix_newlines):
-        result = SnapshotFile()
-        result.unix_newlines = unix_newlines
-        return result
-
-    def remove_all_indices(self, indices: List[int]):
-        if not indices:
-            return
-        self._was_set_at_test_time = True
-        self.snapshots = self.snapshots.minus_sorted_indices(indices)
+# import threading
+# from typing import List
+# import base64
+
+# from .SnapshotValue import SnapshotValue
+# from .ConvertToWindowsNewlines import ConvertToWindowsNewlines
+# from .ParseException import ParseException
+# from .SnapshotReader import SnapshotReader
+# from .SnapshotValueReader import SnapshotValueReader
+
+
+# class SnapshotFile:
+#     HEADER_PREFIX = "📷 "
+#     END_OF_FILE = "[end of file]"
+
+#     def __init__(self):
+#         self.unix_newlines = True
+#         self.metadata = None
+#         self._snapshots = {}
+#         self._lock = threading.Lock()
+#         self._was_set_at_test_time = False
+
+#     @property
+#     def snapshots(self):
+#         return self._snapshots
+
+#     @snapshots.setter
+#     def snapshots(self, value):
+#         with self._lock:
+#             self._snapshots = value
+
+#     @property
+#     def was_set_at_test_time(self):
+#         return self._was_set_at_test_time
+
+#     def set_at_test_time(self, key, snapshot):
+#         with self._lock:
+#             old_snapshots = self._snapshots.copy()
+#             self._snapshots[key] = snapshot
+#             self._was_set_at_test_time = (
+#                 True if self._snapshots != old_snapshots else self._was_set_at_test_time
+#             )
+
+#     def serialize(self, value_writer_raw):
+#         value_writer = (
+#             value_writer_raw
+#             if self.unix_newlines
+#             else ConvertToWindowsNewlines(value_writer_raw)
+#         )
+#         if self.metadata:
+#             self.write_entry(
+#                 value_writer,
+#                 f"📷 {self.metadata[0]}",
+#                 None,
+#                 SnapshotValue.of(self.metadata[1]),
+#             )
+#         for key, snapshot in self._snapshots.items():
+#             self.write_entry(value_writer, key, None, snapshot.subject)
+#             for facet_key, facet_value in snapshot.facets.items():
+#                 self.write_entry(value_writer, key, facet_key, facet_value)
+#         self.write_entry(value_writer, "", "end of file", SnapshotValue.of(""))
+
+#     def write_entry(value_writer, key, facet, value):
+#         value_writer.write("╔═ ")
+#         value_writer.write(SnapshotValueReader.nameEsc.escape(key))
+#         if facet is not None:
+#             value_writer.write("[")
+#             value_writer.write(SnapshotValueReader.nameEsc.escape(facet))
+#             value_writer.write("]")
+#         value_writer.write(" ═╗")
+#         if value.is_binary:
+#             binary_length = len(value.value_binary())
+#             value_writer.write(f" base64 length {binary_length} bytes")
+#         value_writer.write("\n")
+
+#         if key == "" and facet == "end of file":
+#             return
+
+#         if value.is_binary:
+#             # Base64 encoding and replacing \r with an empty string
+#             binary_data = value.value_binary()
+#             encoded = base64.b64encode(binary_data).decode("utf-8")
+#             # Assuming efficientReplace is a more efficient method for replacing characters
+#             # Here, we just use the regular replace method for simplicity
+#             escaped = encoded.replace("\r", "")
+#             value_writer.write(escaped)
+#         else:
+#             # For string values, applying specific escape logic and then replacing "\n╔" with a special sequence
+#             text_data = value.value_string()
+#             escaped = SnapshotValueReader.bodyEsc(text_data).replace("\n╔", "\n\uDF41")
+#             value_writer.write(escaped)
+#         value_writer.write("\n")
+
+#     @staticmethod
+#     def parse(value_reader):
+#         try:
+#             result = SnapshotFile()
+#             result.unix_newlines = value_reader.unix_newlines
+#             reader = SnapshotReader(value_reader)
+
+#             # Check if the first value starts with 📷
+#             if reader.peek_key() and reader.peek_key().startswith(
+#                 SnapshotFile.HEADER_PREFIX
+#             ):
+#                 metadata_name = reader.peek_key()[len(SnapshotFile.HEADER_PREFIX) :]
+#                 metadata_value = reader.value_reader.next_value().value_string()
+#                 # Assuming 'entry' function creates a dictionary entry in Python
+#                 result.metadata = (metadata_name, metadata_value)
+
+#             while reader.peek_key() is not None:
+#                 key = reader.peek_key()
+#                 snapshot = reader.next_snapshot()
+#                 # Update snapshots dictionary with new key-value pair
+#                 result.snapshots.update({key: snapshot})
+
+#             return result
+
+#         except ValueError as e:
+#             if isinstance(e, ParseException):
+#                 raise e
+#             else:
+#                 raise ParseException(value_reader.line_reader, e) from None
+
+#     @staticmethod
+#     def create_empty_with_unix_newlines(unix_newlines):
+#         result = SnapshotFile()
+#         result.unix_newlines = unix_newlines
+#         return result
+
+#     def remove_all_indices(self, indices: List[int]):
+#         if not indices:
+#             return
+#         self._was_set_at_test_time = True
+#         self.snapshots = self.snapshots.minus_sorted_indices(indices)

python/selfie-lib/selfie_lib/SnapshotReader.py

@@ -1,5 +1,6 @@
 from .Snapshot import Snapshot
 
+
 class SnapshotReader:
     def __init__(self, value_reader):
         self.value_reader = value_reader
@@ -8,8 +9,10 @@ def peek_key(self):
         next_key = self.value_reader.peek_key()
         if next_key is None or next_key == "[end of file]":
             return None
-        if '[' in next_key:
-            raise ValueError(f"Missing root snapshot, square brackets not allowed: '{next_key}'")
+        if "[" in next_key:
+            raise ValueError(
+                f"Missing root snapshot, square brackets not allowed: '{next_key}'"
+            )
         return next_key
 
     def next_snapshot(self):
@@ -19,18 +22,20 @@ def next_snapshot(self):
             next_key = self.value_reader.peek_key()
             if next_key is None:
                 return snapshot
-            facet_idx = next_key.find('[')
+            facet_idx = next_key.find("[")
             if facet_idx == -1 or (facet_idx == 0 and next_key == "[end of file]"):
                 return snapshot
             facet_root = next_key[:facet_idx]
             if facet_root != root_name:
-                raise ValueError(f"Expected '{next_key}' to come after '{facet_root}', not '{root_name}'")
-            facet_end_idx = next_key.find(']', facet_idx + 1)
+                raise ValueError(
+                    f"Expected '{next_key}' to come after '{facet_root}', not '{root_name}'"
+                )
+            facet_end_idx = next_key.find("]", facet_idx + 1)
             if facet_end_idx == -1:
                 raise ValueError(f"Missing ] in {next_key}")
-            facet_name = next_key[facet_idx + 1:facet_end_idx]
+            facet_name = next_key[facet_idx + 1 : facet_end_idx]
             snapshot = snapshot.plus_facet(facet_name, self.value_reader.next_value())
-    
+
     def skip_snapshot(self):
         root_name = self.peek_key()
         if root_name is None: