config: add configurable heartbeat for pulse (bug 1963745) (#70)

* test_integration: add check for duplicated processing (bug 1963745)
* config: add configurable heartbeat for pulse (bug 1963745)
* pulse_worker: fix doctstring location

Author: shtrom

README.md

@@ -137,6 +137,7 @@ variables:
 - PULSE_QUEUE
 - PULSE_ROUTING_KEY
 - PULSE_SSL (needs to be an empty string to be False, otherwise True)
+- PULSE_HEARTBEAT (needs to be an integer)
 - PULSE_USERID
 
 ### SSH key

config-development.toml

@@ -9,6 +9,7 @@ password = "SET THIS IN PULSE_PASSWORD ENV VARIABLE"
 exchange = "exchange/landodev/pushes"
 routing_key = "gitpushes"
 # The Consumer declares the queue and binds it to the exchange.
+heartbeat = 30
 queue = "queue/githgsyncdev/pushes"
 
 [sentry]

config-docker.toml

@@ -6,6 +6,7 @@ exchange = "exchange/git-hg-sync/test"
 routing_key = "git-hg-sync"
 queue = "queue/git-hg-sync/sync"
 password = "guest"
+heartbeat = 30
 ssl = false
 
 [sentry]

config-production.toml

@@ -9,6 +9,7 @@ password = "SET THIS IN PULSE_PASSWORD ENV VARIABLE"
 exchange = "exchange/landoprod/pushes"
 routing_key = "gitpushes"
 # The Consumer declares the queue and binds it to the exchange.
+heartbeat = 30
 queue = "queue/githgsyncprod/pushes"
 
 [sentry]

config-staging.toml

@@ -9,6 +9,7 @@ password = "SET THIS IN PULSE_PASSWORD ENV VARIABLE"
 exchange = "exchange/landostage/pushes"
 routing_key = "gitpushes"
 # The Consumer declares the queue and binds it to the exchange.
+heartbeat = 30
 queue = "queue/githgsyncstage/pushes"
 
 [sentry]

config.toml.example

@@ -6,6 +6,7 @@ exchange = "exchange/<username>/test"
 routing_key = ""
 queue = "queue/<username>/test"
 password = "<pulse-password>"
+heartbeat = 30
 ssl = true
 
 [sentry]

git_hg_sync/__main__.py

@@ -31,7 +31,7 @@ def get_connection(config: PulseConfig) -> Connection:
         port=config.port,
         userid=config.userid,
         password=config.password,
-        heartbeat=10,
+        heartbeat=config.heartbeat,
         ssl=config.ssl,
     )
 

git_hg_sync/config.py

@@ -19,6 +19,7 @@ class PulseConfig(pydantic.BaseModel):
     routing_key: str
     queue: str
     password: str
+    heartbeat: int
     ssl: bool
 
 

git_hg_sync/pulse_worker.py

@@ -21,9 +21,10 @@ class EntityTypeError(Exception):
 
 
 class PulseWorker(ConsumerMixin):
-    event_handler: EventHandler | None
     """Function that will be called whenever an event is received"""
 
+    event_handler: EventHandler | None
+
     def __init__(
         self,
         connection: kombu.Connection,

tests/conftest.py

@@ -1,11 +1,14 @@
 import sys
+from collections.abc import Callable
+from pathlib import Path
 
 import mozlog
 import mozlog.formatters
 import mozlog.handlers
 import pytest
 
-from git_hg_sync.config import PulseConfig
+from git_hg_sync.config import ClonesConfig, Config, PulseConfig, TrackedRepository
+from git_hg_sync.mapping import BranchMapping
 
 
 @pytest.fixture(autouse=True, scope="session")
@@ -27,5 +30,42 @@ def pulse_config() -> PulseConfig:
         routing_key="#",
         queue="queue/guest/test",
         password="guest",
+        heartbeat=30,
         ssl=False,
     )
+
+@pytest.fixture
+def test_config(pulse_config: PulseConfig) -> Config:
+    return Config(
+        pulse=pulse_config,
+        clones=ClonesConfig(directory=Path("clones")),
+        tracked_repositories=[
+            TrackedRepository(name="mozilla-central", url="https://github.com/mozilla-firefox/firefox.git"),
+        ],
+        branch_mappings=[BranchMapping(
+            branch_pattern = '.*',
+            source_url = "https://github.com/mozilla-firefox/firefox.git",
+            destination_url = 'destination_url',
+            destination_branch  = 'destination_branch',
+        )],
+    )
+
+@pytest.fixture
+def get_payload() -> Callable:
+
+    def get_payload(**kwargs: dict) -> dict:
+        """Return a default payload, with override via kwargs."""
+        payload = {
+            "type": "push",
+            "repo_url": "repo.git",
+            "branches": { "main": 40 * "0"},
+            "tags": {},
+            "time": 0,
+            "push_id": 0,
+            "user": "user",
+            "push_json_url": "push_json_url",
+        }
+        payload.update(kwargs)
+        return payload
+
+    return get_payload