#!/usr/bin/env python3
# Copyright (c) 2020 The Bitcoin Unlimited developers
"""
Tests the electrum call 'blockchain.transaction.get'
"""

import asyncio
from test_framework.nex.nodemessages import TxOut
from test_framework.util import assert_equal
from test_framework.electrumutil import ElectrumTestFramework
from test_framework.serialize import to_hex
from test_framework.utiltx import pad_tx
from test_framework.script import (
    OP_0,
    CScript,
    OP_CHECKSIG,
    OP_DROP,
    OP_DUP,
    OP_EQUAL,
    OP_EQUALVERIFY,
    OP_FALSE,
    OP_HASH160,
)
from test_framework.constants import COIN
from test_framework.environment import (
    on_bch,
    on_nex,
    node,
    NodeFeature,
    node_supports,
)
from test_framework.electrumconnection import ElectrumConnection

if on_bch():
    from test_framework.bch.blocktools import create_transaction
    from test_framework.bch.script import anyonecanspend_unlocking_script
elif on_nex():
    from test_framework.nex.blocktools import create_transaction
    from test_framework.nex.script import anyonecanspend_unlocking_script
else:
    raise NotImplementedError()


TX_GET = "blockchain.transaction.get"
DUMMY_HASH = 0x1111111111111111111111111111111111111111


