#!/usr/bin/env python3
# Copyright (c) 2019 The Bitcoin Unlimited developers
"""
Tests to check if basic electrum server integration works
"""

import asyncio
from decimal import Decimal
from test_framework.util import wait_for, assert_equal, assert_raises_async
from test_framework.electrumutil import (
    get_txid_from_idem,
    ElectrumTestFramework,
)
from test_framework.electrumconnection import ElectrumConnection
from test_framework.connectrum.exc import ElectrumErrorResponse
from test_framework.constants import COIN
from test_framework.environment import on_nex, on_bch


async def address_to_scripthash(cli, address):
    return await cli.call("blockchain.address.get_scripthash", address)


class ElectrumBlockchainAddress(ElectrumTestFramework):
    """
    Basic blockchain.address.* testing, mostly to check that the function
    handle an address correctly. The blockchain.scripthash.* equivalents are
    more thoroughly tested.
    """

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

        n.generate(200)

        await self.test_get_frist_use(n)
        cli = ElectrumConnection()
        await cli.connect()
        try:
            await self.test_invalid_args(cli)
            await self.test_get_balance(n, cli)
            await self.test_get_history(n, cli)
            await self.test_list_unspent(n, cli)
        finally:
            cli.disconnect()

    async def test_invalid_args(self, cli):
        hash_param_methods = (
            "blockchain.address.get_balance",
            "blockchain.address.get_history",
            "blockchain.address.listunspent",
        )

        for method in hash_param_methods:
            await assert_raises_async(
                ElectrumErrorResponse, cli.call, method, "invalidaddress"
            )

    async def test_get_balance(self, n, cli):
        addr = n.getnewaddress()
        balance = 11.42
        n.sendtoaddress(addr, balance)

        async def check_address(address, unconfirmed=0, confirmed=0):
            res = await cli.call("blockchain.address.get_balance", address)

            return (
                res["unconfirmed"] == unconfirmed * COIN
                and res["confirmed"] == confirmed * COIN
            )

        async def check_unconfirmed():
            return await check_address(addr, unconfirmed=balance)

        async def check_confirmed():
            return await check_address(addr, confirmed=balance)

        await wait_for(10, check_unconfirmed)
        n.generate(1)
        await wait_for(10, check_confirmed)

    async def sendtoaddr(self, n, cli, addr):
        utxo = n.listunspent().pop()
        if on_bch():
            inputs = [{"txid": utxo["txid"], "vout": utxo["vout"]}]
        elif on_nex():
            inputs = [{"outpoint": utxo["outpoint"], "amount": utxo["amount"]}]
        else:
            raise NotImplementedError()
        fee = Decimal(300) / COIN
        outputs = {
            addr: utxo["amount"] - fee,
        }
        tx = n.createrawtransaction(inputs, outputs)
        signed = n.signrawtransaction(tx)
        txid = await cli.call("blockchain.transaction.broadcast", signed["hex"])
        return txid

    async def test_get_frist_use(self, n):
        cli = ElectrumConnection()
        await cli.connect()

        # New address that has never received coins. Should return an error.
        addr = n.getnewaddress()
        scripthash = await address_to_scripthash(cli, addr)
        await assert_raises_async(
            ElectrumErrorResponse, cli.call, "blockchain.address.get_first_use", addr
        )
        await assert_raises_async(
            ElectrumErrorResponse,
            cli.call,
            "blockchain.scripthash.get_first_use",
            scripthash,
        )

        # Send coin to the new address
        txid = await get_txid_from_idem(n, await self.sendtoaddr(n, cli, addr))

        # Wait for electrum server to see the utxo.
        async def wait_for_utxo():
            utxo = await cli.call("blockchain.address.listunspent", addr)
            if len(utxo) == 1:
                return utxo
            return None

        await wait_for(10, wait_for_utxo)

        # Observe that get_first_use returns the tx when it's in the mempool
        res = await cli.call("blockchain.address.get_first_use", addr)
        res2 = await cli.call(
            "blockchain.scripthash.get_first_use",
            await address_to_scripthash(cli, addr),
        )
        assert_equal(res, res2)
        assert_equal(
            "0000000000000000000000000000000000000000000000000000000000000000",
            res["block_hash"],
        )
        assert_equal(0, res["height"])
        assert_equal(txid, res["tx_hash"])

        # Confirm tx, observe that block height and gets set.
        n.generate(1)
        await self.sync_height(cli)
        res = await cli.call("blockchain.address.get_first_use", addr)
        res2 = await cli.call(
            "blockchain.scripthash.get_first_use",
            await address_to_scripthash(cli, addr),
        )
        assert_equal(res, res2)
        assert_equal(n.getbestblockhash(), res["block_hash"])
        assert_equal(n.getblockcount(), res["height"])
        assert_equal(txid, res["tx_hash"])

        # Send another tx, observe that the first one is till returned.
        await self.sendtoaddr(n, cli, addr)
        res = await cli.call("blockchain.address.get_first_use", addr)
        assert_equal(txid, res["tx_hash"])

        # Also when the second tx is confirmed, the first is returned.
        n.generate(1)
        await self.sync_height(cli)
        res = await cli.call("blockchain.address.get_first_use", addr)
        assert_equal(txid, res["tx_hash"])
        cli.disconnect()

    async def test_get_first_use_edition_2027_pre(self, n):
        cli = ElectrumConnection()
        await cli.connect(edition="2027-pre")

        # New address that has never received coins. Should return null with edition 2027-pre.
        addr = n.getnewaddress()
        res = await cli.call("blockchain.address.get_first_use", addr)
        assert_equal(None, res)

        scripthash = await address_to_scripthash(cli, addr)
        res = await cli.call("blockchain.scripthash.get_first_use", scripthash)
        assert_equal(None, res)

        # Send coin to the new address
        txid = await get_txid_from_idem(n, await self.sendtoaddr(n, cli, addr))

        # Wait for transaction to propagate to mempool
        await self.wait_for_mempool_tx(cli, txid, n)

        # With edition 2027-pre, get_first_use should still work when address has history
        res = await cli.call("blockchain.address.get_first_use", addr)
        assert_equal(txid, res["tx_hash"])

        cli.disconnect()

    async def test_list_unspent(self, n, cli):
        addr = n.getnewaddress()
        utxo = await cli.call("blockchain.address.listunspent", addr)
        assert_equal(0, len(utxo))

        txidem = n.sendtoaddress(addr, 21)
        txid = await get_txid_from_idem(n, txidem)

        async def fetch_utxo():
            utxo = await cli.call("blockchain.address.listunspent", addr)
            if len(utxo) > 0:
                return utxo
            return None

        utxo = await wait_for(10, fetch_utxo)
        assert_equal(1, len(utxo))

        assert_equal(0, utxo[0]["height"])
        assert_equal(txid, utxo[0]["tx_hash"])
        assert_equal(21 * COIN, utxo[0]["value"])
        assert utxo[0]["tx_pos"] in [0, 1]

        n.generate(1)

        async def wait_for_confheight():
            utxo = await cli.call("blockchain.address.listunspent", addr)
            return len(utxo) == 1 and utxo[0]["height"] == n.getblockcount()

        await wait_for(10, wait_for_confheight)

    async def test_get_history(self, n, cli):
        addr = n.getnewaddress()
        txidem = n.sendtoaddress(addr, 11)
        txid = await get_txid_from_idem(n, txidem)

        async def fetch_history():
            h = await cli.call("blockchain.address.get_history", addr)
            if len(h) > 0:
                return h
            return None

        history = await wait_for(10, fetch_history)
        assert_equal(1, len(history))

        h = await cli.call(
            "blockchain.address.get_history", addr, None
        )  # same as above, passing None
        assert_equal(1, len(h))

        unconfirmed_heights = [0, -1]
        assert history[0]["height"] in unconfirmed_heights
        assert_equal(txid, history[0]["tx_hash"])

        n.generate(1)

        async def wait_for_confheight():
            h = await cli.call("blockchain.address.get_history", addr)
            return len(h) == 1 and h[0]["height"] == n.getblockcount()

        await wait_for(10, wait_for_confheight)


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