Add failing test for verify_hash.g

Author: goloroden

eventsourcingdb/event/event.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime
 import json
 from hashlib import sha256
@@ -10,7 +10,7 @@
 Self = TypeVar("Self", bound="Event")
 
 
-@dataclass(frozen=True)
+@dataclass
 class Event:
     data: dict
     source: str
@@ -19,7 +19,7 @@ class Event:
     spec_version: str
     event_id: str
     time: datetime
-    time_str: str
+    _time_from_server: str = field(init=False, repr=False)
     data_content_type: str
     predecessor_hash: str
     hash: str
@@ -48,10 +48,10 @@ def parse(unknown_object: dict) -> "Event":
         if not isinstance(event_id, str):
             raise ValidationError(f"Failed to parse event_id '{event_id}' to string.")
 
-        time_str = unknown_object.get("time")
-        if not isinstance(time_str, str):
-            raise ValidationError(f"Failed to parse time '{time_str}' to string.")
-        time = Event.__parse_time(time_str)
+        time_from_server = unknown_object.get("time")
+        if not isinstance(time_from_server, str):
+            raise ValidationError(f"Failed to parse time '{time_from_server}' to string.")
+        time = Event.__parse_time(time_from_server)
 
         data_content_type = unknown_object.get("datacontenttype")
         if not isinstance(data_content_type, str):
@@ -82,28 +82,30 @@ def parse(unknown_object: dict) -> "Event":
         if not isinstance(data, dict):
             raise ValidationError(f"Failed to parse data '{data}' to object.")
 
-        return Event(
+        event = Event(
             data=data,
             source=source,
             subject=subject,
             type=event_type,
             spec_version=spec_version,
             event_id=event_id,
             time=time,
-            time_str=time_str,
             data_content_type=data_content_type,
             predecessor_hash=predecessor_hash,
             hash=hash,
             trace_parent=trace_parent,
             trace_state=trace_state,
         )
+        event._time_from_server = time_from_server
+
+        return event
 
     def verify_hash(self) -> None:
         metadata = "|".join([
             self.spec_version,
             self.event_id,
             self.predecessor_hash,
-            self.time_str,
+            self._time_from_server,
             self.source,
             self.subject,
             self.type,
@@ -151,17 +153,17 @@ def to_json(self) -> dict[str, Any]:
         return json
 
     @staticmethod
-    def __parse_time(time_str: str) -> datetime:
-        if not isinstance(time_str, str):
-            raise ValidationError(f"Failed to parse time '{time_str}' to datetime.")
+    def __parse_time(time_from_server: str) -> datetime:
+        if not isinstance(time_from_server, str):
+            raise ValidationError(f"Failed to parse time '{time_from_server}' to datetime.")
 
-        rest, sub_seconds = time_str.split(".")
+        rest, sub_seconds = time_from_server.split(".")
         sub_seconds = f"{sub_seconds[:6]:06}"
         try:
             return datetime.fromisoformat(f"{rest}.{sub_seconds}")
         except ValueError as value_error:
             raise ValidationError(
-                f"Failed to parse time '{time_str}' to datetime."
+                f"Failed to parse time '{time_from_server}' to datetime."
             ) from value_error
         except Exception as other_error:
             raise InternalError(str(other_error)) from other_error

tests/event/test_verify_hash.py

@@ -1,6 +1,7 @@
 import pytest
 
 from eventsourcingdb import EventCandidate
+from hashlib import sha256
 
 from ..conftest import TestData
 from ..shared.database import Database
@@ -27,3 +28,29 @@ async def test_verifies_the_event_hash(
 
         written_event = written_events[0]
         written_event.verify_hash()
+
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_fails_if_the_event_hash_is_invalid(
+        database: Database,
+        test_data: TestData,
+    ) -> None:
+        client = database.get_client()
+
+        written_events = await client.write_events(
+            [
+                EventCandidate(
+                    source=test_data.TEST_SOURCE_STRING, subject="/test", type="io.eventsourcingdb.test", data={"value": 23}
+                )
+            ],
+        )
+
+        assert len(written_events) == 1
+
+        written_event = written_events[0]
+
+        invalid_hash = sha256("invalid hash").hexdigest()
+        written_event.hash = invalid_hash
+
+        with pytest.raises(ValidationError):
+            written_event.verify_hash()