Testing Conflict Resolvers with Synthetic Conflicts #

Your conflict resolver passes every unit test against hand-built dictionaries, then loses data the first time a real fork hits it — because a fork built in memory never exercised the parts that matter: the _conflicts array CouchDB computes, the winner-selection tiebreak, and the tombstone writes. To test resolver code honestly you have to make CouchDB hold a genuinely conflicted document on demand, deterministically, in a test. The reliable way to do that without waiting on real replication is to inject a second leaf directly with POST /db/_bulk_docs and {"new_edits": false}, handing CouchDB your own _rev values so two revisions of the same generation coexist as a fork. This page builds a pytest harness around a make_conflict() helper that plants that fork, asserts ?conflicts=true shows it, runs your resolver, and asserts the document converges. It is the safety net for the auto-merge rule engines you deploy, and it exercises the exact resolver functions built in Writing Custom Conflict Resolver Functions in Python.

Immediate Triage / Prerequisites #

If your resolver’s tests build “conflicts” as plain Python lists and never touch CouchDB, they are testing your merge arithmetic but not your resolver — the winner-selection order, the _conflicts computation, and the tombstone _bulk_docs write are all untested. The fastest way to see the gap is to point an existing test at a real database and watch it fail to find any conflict at all:

# a genuinely forked doc reports a non-empty _conflicts array; an in-memory fake never will
curl -s "http://localhost:5984/test_conflicts/order-1?conflicts=true" | \
  python3 -c "import sys,json; print('conflicts:', json.load(sys.stdin).get('_conflicts', []))"

Prerequisites: Python 3.8+, pytest and requests (pip install pytest requests), and a CouchDB 3.x you can create and delete throwaway databases on — a local container is ideal (docker run -p 5984:5984 -e COUCHDB_USER=admin -e COUCHDB_PASSWORD=admin couchdb:3). The key primitive is new_edits: a normal write has new_edits implicitly true, meaning CouchDB assigns the next _rev. Setting {"new_edits": false} on _bulk_docs tells CouchDB to store the revisions exactly as given — you supply the _rev and the _revisions ancestry, and it inserts them into the tree without generating anything. That is what lets you plant two sibling leaves and manufacture a fork deterministically. The tree shape you are building is described in revision tree mechanics.

Step-by-Step Implementation #

Follow these steps to build a fixture that plants a fork, verifies CouchDB sees it, and asserts your resolver converges the document. Each step includes a check.

1. Write the base revision normally. Create the document with an ordinary PUT (or _bulk_docs with new_edits defaulted true) and capture its _rev — this becomes the shared ancestor both conflicting leaves will point at.

   r = session.put(f"{db}/order-1", json={"type": "order", "qty": 1})
   base_rev = r.json()["rev"]         # e.g. "1-aaa..."; the fork's common parent
   assert r.status_code in (201, 202)

   Verify the document has no conflicts yet: a read with ?conflicts=true must return no _conflicts key.

2. Plant a conflicting leaf with new_edits:false. Post two generation-2 documents that share order-1 but carry different hand-crafted revision IDs and a _revisions history naming the shared parent. Because new_edits is false, CouchDB stores both verbatim as siblings:

   base_gen, base_hash = base_rev.split("-")
   def leaf(hash_, **body):
       return {"_id": "order-1", "_rev": f"2-{hash_}",
               "_revisions": {"start": 2, "ids": [hash_, base_hash]}, **body}
   body = {"docs": [leaf("bbb", qty=5), leaf("ccc", qty=9)], "new_edits": False}
   resp = session.post(f"{db}/_bulk_docs", json=body)
   assert resp.status_code in (201, 202)

   The generation must be exactly one deeper than the parent and _revisions.ids[-1] must equal the parent’s hash, or CouchDB will store an orphan leaf instead of a fork. Every hash here is a valid hex string you control, so the result is fully deterministic across runs.

