kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

get tests working

Browse Source
This commit is contained in:
William Luke 2021-11-17 15:21:23 +03:00
parent 11e9881009
commit 3ba5ef9c8c
5 changed files with 89 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
apps/cic-eth/cic_eth/server/app.py
View File

@ -16,9 +16,9 @@ def create_app(test_config=None):
app.app.json_encoder = encoder.JSONEncoder
app.add_api('server.yaml', arguments={
'title': 'Grassroots Economics'}, pythonic_params=True)
app.run(port=5000)
return app.app
return app
if __name__ == '__main__':
create_app()
app = create_app()
app.run(port=5000)

4
apps/cic-eth/cic_eth/server/celery_helper.py
View File

@ -39,17 +39,17 @@ def call(method, *args):
callback_task='cic_eth.callbacks.redis.redis',
callback_queue=celery_queue,
)
log.debug(f"cic_eth.api.{method}({args})")
getattr(api, method)(*args)
ps.get_message()
try:
o = ps.get_message(timeout=config.get('REDIS_TIMEOUT'))
log.debug(f"cic_eth.api.{method}({args})\n {o}")
except TimeoutError as e:
sys.stderr.write(
f'cic_eth.api.{method}({args}) timed out:\n {e}')
sys.exit(1)
log.debug(f"cic_eth.api.{method}({args})\n {o}")
ps.unsubscribe()
try:
result = json.loads(o['data'])["result"]

19
apps/cic-eth/cic_eth/server/controllers/account_controller.py
View File

@ -1,4 +1,5 @@
import logging
from cic_eth.server.celery_helper import call
from cic_eth.server.models import TokenBalance, Transaction
@ -17,12 +18,12 @@ def account_balance(address, token_symbol, include_pending=True):
:param include_pending:
:type include_pending: bool
:rtype: TokenBalance
:rtype: [TokenBalance]
"""
data = call("balance", address,token_symbol, include_pending)
log.debug(data)
balance = call("balance", address, token_symbol, include_pending)
log.debug(balance)
#[{'address': '3ff776b6f888980def9d4220858803f9dc5e341e', 'converters': [], 'balance_network': 0}]
return list(map(lambda b: TokenBalance(**b), data))
return list(map(lambda b: TokenBalance(**b), balance))
def create_account_post(password="", register=True):
@ -52,11 +53,14 @@ def list_transactions(address, limit=10):
:param limit:
:type limit: int
:rtype: Token
:rtype: [Transaction]
"""
data = call('list', address, limit)
# data = task.get()
return list(map(lambda t: Transaction(**t), data))
log.debug(data)
transactions = list(map(lambda t: Transaction(**t), data))
log.debug(transactions)
return transactions
def transfer(from_address, to_address, value, token_symbol):
@ -102,6 +106,7 @@ def transfer_from(from_address, to_address, value, token_symbol, spender_address
:rtype: str, 0x-hex
"""
data = call("transfer_from", from_address, to_address, int(value * (10**6)), token_symbol, spender_address)
data = call("transfer_from", from_address, to_address, int(
value * (10**6)), token_symbol, spender_address)
return data

7
apps/cic-eth/cic_eth/server/openapi/server.yaml
View File

@ -59,7 +59,10 @@ paths:
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/TokenBalance"
x-openapi-router-controller: cic_eth.server.controllers.account_controller
/transactions:
description:
@ -89,7 +92,9 @@ paths:
content:
application/json:
schema:
$ref: "#/components/schemas/Token"
type: array
items:
$ref: "#/components/schemas/Transaction"
x-openapi-router-controller: cic_eth.server.controllers.account_controller
/create_account:
description:

80
apps/cic-eth/tests/server/test_account_controller.py
View File

@ -1,43 +1,55 @@
# coding: utf-8
from __future__ import absolute_import
import logging
import time
import hexathon
import pytest
from cic_eth.server.app import create_app
log = logging.getLogger(__name__)
@pytest.fixture
def client():
print("Here")
log.debug("Here")
app = create_app({'TESTING': True})
with app.test_client() as client:
with app.app.test_client() as client:
# with app.app_context():
# init_db()
yield client
def test_account(client):
query_string = [('password', ''),
('register', True)]
# Default Token
query_string = []
response = client.open(
'/create_account',
method='POST',
'/token',
method='GET',
query_string=query_string)
address_1 = response.get_json()
print(address_1)
default_token = response.get_json()
# Create Account 1
query_string = [('password', ''),
('register', True)]
response = client.open(
'/create_account',
method='POST',
query_string=query_string)
address_2 = response.get_json()
print(address_2)
time.sleep(10)
# Balance
address_1 = hexathon.valid(response.get_json())
# Create Account 2
query_string = [('password', ''),
('register', True)]
response = client.open(
'/create_account',
method='POST',
query_string=query_string)
address_2 = hexathon.valid(response.get_json())
time.sleep(30) # Required to allow balance to show
# Balance Account 1
query_string = [('address', address_1),
('token_symbol', 'COFE'),
('include_pending', True)]
@ -46,9 +58,17 @@ def test_account(client):
method='GET',
query_string=query_string)
balance = response.get_json()
print(balance)
# Transfer
assert (balance[0] == {
"address": default_token.get('address').lower(),
"balance_available": 30000000000,
"balance_incoming": 0,
"balance_network": 30000000000,
"balance_outgoing": 0,
"converters": []
})
# Transfer
query_string = [('from_address', address_1),
('to_address', address_2),
('value', 100),
@ -58,7 +78,35 @@ def test_account(client):
method='POST',
query_string=query_string)
transfer = response.get_json()
print(transfer)
# Balance Account 1
query_string = [('address', address_1),
('token_symbol', 'COFE'),
('include_pending', True)]
response = client.open(
'/balance',
method='GET',
query_string=query_string)
balance_after_transfer = response.get_json()
assert (balance_after_transfer[0] == {
"address": default_token.get('address').lower(),
"balance_available": 29900000000,
"balance_incoming": 0,
"balance_network": 30000000000,
"balance_outgoing": 100000000,
"converters": []
})
# Transactions Account 1
query_string = [('address', address_1),
('limit', 10)]
response = client.open(
'/transactions',
method='GET',
query_string=query_string)
transactions = response.get_json()
log.debug(transactions)
exit(1)
# def test_list_transactions(client):