events: add branches/tags validation

Author: shtrom

git_hg_sync/events.py

@@ -1,19 +1,27 @@
-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

@@ -42,6 +42,24 @@ def test_parse_invalid_data_types(raw_push_entity: dict) -> None:
         raw_push_entity["branches"] = ["main"]
         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"