|
| 1 | +"""Live integration: leader-redirect-on-connect against a real cluster. |
| 2 | +
|
| 3 | +End-to-end coverage for the ``_resolve_leader`` step added to |
| 4 | +``_build_and_connect``. Mocked unit coverage is in |
| 5 | +``tests/test_leader_redirect_on_connect.py``; this file pins the |
| 6 | +behaviour against the live cluster so a regression in the wire |
| 7 | +layer or in ``ClusterClient.find_leader`` would surface here even |
| 8 | +if the unit tests stay green. |
| 9 | +
|
| 10 | +Tests use ``cluster_control`` from ``dqlitetestlib`` (bootstrapped |
| 11 | +in the top-level ``tests/conftest.py``) where they need |
| 12 | +deterministic leader manipulation. Each test that mutates cluster |
| 13 | +topology restores the original leader on its way out so subsequent |
| 14 | +tests in the session see a stable starting state. |
| 15 | +""" |
| 16 | + |
| 17 | +from __future__ import annotations |
| 18 | + |
| 19 | +import contextlib |
| 20 | +import os |
| 21 | +from typing import TYPE_CHECKING |
| 22 | + |
| 23 | +import pytest |
| 24 | + |
| 25 | +from dqlitedbapi import connect |
| 26 | +from dqlitedbapi.exceptions import OperationalError |
| 27 | + |
| 28 | +if TYPE_CHECKING: |
| 29 | + from dqlitetestlib import TestClusterControl # type: ignore[import-not-found] |
| 30 | + |
| 31 | + |
| 32 | +def _node_addresses() -> list[str]: |
| 33 | + """Read DQLITE_TEST_CLUSTER_NODES with the python-dqlite-dev |
| 34 | + default.""" |
| 35 | + raw = os.environ.get( |
| 36 | + "DQLITE_TEST_CLUSTER_NODES", |
| 37 | + "localhost:9001,localhost:9002,localhost:9003", |
| 38 | + ) |
| 39 | + return [s.strip() for s in raw.split(",") if s.strip()] |
| 40 | + |
| 41 | + |
| 42 | +# --- normal flows --- |
| 43 | + |
| 44 | + |
| 45 | +@pytest.mark.integration |
| 46 | +def test_connect_with_seed_as_leader_succeeds() -> None: |
| 47 | + """Happy path: the seed address is the current leader. |
| 48 | + ``_resolve_leader`` returns the seed verbatim and the OPEN |
| 49 | + runs against it.""" |
| 50 | + seed = os.environ.get("DQLITE_TEST_CLUSTER", "localhost:9001") |
| 51 | + conn = connect(seed, timeout=5.0) |
| 52 | + try: |
| 53 | + cur = conn.cursor() |
| 54 | + cur.execute("SELECT 1") |
| 55 | + assert cur.fetchall() == [(1,)] |
| 56 | + finally: |
| 57 | + conn.close() |
| 58 | + |
| 59 | + |
| 60 | +@pytest.mark.integration |
| 61 | +def test_connect_via_follower_address_redirects_to_leader() -> None: |
| 62 | + """The seed is a follower; ``_resolve_leader`` follows the |
| 63 | + redirect and OPEN runs against the actual leader. Without |
| 64 | + leader-redirect-on-connect this would fail with |
| 65 | + ``SQLITE_IOERR_NOT_LEADER`` from the follower's OPEN handler. |
| 66 | +
|
| 67 | + Picks the non-leader nodes from ``DQLITE_TEST_CLUSTER_NODES`` |
| 68 | + and connects through each — at least two of the three nodes |
| 69 | + are followers in the steady state, so this is a robust pin. |
| 70 | + """ |
| 71 | + import asyncio |
| 72 | + |
| 73 | + from dqliteclient.cluster import ClusterClient |
| 74 | + from dqliteclient.node_store import MemoryNodeStore |
| 75 | + |
| 76 | + addresses = _node_addresses() |
| 77 | + |
| 78 | + async def _resolve() -> str: |
| 79 | + store = MemoryNodeStore(addresses) |
| 80 | + cluster = ClusterClient(store, timeout=5.0) |
| 81 | + return await cluster.find_leader() |
| 82 | + |
| 83 | + leader_addr = asyncio.run(_resolve()) |
| 84 | + follower_addrs = [a for a in addresses if a != leader_addr] |
| 85 | + assert follower_addrs, ( |
| 86 | + f"expected at least one follower in {addresses!r}; leader is {leader_addr!r}" |
| 87 | + ) |
| 88 | + |
| 89 | + for follower in follower_addrs: |
| 90 | + conn = connect(follower, timeout=5.0) |
| 91 | + try: |
| 92 | + cur = conn.cursor() |
| 93 | + cur.execute("SELECT 1") |
| 94 | + assert cur.fetchall() == [(1,)] |
| 95 | + finally: |
| 96 | + conn.close() |
| 97 | + |
| 98 | + |
| 99 | +@pytest.mark.integration |
| 100 | +def test_connect_after_leader_flip_routes_to_new_leader( |
| 101 | + cluster_control: TestClusterControl, |
| 102 | +) -> None: |
| 103 | + """Force a leader flip; a brand-new ``connect()`` against the |
| 104 | + OLD-leader address (now a follower) succeeds because |
| 105 | + ``_resolve_leader`` follows the redirect to the new leader. |
| 106 | + Restores the original leader on the way out.""" |
| 107 | + import asyncio |
| 108 | + |
| 109 | + starting = asyncio.run(cluster_control.current_leader_node()) |
| 110 | + seed = starting.address.replace("127.0.0.1", "localhost") |
| 111 | + |
| 112 | + flip = asyncio.run(cluster_control.force_leader_flip()) |
| 113 | + assert flip.target.node_id != starting.node_id |
| 114 | + |
| 115 | + try: |
| 116 | + # The seed is now a follower (the demoted ex-leader). |
| 117 | + # ``_resolve_leader`` should follow the redirect and the |
| 118 | + # OPEN should reach the new leader. |
| 119 | + conn = connect(seed, timeout=5.0) |
| 120 | + try: |
| 121 | + cur = conn.cursor() |
| 122 | + cur.execute("SELECT 1") |
| 123 | + assert cur.fetchall() == [(1,)] |
| 124 | + finally: |
| 125 | + conn.close() |
| 126 | + finally: |
| 127 | + with contextlib.suppress(Exception): |
| 128 | + asyncio.run(cluster_control.transfer_leadership_to(starting.node_id)) |
| 129 | + |
| 130 | + |
| 131 | +# --- failure flows --- |
| 132 | + |
| 133 | + |
| 134 | +@pytest.mark.integration |
| 135 | +def test_connect_to_unreachable_seed_raises_operational_error() -> None: |
| 136 | + """Seed unreachable: ``_resolve_leader`` cannot reach any node |
| 137 | + in its 1-node store; surfaces as ``OperationalError`` with the |
| 138 | + canonical ``Failed to find leader from`` prefix.""" |
| 139 | + # Pick a port we know nothing is listening on. |
| 140 | + with pytest.raises(OperationalError, match="Failed to find leader"): |
| 141 | + conn = connect("127.0.0.1:1", timeout=1.0) |
| 142 | + conn.connect() # explicit connect for clearer failure point |
0 commit comments