class ElectrumTransactionGet(ElectrumTestFramework):

    async def test_bch(self, cli, n, coinbases):
        fee = 300
        prevtx = coinbases[0]
        nonstandard_tx = create_transaction(
            prevtx=prevtx,
            value=prevtx.vout[0].n_value - fee,
            n=0,
            sig=anyonecanspend_unlocking_script(),
            out=CScript([OP_FALSE, OP_DROP]),
        )

        prevtx = coinbases[1]
        p2sh_tx = create_transaction(
            prevtx=prevtx,
            value=prevtx.vout[0].n_value - fee,
            n=0,
            sig=anyonecanspend_unlocking_script(),
            out=CScript([OP_HASH160, DUMMY_HASH, OP_EQUAL]),
        )

        prevtx = coinbases[2]
        p2pkh_tx = create_transaction(
            prevtx=prevtx,
            value=prevtx.vout[0].n_value - fee,
            n=0,
            sig=anyonecanspend_unlocking_script(),
            out=CScript([OP_DUP, OP_HASH160, DUMMY_HASH, OP_EQUALVERIFY, OP_CHECKSIG]),
        )
        prevtx = coinbases[3]
        unconfirmed_tx = create_transaction(
            prevtx=prevtx,
            value=prevtx.vout[0].n_value - fee,
            n=0,
            sig=anyonecanspend_unlocking_script(),
            out=CScript([OP_DUP, OP_HASH160, DUMMY_HASH, OP_EQUALVERIFY, OP_CHECKSIG]),
        )

        for tx in [nonstandard_tx, p2sh_tx, p2pkh_tx, unconfirmed_tx]:
            pad_tx(tx)

        await self.mine_blocks(cli, n, 1, [nonstandard_tx, p2sh_tx, p2pkh_tx])

        n.sendrawtransaction(to_hex(unconfirmed_tx))
        await self.wait_for_mempool_tx(cli, unconfirmed_tx.get_rpc_hex_id())

        await self.test_verbose(
            n,
            cli,
            nonstandard_tx=nonstandard_tx.get_rpc_hex_id(),
            p2sh_tx=p2sh_tx.get_rpc_hex_id(),
            p2pkh_tx=p2pkh_tx.get_rpc_hex_id(),
            unconfirmed_tx=unconfirmed_tx.get_rpc_hex_id(),
        )
        await self.test_non_verbose(cli, coinbases, unconfirmed_tx)
        await self.test_input_amounts(
            cli, coinbases[0].get_rpc_hex_id(), unconfirmed_tx.get_rpc_hex_id()
        )
        await self.test_by_txidem(cli, p2sh_tx)

    async def test_nex(self, cli, n, coinbases):
        fee = 300
        prevtx = coinbases[0]
        p2st_tx = create_transaction(
            prevtx=prevtx,
            value=prevtx.vout[0].n_value - fee,
            n=0,
            sig=anyonecanspend_unlocking_script(),
            out=CScript([OP_0, DUMMY_HASH, OP_0]),
        )

        prevtx = coinbases[1]
        p2pkh_tx = create_transaction(
            prevtx=prevtx,
            value=prevtx.vout[0].n_value - fee,
            n=0,
            sig=anyonecanspend_unlocking_script(),
            out=CScript([OP_DUP, OP_HASH160, DUMMY_HASH, OP_EQUALVERIFY, OP_CHECKSIG]),
            out_type=TxOut.TYPE_SATOSCRIPT,
        )
        prevtx = coinbases[2]
        unconfirmed_tx = create_transaction(
            prevtx=prevtx,
            value=prevtx.vout[0].n_value - fee,
            n=0,
            sig=anyonecanspend_unlocking_script(),
            out=CScript([OP_DUP, OP_HASH160, DUMMY_HASH, OP_EQUALVERIFY, OP_CHECKSIG]),
            out_type=TxOut.TYPE_SATOSCRIPT,
        )

        for tx in [p2st_tx, p2pkh_tx, unconfirmed_tx]:
            pad_tx(tx)

        await self.mine_blocks(cli, n, 1, [p2st_tx, p2pkh_tx])
        n.sendrawtransaction(to_hex(unconfirmed_tx))
        await self.wait_for_mempool_tx(cli, unconfirmed_tx.get_rpc_hex_id())

        await self.test_verbose(
            n,
            cli,
            p2st_tx=p2st_tx.get_rpc_hex_id(),
            p2pkh_tx=p2pkh_tx.get_rpc_hex_id(),
            unconfirmed_tx=unconfirmed_tx.get_rpc_hex_id(),
        )
        await self.test_non_verbose(cli, coinbases, unconfirmed_tx)
        await self.test_input_amounts(
            cli, coinbases[0].get_rpc_hex_id(), unconfirmed_tx.get_rpc_hex_id()
        )
        await self.test_by_txidem(cli, p2st_tx)

    async def run_test(self):
        n = self.nodes[0]

        await self.bootstrap_p2p()
        cli = ElectrumConnection()
        await cli.connect()
        coinbases = await self.mine_blocks(cli, n, 104)

        if on_bch():
            await self.test_bch(cli, n, coinbases)
        elif on_nex():
            await self.test_nex(cli, n, coinbases)
        else:
            raise NotImplementedError()

        if node_supports(node(), NodeFeature.TOKENS):
            await self.test_token(n, cli)
        else:
            print(f"Skipping token tests for {node()}")

        cli.disconnect()

    async def test_non_verbose(self, cli, coinbases, unconfirmed):
        for tx in coinbases + [unconfirmed]:
            assert_equal(to_hex(tx), await cli.call(TX_GET, tx.get_rpc_hex_id()))

    async def test_input_amounts(self, cli, coinbase_tx, unconfirmed_tx):
        """
        Test that input amount are included for vin.
        """

        # Nex coinbases don't have inputs.
        if not on_nex():
            # Coinbase does not have a input amount
            tx = await cli.call(TX_GET, coinbase_tx, True)
            assert_equal(None, tx["vin"][0]["value_satoshi"])
            assert_equal(None, tx["vin"][0]["value_coin"])

        # This tx spends a coinbase transaction
        if on_nex():
            coinbase_reward = 10000000
        else:
            coinbase_reward = 50
        tx = await cli.call(TX_GET, unconfirmed_tx, True)
        assert_equal(coinbase_reward, tx["vin"][0]["value_coin"])
        assert_equal(coinbase_reward * COIN, tx["vin"][0]["value_satoshi"])

    # pylint: disable=too-many-locals
    async def test_token(self, n, cli):
        # We use the node to generate token transactions, so
        # we need to give it a coin
        n.generate(101)

        def has_token_data(x, input_or_output):
            if on_bch():
                return "tokenData" in x and not x["tokenData"] is None
            if on_nex():
                if input_or_output == "input":
                    return "group" in x and not x["group"] is None

                assert input_or_output == "output"
                return (
                    "group" in x["scriptPubKey"]
                    and not x["scriptPubKey"]["group"] is None
                )
            raise Exception("NYI")

        token_id, mint_txid = await self.create_token(
            to_addr=n.getnewaddress(), mint_amount=42
        )
        await self.wait_for_mempool_tx(cli, mint_txid, n=n)

        send_txid = await self.send_token(token_id, n.getnewaddress(), 42)
        await self.wait_for_mempool_tx(cli, send_txid, n=n)

        sendtx_node = n.getrawtransaction(send_txid, True)
        sendtx_rostrum = await cli.call(TX_GET, send_txid, True)

        if on_bch():
            # we know the token input is at index 0 in this test framework
            assert_equal(sendtx_rostrum["vin"][0]["tokenData"]["category"], token_id)
            assert_equal(sendtx_rostrum["vin"][0]["tokenData"]["amount"], "42")
            assert_equal(sendtx_rostrum["vin"][0]["tokenData"]["nft"], None)

            # node may have mangled the output index in 'createrawtransaction'; we need to look for it

            found_output = False
            for vout in sendtx_rostrum["vout"]:
                token = vout["tokenData"]
                if token is None:
                    continue

                assert_equal(token["category"], token_id)
                assert_equal(token["amount"], "42")
                assert_equal(token["nft"], None)
                found_output = True
            assert found_output, f"Did not find token output in tx ${sendtx_rostrum}"

        if on_nex():
            assert any(has_token_data(x, "input") for x in sendtx_rostrum["vin"])
            assert any(has_token_data(x, "output") for x in sendtx_rostrum["vout"])

            matched_input = False
            matched_output = False
            for txin in sendtx_rostrum["vin"]:
                if txin["group"] is not None:
                    assert_equal(token_id, txin["group"])
                    assert_equal(42, txin["groupQuantity"])
                    assert_equal(0, txin["groupAuthority"])
                    matched_input = True

            for i, output in enumerate(sendtx_node["vout"]):
                if "group" in output["scriptPubKey"]:
                    assert_equal(
                        output["scriptPubKey"]["group"],
                        sendtx_rostrum["vout"][i]["scriptPubKey"]["group"],
                    )
                    matched_output = True

            assert matched_input
            assert matched_output

    # pylint: disable=too-many-positional-arguments,too-many-statements,too-many-arguments
    async def test_verbose(
        self,
        n,
        cli,
        *,
        nonstandard_tx=None,
        p2sh_tx=None,
        p2st_tx=None,
        p2pkh_tx=None,
        unconfirmed_tx=None,
    ):
        """
        The spec is unclear. It states:

        "whatever the coin daemon returns when asked for a
         verbose form of the raw transaction"

        We've implemented this call in rostrum with
        "common denominators" between the bitcoind
        implementations and some extras.
        """

        # All confirmed transactions are confirmed in the tip
        block = n.getbestblockhash()
        tipheight = n.getblockcount()
        if on_bch():
            coinbase_tx = n.getblock(block, 1)["tx"][0]
        elif on_nex():
            coinbase_tx = n.getblock(block, 1)["txid"][0]
        else:
            raise NotImplementedError()

        # pylint: disable=too-many-branches
        async def check_tx(txid, is_confirmed=True, check_output_type=False):
            electrum = await cli.call(TX_GET, txid, True)
            if is_confirmed:
                bitcoind = n.getrawtransaction(txid, True, block)
            else:
                bitcoind = n.getrawtransaction(txid, True)

            if not is_confirmed:
                # Transaction is unconfirmed. We handle this slightly different
                # than bitcoind.
                assert_equal(None, electrum["blockhash"])
                assert_equal(None, electrum["confirmations"])
                assert_equal(None, electrum["time"])
                assert_equal(None, electrum["height"])
            else:
                assert_equal(n.getbestblockhash(), electrum["blockhash"])
                assert_equal(1, electrum["confirmations"])
                assert_equal(bitcoind["time"], electrum["time"])
                assert_equal(tipheight, electrum["height"])

            assert_equal(bitcoind["txid"], electrum["txid"])
            assert_equal(bitcoind["locktime"], electrum["locktime"])
            assert_equal(bitcoind["size"], electrum["size"])
            assert_equal(bitcoind["hex"], electrum["hex"])
            assert_equal(bitcoind["version"], electrum["version"])
            if "fee" in bitcoind:
                assert_equal(bitcoind["fee"], electrum["fee"])
            else:
                assert "fee" in electrum

            # inputs
            assert_equal(len(bitcoind["vin"]), len(bitcoind["vin"]))
            for i in range(len(bitcoind["vin"])):
                if "coinbase" in bitcoind["vin"][i]:
                    # bitcoind drops txid and other fields, butadds 'coinbase' for coinbase
                    # inputs
                    assert_equal(
                        bitcoind["vin"][i]["coinbase"], electrum["vin"][i]["coinbase"]
                    )
                    assert_equal(
                        bitcoind["vin"][i]["sequence"], electrum["vin"][i]["sequence"]
                    )
                    continue

                if on_nex():
                    assert_equal(
                        bitcoind["vin"][i]["outpoint"], electrum["vin"][i]["outpoint"]
                    )
                    assert "addresses" in electrum["vin"][i]
                    assert "group" in electrum["vin"][i]
                if on_bch():
                    assert_equal(bitcoind["vin"][i]["txid"], electrum["vin"][i]["txid"])
                    assert_equal(bitcoind["vin"][i]["vout"], electrum["vin"][i]["vout"])
                    assert "tokenData" in electrum["vin"][i]

                assert_equal(
                    bitcoind["vin"][i]["sequence"], electrum["vin"][i]["sequence"]
                )
                assert_equal(
                    bitcoind["vin"][i]["scriptSig"]["hex"],
                    electrum["vin"][i]["scriptSig"]["hex"],
                )

                # There is more than one way to represent script as assembly.
                # For instance '51' can be represented as '1' or 'OP_PUSHNUM_1'.
                # Just check for existance.
                assert "asm" in electrum["vin"][i]["scriptSig"]

            # outputs
            assert_equal(len(bitcoind["vout"]), len(bitcoind["vout"]))
            for i in range(len(bitcoind["vout"])):
                assert_equal(bitcoind["vout"][i]["n"], electrum["vout"][i]["n"])

                if on_nex():
                    assert_equal(
                        bitcoind["vout"][i]["type"], electrum["vout"][i]["type"]
                    )
                    if "scriptHash" in bitcoind["vout"][i]["scriptPubKey"]:
                        assert_equal(
                            bitcoind["vout"][i]["scriptPubKey"]["scriptHash"].lower(),
                            electrum["vout"][i]["scriptPubKey"]["scriptHash"].lower(),
                        )
                        assert_equal(
                            bitcoind["vout"][i]["scriptPubKey"]["argsHash"],
                            electrum["vout"][i]["scriptPubKey"]["argsHash"],
                        )
                    else:
                        assert_equal(
                            None, electrum["vout"][i]["scriptPubKey"]["scriptHash"]
                        )
                        assert_equal(
                            None, electrum["vout"][i]["scriptPubKey"]["argsHash"]
                        )

                assert_equal(
                    bitcoind["vout"][i]["value"] * COIN,
                    electrum["vout"][i]["value_coin"] * COIN,
                )

                assert_equal(
                    bitcoind["vout"][i]["value"] * COIN,
                    electrum["vout"][i]["value_satoshi"],
                )

                assert_equal(
                    bitcoind["vout"][i]["scriptPubKey"]["hex"],
                    electrum["vout"][i]["scriptPubKey"]["hex"],
                )
                assert "asm" in electrum["vout"][i]["scriptPubKey"]

                if on_nex():
                    if "addresses" in bitcoind["vout"][i]["scriptPubKey"]:
                        assert_equal(
                            bitcoind["vout"][i]["scriptPubKey"]["addresses"],
                            electrum["vout"][i]["scriptPubKey"]["addresses"],
                        )

                    assert "outpoint_hash" in electrum["vout"][i]

                if on_bch():
                    if "addresses" in bitcoind["vout"][i]["scriptPubKey"]:
                        for addr in bitcoind["vout"][i]["scriptPubKey"]["addresses"]:
                            assert (
                                addr in electrum["vout"][i]["scriptPubKey"]["addresses"]
                            ), f"address {addr} missing in response"
                    else:
                        assert_equal(
                            [], electrum["vout"][i]["scriptPubKey"]["addresses"]
                        )

                if check_output_type:
                    assert_equal(
                        bitcoind["vout"][i]["scriptPubKey"]["type"],
                        electrum["vout"][i]["scriptPubKey"]["type"],
                    )

        if nonstandard_tx:
            await check_tx(nonstandard_tx, check_output_type=False)
        if p2sh_tx:
            await check_tx(p2sh_tx)
        if p2st_tx:
            await check_tx(p2st_tx)
        if p2pkh_tx:
            await check_tx(p2pkh_tx)
        if coinbase_tx:
            await check_tx(coinbase_tx)
        if unconfirmed_tx:
            await check_tx(unconfirmed_tx, is_confirmed=False)

    async def test_by_txidem(self, cli, confirmed_tx):
        if not on_nex():
            return

        assert_equal(
            to_hex(confirmed_tx),
            await cli.call(TX_GET, confirmed_tx.get_rpc_hex_idem(), False),
        )

        # Transactions fetched by idem should have height
        # (issue #214)
        verbose = await cli.call(TX_GET, confirmed_tx.get_rpc_hex_idem(), True)
        assert verbose["height"] >= 100


if __name__ == "__main__":
    asyncio.run(ElectrumTransactionGet().main())
