LabChain/tests/test_rest_api.py

from datetime import datetime


from src.protocol import Protocol
from src.chainbuilder import ChainBuilder
from src.rpc_server import rpc_server
from _thread import start_new_thread
from tests.utils import *
from src.rpc_server import shutdown_server
from tests.test_verifications import block_test

from Crypto.PublicKey import RSA

import requests

sess = requests.Session()
host = "http://localhost:{}/"
port = 2345

NUMBER_FIELDS_IN_BLOCK = 9

def start_server(chain):
    """builds a chainbuilder from a Blockchain, appends and starts a rpc server"""
    proto = Protocol([], GENESIS_BLOCK, 1337)
    chainbuilder = ChainBuilder(proto)

    chainbuilder.primary_block_chain = chain

    return start_new_thread(rpc_server, (port, chainbuilder, None))

def rpc_test(count):
    """starts the rpc server, runs a test and stopps the server"""
    def decorator(fn):
        @block_test()
        def wrapper(gen_chain):
            chain = gen_chain
            key = Key.generate_private_key()
            for i in range(count):
                reward_trans = Transaction([], [TransactionTarget(key, chain.compute_blockreward_next_block())], datetime.now())
                chain = extend_blockchain(chain, [reward_trans])
            start_server(chain)
            fn(chain)
            res = sess.get(host.format(port) + 'shutdown')
            assert res.text == 'Server shutting down...'
        return wrapper
    return decorator


def get_path(path):
    """endpoint link"""
    res = sess.get(host.format(port) + path)
    assert (res.status_code == 200), "Could not reach endpoint " + path
    return res

def get_explorer(path):
    """adds path to explorer"""
    return get_path("explorer/"+path)

def get_statistics(path):
    """adds statistics to path"""
    return get_explorer("statistics/"+path)

@rpc_test(10)
def test_explorer_availability(chain):
    """explorer check for addresses, transcactions, blocks, lasttransactions, lastblocks, blockat"""
    get_explorer('addresses')
    get_explorer('transactions')
    get_explorer('blocks')

    get_explorer('lasttransactions/10')
    get_explorer('lastblocks/10')
    get_explorer('blockat/10')

@rpc_test(10)
def test_statistics_availability(chain):
    """statistics check for hashrate, tps, totalblocks, difficulty, blocktime"""
    get_statistics('hashrate')
    get_statistics('tps')
    get_statistics('totalblocks')
    get_statistics('difficulty')
    get_statistics('blocktime')
    

  

def check_address_data(address):
    """check for sender and receiver"""
    assert 'received' in address, "response object does not contain a received"
    assert 'sent' in address, "response object does not contain a sent"



@rpc_test(1)
def test_address_data(chain):
    """gets the first address and checks the closer information"""
    hash = get_explorer("addresses").json()[0]
    res = get_explorer("sortedtransactions/" + hash)

    res_json = res.json()

    check_address_data(res_json)

    
def checkblock_data(block):
    """blockcheck for nonce, id, difficulty, height, prev_block_hash, transactions, time, merkle_root_hash, hash"""
    assert len(block) == NUMBER_FIELDS_IN_BLOCK, "Wrong number of Fields in response object"
    assert 'nonce' in block, "response object does not contain a nonce"
    assert 'id' in block, "response object does not contain a id"
    assert 'difficulty' in block, "response object does not contain a difficulty"
    assert 'height' in block, "response object does not contain a height"
    assert 'prev_block_hash' in block, "response object does not contain a prev_block_hash"
    assert 'transactions' in block, "response object does not contain a transactions"
    assert 'time' in block, "response object does not contain a time"
    assert 'merkle_root_hash' in block, "response object does not contain a merkle_root_hash"
    assert 'hash' in block, "response object does not contain a hash"


@rpc_test(1)
def test_lastblocks_block_data(chain):
    """checks the blockdata"""
    res = get_explorer('lastblocks/10')

    res_json = res.json()
    assert len(res_json) == 2, "Incorrect count of Blocks"

    block = res_json[0]
    checkblock_data(block)



@rpc_test(1)
def test_transactions_data(chain):
    """check in transaction for hash, block_id, targets, timestamp, number_confirmations, block_hash, signatures, inputs"""
    res = sess.get(host.format(port) + 'explorer/transactions')
    assert (res.status_code == 200)

    res_json = res.json()
    assert len(res_json) == 1, "Incorrect count of transactions"

    transaction = res_json[0]
    assert len(transaction) == 8, "Wrong number of Fields in resonse object"

    assert 'hash' in transaction, "response object does not contain a hash"
    assert 'block_id' in transaction, "response object does not contain a block_id"
    assert 'targets' in transaction, "response object does not contain targets"
    assert 'timestamp' in transaction, "response object does not contain a timestamp"
    assert 'number_confirmations' in transaction, "response object does not contain a number_confirmations"
    assert 'block_hash' in transaction, "response object does not contain a block_hash"
    assert 'signatures' in transaction, "response object does not contain signatures"
    assert 'inputs' in transaction, "response object does not contain inputs"

    assert len(transaction["inputs"]) == len(transaction["signatures"]), "Difference in amount of signatures and inputs"



def check_data_count(subpath):
    """checks the correctness of the count from the subpath"""
    @rpc_test(10)
    def inner(chain):
        path = "explorer/"+subpath
        url = host.format(port) + path + "/" +str(5)
        res = sess.get(url)
        assert (res.status_code == 200), url

        res_json = res.json()
        assert len(res_json) == 5, "Incorrect count of " + subpath

        url = host.format(port) + path + "/" + str(20)
        res = sess.get(url)
        assert (res.status_code == 200), url

        res_json = res.json()
        if(subpath == "lastblocks"):
            assert len(res_json) == 11, "Incorrect count of " + subpath
        if(subpath == "lasttransactions"):
            assert len(res_json) == 10, "Incorrect count of " + subpath
    inner()

def test_blocks_count():
    check_data_count("lastblocks")

def test_transactions_count():
    check_data_count("lasttransactions")