plm-lottery/tests/unit/fake_network.py

"""Test doubles for what the services ask of the network (B-76 and after)."""

from app.electrum.client import ElectrumRpcError


def rpc_error(message: str, code: int = -26) -> ElectrumRpcError:
    """A server that answered with a JSON-RPC `error` object — an explicit refusal."""
    return ElectrumRpcError({"code": code, "message": message})


class FakeNetwork:
    """The ElectrumListener surface a broadcast needs: `known` is what
    `transaction_known` answers (True: some server has it, False: every server denies
    it, None: too few answers to tell); fanned-out transactions are recorded."""

    def __init__(self, known: bool | None = False, proven: bool = True):
        self.known = known
        self.proven = proven  # what `confirmation_proven` answers (B-83)
        self.asked: list[str] = []
        self.fanned_out: list[str] = []
        self.proofs_asked: list[tuple[str, int]] = []

    async def transaction_known(self, txid: str) -> bool | None:
        self.asked.append(txid)
        return self.known

    def fan_out_transaction(self, raw_tx_hex: str) -> None:
        self.fanned_out.append(raw_tx_hex)

    async def confirmation_proven(self, txid: str, height: int) -> bool:
        self.proofs_asked.append((txid, height))
        return self.proven


MISSING_INPUTS = "bad-txns-inputs-missingorspent"


class ScriptedNetwork(FakeNetwork):
    """What the reconciler asks of the listener (B-78), scripted per test.

    `active` and `others` are the history answers of the active server and of each other
    server — a list of {"tx_hash", "height"} entries, or None for a server that did not
    answer. `spender` is what `confirmed_spender` returns (the txid of a mined
    transaction spending the row's inputs, or None). `rebroadcast` is the
    (active, others) pair of verdicts `rebroadcast_everywhere` returns: ("accepted", ""),
    ("refused", message) or None."""

    def __init__(self, *, active=(), others=((), ()), spender=None, rebroadcast=None, others_needed=2, proven=True):
        super().__init__(known=False, proven=proven)
        self.active = None if active is None else list(active)
        self.others = tuple(None if o is None else list(o) for o in others)
        self.spender = spender
        self.rebroadcast = rebroadcast
        self.others_needed = others_needed
        self.history_calls = 0
        self.rebroadcast_calls: list[str] = []

    async def get_histories(self, scripthash):
        from app.electrum.quorum import ServerAnswers

        self.history_calls += 1
        return ServerAnswers(self.active, self.others, self.others_needed)

    async def confirmed_spender(self, scripthash, own_txids, outpoints):
        return self.spender

    async def rebroadcast_everywhere(self, raw_tx_hex):
        from app.electrum.quorum import ServerAnswers

        self.rebroadcast_calls.append(raw_tx_hex)
        active, others = self.rebroadcast if self.rebroadcast is not None else (None, (None, None))
        return ServerAnswers(active, tuple(others), self.others_needed)

    @classmethod
    def proof_of_absence(cls, message: str = MISSING_INPUTS, **kwargs):
        """Nobody of the quorum knows the tx and every server refuses it for missing inputs."""
        refusal = ("refused", message)
        return cls(rebroadcast=(refusal, (refusal, refusal)), **kwargs)