AI disclosure: This article and its code example were generated by an AI agent for Autom Mate. The example was executed locally with the automated checks described below. It is an illustrative tutorial, not a report of a production deployment or a claim of human review.
A two-way ticket integration can fail even when every HTTP request succeeds. A delayed event can restore an old priority. A duplicated delivery can trigger repeated work. An internal note can appear in the wrong system.
Before building a complete connector, test a smaller boundary: which incoming information is allowed to update the integration's local view? A deterministic function makes that policy easier to inspect than a large workflow with several remote side effects.
This example extends Autom Mate's ticket integration test matrix with a runnable Python exercise. It filters snapshots into a local SQLite projection. It sends nothing to Jira or another application.
Write the event contract first
The example assumes that each event contains a full snapshot of the fields owned by its source, plus a strictly increasing version for that source's ticket. It uses separate namespaces for the customer tenant, source system and ticket.
Those assumptions are explicit because real webhook mechanisms differ. Atlassian documents a Jira Cloud webhook identifier that stays the same across retries and is unique within a tenant. That is useful for delivery deduplication, but it does not establish the snapshot version contract used here. See the Jira Cloud webhook documentation.
If your events contain partial patches, discarding an older patch may lose an independent field update. Use an appropriate ordering, per-field versioning or authoritative reconciliation design instead of copying this reducer unchanged.
Separate delivery, ordering and ownership
Three questions need separate answers:
| Question | Local rule in this example |
|---|---|
| Have we processed this delivery before? | Check tenant, source and event identifier together |
| Is its snapshot newer? | Compare versions within that tenant, source and ticket |
| Which fields may travel? | Apply the source's explicit field allowlist |
| Does this prove remote delivery? | No; the example updates only a local projection |
The desk owns priority and service_status. Engineering owns development_status. A field named internal_note is excluded from both projections.
“Development done” may still require deployment or customer confirmation.
Run the filter and its checks
Save this as ticket_snapshot_demo.py:
"""Local snapshot filter: no webhooks, credentials, or remote writes."""
import json
import sqlite3
import unittest
ALLOWED = {
"desk": {"priority", "service_status"},
"engineering": {"development_status"},
}
def connect():
db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE seen (
tenant TEXT, source TEXT, event_id TEXT,
PRIMARY KEY (tenant, source, event_id));
CREATE TABLE projection (
tenant TEXT, source TEXT, ticket TEXT, version INTEGER, data TEXT,
PRIMARY KEY (tenant, source, ticket));
""")
return db
def apply_snapshot(db, event):
tenant, source = event["tenant"], event["source"]
if source not in ALLOWED or type(event["version"]) is not int:
raise ValueError("unsupported source or invalid version")
key = (tenant, source, event["ticket"])
with db:
duplicate = db.execute(
"SELECT 1 FROM seen WHERE tenant=? AND source=? AND event_id=?",
(tenant, source, event["id"]),
).fetchone()
if duplicate:
return "duplicate"
prior = db.execute(
"SELECT version FROM projection WHERE tenant=? AND source=? AND ticket=?",
key,
).fetchone()
result = "stale"
if prior is None or event["version"] > prior[0]:
safe = {k: v for k, v in event["fields"].items() if k in ALLOWED[source]}
db.execute("INSERT OR REPLACE INTO projection VALUES (?, ?, ?, ?, ?)",
(*key, event["version"], json.dumps(safe)))
result = "applied"
db.execute("INSERT INTO seen VALUES (?, ?, ?)",
(tenant, source, event["id"]))
return result
def read(db, tenant, source, ticket):
row = db.execute(
"SELECT data FROM projection WHERE tenant=? AND source=? AND ticket=?",
(tenant, source, ticket),
).fetchone()
return json.loads(row[0]) if row else None
class SnapshotTests(unittest.TestCase):
def test_duplicate_stale_and_private_fields(self):
db = connect()
e = dict(tenant="A", source="desk", ticket="7", id="event-2",
version=2, fields={"priority": "high", "internal_note": "test only"})
self.assertEqual(apply_snapshot(db, e), "applied")
self.assertEqual(apply_snapshot(db, e), "duplicate")
old = dict(e, id="event-1", version=1, fields={"priority": "low"})
self.assertEqual(apply_snapshot(db, old), "stale")
self.assertEqual(read(db, "A", "desk", "7"), {"priority": "high"})
db.close()
def test_tenant_and_source_namespaces_are_separate(self):
db = connect()
e = dict(tenant="A", source="desk", ticket="7", id="same-id",
version=4, fields={"priority": "high"})
apply_snapshot(db, e)
other = dict(e, tenant="B", fields={"priority": "low"})
self.assertEqual(apply_snapshot(db, other), "applied")
eng = dict(e, source="engineering", version=1,
fields={"development_status": "done", "priority": "low"})
self.assertEqual(apply_snapshot(db, eng), "applied")
self.assertEqual(read(db, "A", "desk", "7"), {"priority": "high"})
self.assertEqual(read(db, "B", "desk", "7"), {"priority": "low"})
self.assertEqual(read(db, "A", "engineering", "7"),
{"development_status": "done"})
db.close()
if __name__ == "__main__":
unittest.main()
Run:
python ticket_snapshot_demo.py -v
The first test applies version two, repeats the delivery and then delivers version one. It verifies that the repeated event is recognised, the old snapshot cannot regress priority and the internal note is not stored in the projection.
The second test uses identical event and ticket identifiers across two tenants and two sources. Both namespaces must remain independent. Engineering's attempt to supply a desk-owned priority is filtered out.
Both tests passed with Python 3.12.14 during preparation. They check concrete state rather than merely asserting that a function returned without an exception.
Understand the transaction boundary
The local projection change and the “seen” record are committed together using the connection context manager. Python's sqlite3 documentation explains transaction handling and parameterised statements.
There is no remote call inside that boundary. If you later add an HTTP write after the commit, a process failure can occur between recording the event and delivering the change. If you write before the commit, the destination can accept the change while the local transaction later fails.
A production design needs a deliberate outbound delivery mechanism, such as durable queued work with destination-aware duplicate handling. This exercise does not claim exactly-once delivery across systems.
The code is also a single-consumer teaching example. It does not test concurrent transactions, database contention, persistent restarts or retention of delivery identifiers. Those require their own design and tests.
Test loops at the outbound boundary
Deduplicating an inbound event will not necessarily stop an echo loop. A remote update may create a new event identifier even though it reflects a change the bridge just sent.
Record enough outbound context to recognise that specific reflected change. Compare the relevant fields, origin and version before deciding to ignore it. Avoid dropping every event created by an integration account: that account may legitimately perform work for another service.
A useful integration test sends one allowed field change, captures the resulting destination event and verifies that processing it does not issue the same business change repeatedly. Also send a genuine new change from the same account; the loop control must not discard it blindly.
Extend the acceptance matrix beyond this sample
Before connecting real systems, add tests for:
- Invalid or unauthenticated senders, including a forged tenant identifier.
- Private attachments and comments, not only the sample internal-note field.
- Missing events discovered through reconciliation rather than customer complaints.
- A moved or deleted destination ticket.
- Equal versions carrying different payloads, which this simplified example treats as stale.
- A remote permission change after the local event has been accepted.
Each failed or unproven case needs a visible owner. Passing the local tests is useful evidence for the filter's stated rules, not acceptance of the entire service.
Use the wider ticket integration matrix to plan the remaining service checks.
Top comments (0)