pumpfun-bonkfun-bot_github/tests/regression/verify_create_v2_optional_args.py

"""Verify the IDL instruction decoder accepts omitted trailing optional args.

create_v2's trailing `is_cashback_enabled` OptionBool can legally be absent from
the wire: the committed blocksubscribe fixture carries a 145-byte create_v2
whose args end right after `is_mayhem_mode`. A decoder that insists on the byte
silently drops those coins for every consumer of
`parse_token_creation_from_instruction`.

Offline machine checks, no network and no funds moved:

  1. The raw fixture create_v2 (trailing OptionBool absent) decodes, with
     `is_cashback_enabled` reported as unset (None).
  2. The same data with the byte present still decodes to the OptionBool
     struct form ({"field_0": bool}) consumers already handle.
  3. Mandatory args stay enforced: truncating `is_mayhem_mode` as well must
     fail the decode, not fabricate a default.
  4. Native `option` types decode (update_buyback_config's Option<u64>):
     None tag, Some tag, and omitted-trailing forms.
  5. The pump.fun event parser turns the raw fixture instruction into a
     TokenInfo with is_cashback_coin=False and state_from_event=False.
  6. A create_v2 that omits the trailing creator_fee_bps (OptionU64) decodes
     instead of dropping the coin — the same bug, reintroduced by a new
     optional-typed arg.
  7. A create_v2 that does send creator_fee_bps reads the value back.

Usage:
    uv run tests/regression/verify_create_v2_optional_args.py
"""

import base64
import json
import struct
import sys
from pathlib import Path

import base58

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT / "src"))

from solders.transaction import VersionedTransaction  # noqa: E402

from interfaces.core import Platform  # noqa: E402
from platforms.pumpfun.event_parser import PumpFunEventParser  # noqa: E402
from utils.idl_manager import get_idl_manager  # noqa: E402
from utils.idl_parser import IDLParser  # noqa: E402

FIXTURE = (
    PROJECT_ROOT
    / "cookbook"
    / "pumpfun"
    / "decode"
    / "raw_create_tx_from_blocksubscribe.json"
)
# getTransaction fixtures — real mainnet create_v2 instructions, one omitting the
# trailing creator_fee_bps (OptionU64) and one sending it. Captured with
# encoding: json, so account keys and instruction data are base58 strings rather
# than the raw base64 VersionedTransaction the blocksubscribe fixture carries.
FIXTURE_OMITTED_FEE_BPS = (
    PROJECT_ROOT
    / "cookbook"
    / "pumpfun"
    / "decode"
    / "raw_create_v2_omitted_fee_bps_from_gettransaction.json"
)
FIXTURE_WITH_FEE_BPS = (
    PROJECT_ROOT
    / "cookbook"
    / "pumpfun"
    / "decode"
    / "raw_create_v2_with_fee_bps_from_gettransaction.json"
)
IDL_PATH = PROJECT_ROOT / "idl" / "pump_fun_idl.json"

CREATE_V2_DISCRIMINATOR = bytes.fromhex("d6904cec5f8b31b4")
UPDATE_BUYBACK_CONFIG_DISCRIMINATOR = bytes([251, 224, 171, 146, 160, 26, 113, 233])


def _parser() -> IDLParser:
    return IDLParser(str(IDL_PATH))


def _fixture_create_v2() -> tuple[bytes, list[int], list[bytes]]:
    """Raw create_v2 data, account indices and keys from the committed fixture."""
    fixture = json.loads(FIXTURE.read_text())
    raw = base64.b64decode(fixture["transaction"][0])
    message = VersionedTransaction.from_bytes(raw).message
    account_keys = [bytes(k) for k in message.account_keys]
    for ix in message.instructions:
        data = bytes(ix.data)
        if data.startswith(CREATE_V2_DISCRIMINATOR):
            return data, list(ix.accounts), account_keys
    raise ValueError("fixture has no create_v2 instruction")  # noqa: TRY003


