events: add branches/tags validation

Author: shtrom

git_hg_sync/events.py

@@ -1,19 +1,34 @@
-from pydantic import BaseModel
+from typing import Self
+from pydantic import (
+    BaseModel,
+    Field,
+    ValidationError,
+    ValidationInfo,
+    field_validator,
+    model_validator,
+)
 
 
 class Push(BaseModel):
     repo_url: str
-    branches: dict[
-        str, str
-    ]  # Mapping between branch names (key) and corresponding commit sha (value)
-    tags: dict[
-        str, str
-    ]  # Mapping between tag names (key) and corresponding commit sha (value)
+    branches: dict[str, str] | None = Field(
+        default_factory=dict
+    )  # Mapping between branch names (key) and corresponding commit sha (value)
+    tags: dict[str, str] | None = Field(
+        default_factory=dict
+    )  # Mapping between tag names (key) and corresponding commit sha (value)
     time: int
     pushid: int
     user: str
     push_json_url: str
 
+    @model_validator(mode="after")
+    def check_branch_tags(self) -> Self:
+        """Check that at least one of branches or tags is not empty."""
+        if not self.branches and not self.tags:
+            raise ValueError("Either (non-empty) branches or tags is required")
+        return self
+
 
 class Tag(BaseModel):
     repo_url: str

tests/test_integration.py

@@ -24,7 +24,6 @@ def test_send_and_receive(pulse_config: PulseConfig) -> None:
         "repo_url": "repo.git",
         "branches": {},
         "tags": {},
-        "commit": "sha",
         "time": 0,
         "pushid": 0,
         "user": "user",
@@ -114,3 +113,4 @@ def test_full_app(
 
     # test
     assert "BAR CONTENT" in hg_cat(hg_remote_repo_path, "bar.txt", "default")
+    assert "FIREFOX_128_0esr_RELEASE" in hg_cat(hg_remote_repo_path, ".hgtags", "tags")

tests/test_pulse_worker.py

@@ -43,6 +43,30 @@ def test_parse_invalid_data_types(raw_push_entity: dict) -> None:
         PulseWorker.parse_entity(raw_push_entity)
 
 
+@pytest.mark.parametrize(
+    "remove, exception",
+    [
+        ([], None),
+        (["branches"], None),
+        (["tags"], None),
+        (["branches", "tags"], ValidationError),
+    ],
+)
+def test_parse_branch_and_or_tags(
+    raw_push_entity: dict, remove: list[str], exception: Exception | None
+) -> None:
+    entity = raw_push_entity
+    for f in remove:
+        del entity[f]
+
+    if exception:
+        with pytest.raises(exception):
+            PulseWorker.parse_entity(entity)
+    else:
+        push_entity = PulseWorker.parse_entity(entity)
+        assert isinstance(push_entity, Push)
+
+
 def test_sigint_signal_interception() -> None:
     config_file = HERE / "data" / "config.toml"
     module_path = HERE.parent / "git_hg_sync" / "__main__.py"