3. Assert CouchDB reports the fork. Read the document with ?conflicts=true and assert both that a winner is returned and that the losing leaf appears in _conflicts. This confirms you built a real conflict, not two unrelated writes:

   doc = session.get(f"{db}/order-1", params={"conflicts": "true"}).json()
   assert doc["_conflicts"], "no fork was planted"
   assert doc["_rev"] in ("2-ccc", "2-bbb")   # highest hash wins the default read

4. Run the resolver and assert convergence. Invoke the resolver under test against order-1, then re-read and assert the document has no _conflicts and a single leaf remains. Convergence — not just a written winner — is the real assertion, because a resolver that writes a winner but forgets to tombstone the loser leaves the document still forked. That failure mode and its metrics are covered in Conflict-Resolution Metrics & SLOs for CouchDB Sync.

Complete Working Example #

The pytest module below is self-contained and runnable against a live CouchDB. It provides a db fixture that creates and drops a throwaway database per test (isolation), a make_conflict() helper that plants a base plus any number of conflicting leaves with new_edits:false, and tests that assert the fork exists and that a sample resolver converges it — including a tombstone case.

import uuid

import pytest
import requests

COUCH = "http://admin:admin@localhost:5984"


def _resolver_highest_qty(db_url: str, doc_id: str, session: requests.Session):
    """Sample resolver: keep the branch with the largest qty, tombstone the rest."""
    doc = session.get(f"{db_url}/{doc_id}", params={"conflicts": "true"}).json()
    losers = doc.get("_conflicts", [])
    if not losers:
        return
    branches = [doc] + [
        session.get(f"{db_url}/{doc_id}", params={"rev": rev}).json() for rev in losers
    ]
    winner = max(branches, key=lambda d: d.get("qty", 0))
    merged = {"_id": doc_id, "_rev": doc["_rev"], "type": winner.get("type"),
              "qty": winner["qty"]}                       # rebase onto current winner _rev
    batch = [merged] + [{"_id": doc_id, "_rev": r, "_deleted": True} for r in losers]
    session.post(f"{db_url}/_bulk_docs", json={"docs": batch}).raise_for_status()


@pytest.fixture()
def db():
    """Create an isolated throwaway database, yield its URL, then delete it."""
    name = f"test_conflicts_{uuid.uuid4().hex[:8]}"       # unique per test -> no shared state
    url = f"{COUCH}/{name}"
    requests.put(url).raise_for_status()
    session = requests.Session()
    try:
        yield url, session
    finally:
        requests.delete(url)                              # cleanup even if the test fails


def make_conflict(db_url, session, doc_id, base_body, leaves):
    """Plant a base revision then N conflicting generation-2 leaves via new_edits:false.

    `leaves` is a list of (hash, body) tuples; each becomes a sibling leaf whose
    parent is the base revision, so CouchDB records a genuine fork.
    """
    r = session.put(f"{db_url}/{doc_id}", json=base_body)
    r.raise_for_status()
    _, base_hash = r.json()["rev"].split("-")
    docs = []
    for hash_, body in leaves:
        docs.append({"_id": doc_id, "_rev": f"2-{hash_}",
                     "_revisions": {"start": 2, "ids": [hash_, base_hash]}, **body})
    resp = session.post(f"{db_url}/_bulk_docs",
                        json={"docs": docs, "new_edits": False})   # store revs verbatim
    resp.raise_for_status()
    return base_hash


def test_make_conflict_reports_fork(db):
    db_url, session = db
    make_conflict(db_url, session, "order-1", {"type": "order", "qty": 1},
                  [("bbb", {"type": "order", "qty": 5}),
                   ("ccc", {"type": "order", "qty": 9})])
    doc = session.get(f"{db_url}/order-1", params={"conflicts": "true"}).json()
    assert doc["_conflicts"], "expected a planted conflict"
    assert len(doc["_conflicts"]) == 1                    # one winner + one loser leaf