def _fixture_create_v2_from_gettransaction(
    path: Path,
) -> tuple[bytes, list[int], list[bytes]]:
    """Raw create_v2 data, account indices and keys from a getTransaction fixture.

    Captured with `encoding: json`, so instruction data and account keys are
    base58 strings straight off the RPC response — unlike the blocksubscribe
    fixture above, there is no raw VersionedTransaction to decode.
    """
    fixture = json.loads(path.read_text())
    message = fixture["transaction"]["message"]
    account_keys = [base58.b58decode(key) for key in message["accountKeys"]]
    for ix in message["instructions"]:
        data = base58.b58decode(ix["data"])
        if data.startswith(CREATE_V2_DISCRIMINATOR):
            return data, list(ix["accounts"]), account_keys
    raise ValueError(  # noqa: TRY003
        f"{path.name} has no top-level create_v2 instruction"
    )


def _fixture_create_v2_omitted_fee_bps() -> tuple[bytes, list[int], list[bytes]]:
    """create_v2 that omits the trailing creator_fee_bps (OptionU64)."""
    return _fixture_create_v2_from_gettransaction(FIXTURE_OMITTED_FEE_BPS)


def _fixture_create_v2_with_fee_bps() -> tuple[bytes, list[int], list[bytes]]:
    """create_v2 that sends creator_fee_bps on the wire."""
    return _fixture_create_v2_from_gettransaction(FIXTURE_WITH_FEE_BPS)


def check_omitted_trailing_option_decodes() -> bool:
    """The 145-byte wire form (no is_cashback_enabled byte) must decode."""
    data, accounts, keys = _fixture_create_v2()
    decoded = _parser().decode_instruction(data, keys, accounts)
    if decoded is None:
        print(f"    decode_instruction returned None for {len(data)}-byte create_v2")
        return False
    args = decoded["args"]
    ok = (
        decoded["instruction_name"] == "create_v2"
        and args.get("is_cashback_enabled") is None
        and args.get("is_mayhem_mode") is False
        and bool(args.get("name"))
        and bool(args.get("creator"))
    )
    if not ok:
        print(f"    unexpected args: {args}")
    return ok


def check_present_trailing_option_unchanged() -> bool:
    """With the byte on the wire, the OptionBool struct form must survive."""
    data, accounts, keys = _fixture_create_v2()
    parser = _parser()
    for byte, expected in ((b"\x00", False), (b"\x01", True)):
        decoded = parser.decode_instruction(data + byte, keys, accounts)
        if decoded is None:
            print(f"    decode failed with trailing byte {byte.hex()}")
            return False
        value = decoded["args"].get("is_cashback_enabled")
        if not (isinstance(value, dict) and value.get("field_0") is expected):
            print(f"    trailing byte {byte.hex()} decoded to {value}")
            return False
    return True


def check_mandatory_args_still_enforced() -> bool:
    """Dropping is_mayhem_mode (a plain bool) must fail, not default."""
    data, accounts, keys = _fixture_create_v2()
    decoded = _parser().decode_instruction(data[:-1], keys, accounts)
    if decoded is not None:
        print(f"    truncated create_v2 decoded anyway: {decoded['args']}")
        return False
    return True


def check_native_option_decodes() -> bool:
    """update_buyback_config carries Option<u64>: None, Some and omitted forms."""
    parser = _parser()
    cases = (
        (UPDATE_BUYBACK_CONFIG_DISCRIMINATOR + b"\x00", None),
        (UPDATE_BUYBACK_CONFIG_DISCRIMINATOR + b"\x01" + struct.pack("<Q", 250), 250),
        (UPDATE_BUYBACK_CONFIG_DISCRIMINATOR, None),
    )
    for data, expected in cases:
        decoded = parser.decode_instruction(data, [], [])
        if decoded is None:
            print(f"    decode returned None for {len(data)}-byte data")
            return False
        value = decoded["args"].get("buyback_basis_points")
        if value != expected:
            print(f"    expected {expected}, got {value}")
            return False
    return True