def test_resolver_converges_document(db):
    db_url, session = db
    make_conflict(db_url, session, "order-1", {"type": "order", "qty": 1},
                  [("bbb", {"type": "order", "qty": 5}),
                   ("ccc", {"type": "order", "qty": 9})])
    _resolver_highest_qty(db_url, "order-1", session)
    resolved = session.get(f"{db_url}/order-1", params={"conflicts": "true"}).json()
    assert "_conflicts" not in resolved                   # convergence, not just a write
    assert resolved["qty"] == 9                            # highest-qty branch won


def test_resolver_handles_tombstone_branch(db):
    """A deleted (tombstoned) losing leaf must still be cleared, not resurrected."""
    db_url, session = db
    make_conflict(db_url, session, "order-1", {"type": "order", "qty": 1},
                  [("bbb", {"type": "order", "qty": 5, "_deleted": True}),
                   ("ccc", {"type": "order", "qty": 9})])
    _resolver_highest_qty(db_url, "order-1", session)
    resolved = session.get(f"{db_url}/order-1", params={"conflicts": "true"}).json()
    assert "_conflicts" not in resolved
    assert resolved["qty"] == 9


if __name__ == "__main__":
    raise SystemExit(pytest.main([__file__, "-v"]))

Run it with pytest -v against a CouchDB on localhost:5984. Each test gets its own database, plants a deterministic fork, and asserts convergence — including the tombstone branch, where one of the planted leaves is already _deleted and the resolver must still clear the conflict rather than resurrect it.

Gotchas & Edge Cases #

• new_edits:false requires a valid generation and ancestry. The _rev you plant must be exactly one generation deeper than the parent, and _revisions.ids must list the new hash first and the parent hash last. Get the generation wrong and CouchDB stores a disconnected leaf that does not read as a conflict of your base — your assertion then fails for a confusing reason. It also silently accepts revisions you supply, so a typo produces a different tree than you intended, not an error.
• Isolate databases, do not just delete documents. Sharing one database across tests leaks revision trees: a leftover _conflicts array or a tombstone from a prior test bleeds into the next and makes failures order-dependent. Create a uniquely named database per test and drop it in teardown, as the fixture does; it is cheaper than debugging cross-test contamination.
• Test the tombstone path explicitly. Real forks routinely include a deleted branch (a client deleted the document while another edited it). Plant a leaf with _deleted: true and assert the resolver still converges — a resolver that only reads live branches can miss that a losing leaf is a tombstone and leave it dangling.
• Beware flakiness from shared or reused IDs. If two tests use the same document ID against the same database, or you reuse hashes like bbb across a database that was not dropped, CouchDB may treat a replant as an already-known revision and store nothing. Unique database names plus fixed hashes within a test give determinism without cross-test collisions.
• CouchDB in CI must be reachable and clean. Run it as a service container and gate the suite on a GET /_up readiness check before the first test, so a slow-starting node fails fast with a clear message instead of a wall of connection errors. Seeding via two isolated databases and a real replication is a valid alternative for end-to-end coverage, but it is slower and less deterministic than new_edits:false for unit scope.

Verification & Observability #

Confirm the harness itself is trustworthy before you rely on it to gate deploys. First, prove the planted fork is real at the API level, independent of your assertions:

# every live leaf of the planted doc, including conflicting branches
curl -s "http://admin:admin@localhost:5984/test_conflicts_xyz/order-1?open_revs=all" | \
  python3 -c "import sys,json; print('leaves:', len(json.load(sys.stdin)))"

Two or more leaves confirm make_conflict() forked the tree; one leaf means the plant failed and every downstream assertion is meaningless. In CI, assert that each test database is gone after the run (GET /_all_dbs should list none of your test_conflicts_* names) so a crashed teardown cannot slowly fill the node. Track the suite’s own conflict-injection count and resolver pass rate over time — a resolver whose synthetic-conflict tests start flaking usually points at a nondeterministic merge (e.g. relying on wall-clock time), which is exactly the class of bug this harness exists to catch before it reaches the auto-merge rule engines in production.

FAQ #

Related #

• Writing Custom Conflict Resolver Functions in Python
• Writing Declarative Merge Rules in YAML
• Conflict-Resolution Metrics & SLOs for CouchDB Sync

Part of: Auto-Merge Rule Engines