def check_event_parser_reads_raw_fixture() -> bool:
    """parse_token_creation_from_instruction must not need the appended byte."""
    data, accounts, keys = _fixture_create_v2()
    parser = PumpFunEventParser(
        idl_parser=get_idl_manager().get_parser(Platform.PUMP_FUN)
    )
    token_info = parser.parse_token_creation_from_instruction(data, accounts, keys)
    if token_info is None:
        print("    parse_token_creation_from_instruction returned None")
        return False
    ok = (
        token_info.is_cashback_coin is False
        and getattr(token_info, "state_from_event", False) is False
    )
    if not ok:
        print(
            f"    is_cashback_coin={token_info.is_cashback_coin} "
            f"state_from_event={getattr(token_info, 'state_from_event', None)}"
        )
    return ok


def check_omitted_option_u64_decodes() -> bool:
    """create_v2 omitting creator_fee_bps must decode, not drop the coin.

    create_v2 carries two trailing optionals: creator_fee_bps (OptionU64) and
    is_holder_reward (OptionBool). Live transactions omit either or both, and
    dropping those coins is the same bug again.
    """
    parser = _parser()
    data, accounts, keys = _fixture_create_v2_omitted_fee_bps()
    decoded = parser.decode_instruction(data, keys, accounts)
    if decoded is None:
        print("  FAIL create_v2 with omitted creator_fee_bps decoded as None")
        return False
    args = decoded["args"]
    for name in ("creator_fee_bps", "is_holder_reward"):
        value = args.get(name)
        if value is not None:
            print(f"  FAIL {name} omitted but decoded as {value}, expected None")
            return False
    print(
        "  OK  omitted optionals are None: creator_fee_bps=None, is_holder_reward=None"
    )
    return True


def check_option_u64_present_decodes() -> bool:
    """A create_v2 that does send creator_fee_bps reads the value back."""
    parser = _parser()
    data, accounts, keys = _fixture_create_v2_with_fee_bps()
    decoded = parser.decode_instruction(data, keys, accounts)
    if decoded is None:
        print("  FAIL present creator_fee_bps did not decode")
        return False
    args = decoded["args"]
    creator_fee_bps = args.get("creator_fee_bps")
    is_holder_reward = args.get("is_holder_reward")
    # OptionU64 is wrapped as {"field_0": <value>}
    if not isinstance(creator_fee_bps, dict) or "field_0" not in creator_fee_bps:
        print(f"  FAIL creator_fee_bps has unexpected shape: {creator_fee_bps}")
        return False
    expected_value = creator_fee_bps["field_0"]
    if is_holder_reward is not None:
        print(f"  FAIL is_holder_reward should be None, got {is_holder_reward}")
        return False
    print(
        f"  OK  creator_fee_bps={expected_value} (wrapped as {creator_fee_bps}), "
        f"is_holder_reward=None"
    )
    return True


def main() -> int:
    checks = [
        (
            "omitted trailing OptionBool decodes as unset",
            check_omitted_trailing_option_decodes,
        ),
        (
            "present trailing OptionBool keeps struct form",
            check_present_trailing_option_unchanged,
        ),
        ("mandatory args still enforced", check_mandatory_args_still_enforced),
        ("native Option<u64> decodes", check_native_option_decodes),
        ("event parser reads the raw fixture", check_event_parser_reads_raw_fixture),
        (
            "create_v2 omitting creator_fee_bps decodes",
            check_omitted_option_u64_decodes,
        ),
        ("create_v2 with creator_fee_bps decodes", check_option_u64_present_decodes),
    ]
    failed = 0
    for label, check in checks:
        try:
            ok = check()
        except Exception as error:  # noqa: BLE001 - report and continue
            print(f"FAIL {label}: {type(error).__name__}: {error}")
            failed += 1
            continue
        print(f"{'PASS' if ok else 'FAIL'} {label}")
        failed += 0 if ok else 1
    print(f"\n{len(checks) - failed}/{len(checks)} checks passed")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())