add swagger server

This commit is contained in:
William Luke 2021-11-12 11:14:40 +03:00
parent c48177c78c
commit 59c8895db5
21 changed files with 1335 additions and 2 deletions

2
.envrc
View File

@ -6,3 +6,5 @@ export TOKEN_SYMBOL=COFE
export TOKEN_DECIMALS=6
export FAUCET_AMOUNT=30000000000
export DATABASE_DEBUG=1
export DOCKER_REGISTRY="registry.gitlab.com/grassrootseconomics"
export DEV_DEBUG=3

View File

View File

@ -0,0 +1,21 @@
#!/usr/bin/env python3
from cic_eth.server import encoder
import cic_eth.cli
import connexion
from cic_eth.graphql.config import config
celery_app = cic_eth.cli.CeleryApp.from_config(config)
celery_app.set_default()
def main():
app = connexion.App(__name__, specification_dir='./swagger/')
app.app.json_encoder = encoder.JSONEncoder
app.add_api('swagger.yaml', arguments={
'title': 'Grassroots Economics'}, pythonic_params=True)
app.run(port=5000)
if __name__ == '__main__':
main()

View File

@ -0,0 +1,185 @@
import json
import logging
import sys
import uuid
import connexion
import redis
import six
from cic_eth.api.api_task import Api
from cic_eth.graphql.config import config
from cic_eth.server import util
from cic_eth.server.models.model0x_address import Model0xAddress # noqa: E501
from cic_eth.server.models.token import Token # noqa: E501
from cic_eth.server.models.token_balance import TokenBalance # noqa: E501
log = logging.getLogger(__name__)
chain_spec = config.get('CHAIN_SPEC')
celery_queue = config.get('CELERY_QUEUE')
api = Api(
chain_spec,
queue=celery_queue,
)
redis_host = config.get('REDIS_HOST')
redis_port = config.get('REDIS_PORT')
redis_db = config.get('REDIS_DB')
logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger(__name__)
def call(method, *args):
redis_channel = str(uuid.uuid4())
r = redis.Redis(redis_host, redis_port, redis_db)
ps = r.pubsub()
ps.subscribe(redis_channel)
log.debug(f"message 1: {ps.get_message()}") # Subscription Object
print(f"Channel init {redis_channel}")
print(args)
api = Api(
chain_spec,
queue=celery_queue,
callback_param='{}:{}:{}:{}'.format(
redis_host, redis_port, redis_db, redis_channel),
callback_task='cic_eth.callbacks.redis.redis',
callback_queue=celery_queue,
)
getattr(api, method)(*args)
log.debug(f"message 2: {ps.get_message()}") # returns None !?
try:
o = ps.get_message(timeout=config.get('REDIS_TIMEOUT'))
log.debug(f"message 3: {o}")
except TimeoutError as e:
sys.stderr.write(
'got no new address from cic-eth before timeout: {}\n'.format(e))
sys.exit(1)
ps.unsubscribe()
print(o)
m = json.loads(o['data'])
return m["result"]
def account_balance(address, token_symbol, include_pending=True): # noqa: E501
"""account_balance
Retrieve Address Balance # noqa: E501
:param address:
:type address: dict | bytes
:param token_symbol:
:type token_symbol: str
:param include_pending:
:type include_pending: bool
:rtype: TokenBalance
"""
task = api.balance(address=address, token_symbol=token_symbol,
include_pending=include_pending)
data = task.get() # api call('balance', address, token_symbol, include_pending)
log.debug(data)
#[{'address': '3ff776b6f888980def9d4220858803f9dc5e341e', 'converters': [], 'balance_network': 0}]
return data
def create_account_post(password, register): # noqa: E501
"""create_account_post
Creates a new blockchain address # noqa: E501
:param password:
:type password: str
:param register:
:type register: bool
:rtype: Model0xAddress
"""
data = call("create_account", password, register)
return data
def list_transactions(address, limit=10): # noqa: E501
"""list_transactions
Retrieve Address Balance # noqa: E501
:param address:
:type address: dict | bytes
:param limit:
:type limit: int
:rtype: Token
"""
api = Api(
chain_spec,
queue=celery_queue,
)
task = api.list(address=address, limit=limit)
data = task.get()
return data
def transfer(from_address, to_address, value, token_symbol): # noqa: E501
"""transfer
Performs a transfer of ERC20 tokens from one address to another. # noqa: E501
:param from_address:
:type from_address: dict | bytes
:param to_address:
:type to_address: dict | bytes
:param value:
:type value: int
:param token_symbol:
:type token_symbol: str
:rtype: Token
"""
api = Api(
chain_spec,
queue=celery_queue,
)
t = api.transfer(from_address, to_address,
int(value * (10**6)), token_symbol)
log.debug(f"t {t}")
log.debug(f"transfer {t.get_leaf()}")
log.debug(f"transfer {t.successful()}")
return t.get()
def transfer_from(from_address, to_address, value, token_symbol, spender_address): # noqa: E501
"""transfer_from
Performs a transfer of ERC20 tokens by one address on behalf of another address to a third party. # noqa: E501
:param from_address: Ethereum address of sender
:type from_address: dict | bytes
:param to_address: Ethereum address of recipient
:type to_address: dict | bytes
:param value: Estimated return from conversion
:type value: int
:param token_symbol: ERC20 token symbol of token to send
:type token_symbol: str
:param spender_address: Ethereum address of recipient
:type spender_address: dict | bytes
:rtype: Token
"""
api = Api(
chain_spec,
queue=celery_queue,
)
t = api.transfer_from(from_address, to_address,
int(value * (10**6)), token_symbol, spender_address)
log.debug(f"t {t}")
log.debug(f"transfer {t.get_leaf()}")
log.debug(f"transfer {t.successful()}")
return t.get()

View File

@ -0,0 +1,6 @@
from typing import List
"""
controller generated to handled auth operation described at:
https://connexion.readthedocs.io/en/latest/security.html
"""

View File

@ -0,0 +1,37 @@
import logging
import connexion
import six
from cic_eth.api.api_task import Api
from cic_eth.graphql.config import config
from cic_eth.server import util
from cic_eth.server.models import Token
from cic_eth.server.models.token import Token # noqa: E501
log = logging.getLogger(__name__)
chain_spec = config.get('CHAIN_SPEC')
celery_queue = config.get('CELERY_QUEUE')
api = Api(
chain_spec,
queue=celery_queue,
)
def get_default_token(): # noqa: E501
"""get_default_token
Retrieve transactions # noqa: E501
:rtype: Token
"""
task = api.default_token()
data = task.get() # api call('balance', address, token_symbol, include_pending)
task = api.default_token()
data = task.get()
print(data)
log.debug(data)
return Token(address=data['address'], symbol=data['symbol'], decimals=data['decimals'], name=data['name'])

View File

@ -0,0 +1,20 @@
from connexion.apps.flask_app import FlaskJSONEncoder
import six
from cic_eth.server.models.base_model_ import Model
class JSONEncoder(FlaskJSONEncoder):
include_nulls = False
def default(self, o):
if isinstance(o, Model):
dikt = {}
for attr, _ in six.iteritems(o.swagger_types):
value = getattr(o, attr)
if value is None and not self.include_nulls:
continue
attr = o.attribute_map[attr]
dikt[attr] = value
return dikt
return FlaskJSONEncoder.default(self, o)

View File

@ -0,0 +1,8 @@
# coding: utf-8
# flake8: noqa
from __future__ import absolute_import
# import models into model package
from cic_eth.server.models.model0x_address import Model0xAddress
from cic_eth.server.models.token import Token
from cic_eth.server.models.token_balance import TokenBalance

View File

@ -0,0 +1,69 @@
import pprint
import six
import typing
from cic_eth.server import util
T = typing.TypeVar('T')
class Model(object):
# swaggerTypes: The key is attribute name and the
# value is attribute type.
swagger_types = {}
# attributeMap: The key is attribute name and the
# value is json key in definition.
attribute_map = {}
@classmethod
def from_dict(cls: typing.Type[T], dikt) -> T:
"""Returns the dict as a model"""
return util.deserialize_model(dikt, cls)
def to_dict(self):
"""Returns the model properties as a dict
:rtype: dict
"""
result = {}
for attr, _ in six.iteritems(self.swagger_types):
value = getattr(self, attr)
if isinstance(value, list):
result[attr] = list(map(
lambda x: x.to_dict() if hasattr(x, "to_dict") else x,
value
))
elif hasattr(value, "to_dict"):
result[attr] = value.to_dict()
elif isinstance(value, dict):
result[attr] = dict(map(
lambda item: (item[0], item[1].to_dict())
if hasattr(item[1], "to_dict") else item,
value.items()
))
else:
result[attr] = value
return result
def to_str(self):
"""Returns the string representation of the model
:rtype: str
"""
return pprint.pformat(self.to_dict())
def __repr__(self):
"""For `print` and `pprint`"""
return self.to_str()
def __eq__(self, other):
"""Returns true if both objects are equal"""
return self.__dict__ == other.__dict__
def __ne__(self, other):
"""Returns true if both objects are not equal"""
return not self == other

View File

@ -0,0 +1,37 @@
# coding: utf-8
from __future__ import absolute_import
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from cic_eth.server.models.base_model_ import Model
from cic_eth.server import util
class Model0xAddress(Model):
"""NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self): # noqa: E501
"""Model0xAddress - a model defined in Swagger
"""
self.swagger_types = {
}
self.attribute_map = {
}
@classmethod
def from_dict(cls, dikt) -> 'Model0xAddress':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The 0xAddress of this Model0xAddress. # noqa: E501
:rtype: Model0xAddress
"""
return util.deserialize_model(dikt, cls)

View File

@ -0,0 +1,149 @@
# coding: utf-8
from __future__ import absolute_import
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from cic_eth.server.models.base_model_ import Model
from cic_eth.server import util
class Token(Model):
"""NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self, symbol: str = None, address: str = None, name: str = None, decimals: str = None): # noqa: E501
"""Token - a model defined in Swagger
:param symbol: The symbol of this Token. # noqa: E501
:type symbol: str
:param address: The address of this Token. # noqa: E501
:type address: str
:param name: The name of this Token. # noqa: E501
:type name: str
:param decimals: The decimals of this Token. # noqa: E501
:type decimals: str
"""
self.swagger_types = {
'symbol': str,
'address': str,
'name': str,
'decimals': str
}
self.attribute_map = {
'symbol': 'symbol',
'address': 'address',
'name': 'name',
'decimals': 'decimals'
}
self._symbol = symbol
self._address = address
self._name = name
self._decimals = decimals
@classmethod
def from_dict(cls, dikt) -> 'Token':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The Token of this Token. # noqa: E501
:rtype: Token
"""
return util.deserialize_model(dikt, cls)
@property
def symbol(self) -> str:
"""Gets the symbol of this Token.
Token Symbol # noqa: E501
:return: The symbol of this Token.
:rtype: str
"""
return self._symbol
@symbol.setter
def symbol(self, symbol: str):
"""Sets the symbol of this Token.
Token Symbol # noqa: E501
:param symbol: The symbol of this Token.
:type symbol: str
"""
self._symbol = symbol
@property
def address(self) -> str:
"""Gets the address of this Token.
Token Address # noqa: E501
:return: The address of this Token.
:rtype: str
"""
return self._address
@address.setter
def address(self, address: str):
"""Sets the address of this Token.
Token Address # noqa: E501
:param address: The address of this Token.
:type address: str
"""
self._address = address
@property
def name(self) -> str:
"""Gets the name of this Token.
Token Name # noqa: E501
:return: The name of this Token.
:rtype: str
"""
return self._name
@name.setter
def name(self, name: str):
"""Sets the name of this Token.
Token Name # noqa: E501
:param name: The name of this Token.
:type name: str
"""
self._name = name
@property
def decimals(self) -> str:
"""Gets the decimals of this Token.
Decimals # noqa: E501
:return: The decimals of this Token.
:rtype: str
"""
return self._decimals
@decimals.setter
def decimals(self, decimals: str):
"""Sets the decimals of this Token.
Decimals # noqa: E501
:param decimals: The decimals of this Token.
:type decimals: str
"""
self._decimals = decimals

View File

@ -0,0 +1,193 @@
# coding: utf-8
from __future__ import absolute_import
from datetime import date, datetime # noqa: F401
from typing import List, Dict # noqa: F401
from cic_eth.server.models.base_model_ import Model
from cic_eth.server import util
class TokenBalance(Model):
"""NOTE: This class is auto generated by the swagger code generator program.
Do not edit the class manually.
"""
def __init__(self, address: str = None, converters: List[str] = None, balance_network: int = None, balance_incoming: int = None, balance_outgoing: int = None, balance_available: int = None): # noqa: E501
"""TokenBalance - a model defined in Swagger
:param address: The address of this TokenBalance. # noqa: E501
:type address: str
:param converters: The converters of this TokenBalance. # noqa: E501
:type converters: List[str]
:param balance_network: The balance_network of this TokenBalance. # noqa: E501
:type balance_network: int
:param balance_incoming: The balance_incoming of this TokenBalance. # noqa: E501
:type balance_incoming: int
:param balance_outgoing: The balance_outgoing of this TokenBalance. # noqa: E501
:type balance_outgoing: int
:param balance_available: The balance_available of this TokenBalance. # noqa: E501
:type balance_available: int
"""
self.swagger_types = {
'address': str,
'converters': List[str],
'balance_network': int,
'balance_incoming': int,
'balance_outgoing': int,
'balance_available': int
}
self.attribute_map = {
'address': 'address',
'converters': 'converters',
'balance_network': 'balance_network',
'balance_incoming': 'balance_incoming',
'balance_outgoing': 'balance_outgoing',
'balance_available': 'balance_available'
}
self._address = address
self._converters = converters
self._balance_network = balance_network
self._balance_incoming = balance_incoming
self._balance_outgoing = balance_outgoing
self._balance_available = balance_available
@classmethod
def from_dict(cls, dikt) -> 'TokenBalance':
"""Returns the dict as a model
:param dikt: A dict.
:type: dict
:return: The TokenBalance of this TokenBalance. # noqa: E501
:rtype: TokenBalance
"""
return util.deserialize_model(dikt, cls)
@property
def address(self) -> str:
"""Gets the address of this TokenBalance.
:return: The address of this TokenBalance.
:rtype: str
"""
return self._address
@address.setter
def address(self, address: str):
"""Sets the address of this TokenBalance.
:param address: The address of this TokenBalance.
:type address: str
"""
self._address = address
@property
def converters(self) -> List[str]:
"""Gets the converters of this TokenBalance.
:return: The converters of this TokenBalance.
:rtype: List[str]
"""
return self._converters
@converters.setter
def converters(self, converters: List[str]):
"""Sets the converters of this TokenBalance.
:param converters: The converters of this TokenBalance.
:type converters: List[str]
"""
self._converters = converters
@property
def balance_network(self) -> int:
"""Gets the balance_network of this TokenBalance.
:return: The balance_network of this TokenBalance.
:rtype: int
"""
return self._balance_network
@balance_network.setter
def balance_network(self, balance_network: int):
"""Sets the balance_network of this TokenBalance.
:param balance_network: The balance_network of this TokenBalance.
:type balance_network: int
"""
self._balance_network = balance_network
@property
def balance_incoming(self) -> int:
"""Gets the balance_incoming of this TokenBalance.
:return: The balance_incoming of this TokenBalance.
:rtype: int
"""
return self._balance_incoming
@balance_incoming.setter
def balance_incoming(self, balance_incoming: int):
"""Sets the balance_incoming of this TokenBalance.
:param balance_incoming: The balance_incoming of this TokenBalance.
:type balance_incoming: int
"""
self._balance_incoming = balance_incoming
@property
def balance_outgoing(self) -> int:
"""Gets the balance_outgoing of this TokenBalance.
:return: The balance_outgoing of this TokenBalance.
:rtype: int
"""
return self._balance_outgoing
@balance_outgoing.setter
def balance_outgoing(self, balance_outgoing: int):
"""Sets the balance_outgoing of this TokenBalance.
:param balance_outgoing: The balance_outgoing of this TokenBalance.
:type balance_outgoing: int
"""
self._balance_outgoing = balance_outgoing
@property
def balance_available(self) -> int:
"""Gets the balance_available of this TokenBalance.
:return: The balance_available of this TokenBalance.
:rtype: int
"""
return self._balance_available
@balance_available.setter
def balance_available(self, balance_available: int):
"""Sets the balance_available of this TokenBalance.
:param balance_available: The balance_available of this TokenBalance.
:type balance_available: int
"""
self._balance_available = balance_available

View File

@ -0,0 +1,289 @@
openapi: 3.0.3
info:
title: Grassroots Economics
description: CIC ETH API
termsOfService: bzz://grassrootseconomics.eth/terms
contact:
name: Grassroots Economics
url: https://www.grassrootseconomics.org
email: will@grassecon.org
license:
name: GPLv3
version: 0.1.0
servers:
- url: /
paths:
/token:
description: Retrieves the default fallback token of the custodial network.
get:
tags:
- Token
description: Retrieve transactions
operationId: get_default_token
responses:
"200":
description: OK.
content:
application/json:
schema:
$ref: "#/components/schemas/Token"
x-openapi-router-controller: cic_eth.server.controllers.token_controller
/balance:
description: Retrieves the current token balance of the given address
get:
tags:
- Account
description: Retrieve Address Balance
operationId: account_balance
parameters:
- name: address
in: query
required: true
style: form
explode: true
schema:
$ref: "#/components/schemas/0xAddress"
- name: token_symbol
in: query
required: true
style: form
explode: true
schema:
type: string
- name: include_pending
in: query
required: false
style: form
explode: true
schema:
type: boolean
default: true
responses:
"200":
description: OK.
content:
application/json:
schema:
$ref: "#/components/schemas/TokenBalance"
x-openapi-router-controller: cic_eth.server.controllers.account_controller
/transactions:
description:
Retrieve an aggregate list of latest transactions of internal and
(optionally) external origin in reverse chronological order.
get:
tags:
- Account
description: Retrieve Address Balance
operationId: list_transactions
parameters:
- name: address
in: query
required: true
style: form
explode: true
schema:
$ref: "#/components/schemas/0xAddress"
- name: limit
in: query
required: false
style: form
explode: true
schema:
type: integer
default: 10
responses:
"200":
description: OK.
content:
application/json:
schema:
$ref: "#/components/schemas/Token"
x-openapi-router-controller: cic_eth.server.controllers.account_controller
/create_account:
description:
Creates a new blockchain address encrypted with the given password
and returns the new address
post:
tags:
- Account
description: Creates a new blockchain address
operationId: create_account_post
parameters:
- name: password
in: query
required: true
style: form
explode: true
allowReserved: true
schema:
type: string
- name: register
in: query
required: true
style: form
explode: true
allowReserved: true
schema:
type: boolean
default: true
responses:
"200":
description: OK.
content:
application/json:
schema:
type: string
example: "e88ba386f0efc7117e8d0d17a750fce492516ecb"
x-openapi-router-controller: cic_eth.server.controllers.account_controller
/transfer:
description: Performs a transfer of ERC20 tokens from one address to another.
post:
tags:
- Account
description: Performs a transfer of ERC20 tokens from one address to another.
operationId: transfer
parameters:
- name: from_address
in: query
required: true
style: form
explode: true
schema:
$ref: "#/components/schemas/0xAddress"
- name: to_address
in: query
required: true
style: form
explode: true
schema:
$ref: "#/components/schemas/0xAddress"
- name: value
in: query
required: true
style: form
explode: true
schema:
type: integer
- name: token_symbol
in: query
required: true
style: form
explode: true
schema:
type: string
responses:
"200":
description: OK.
content:
application/json:
schema:
$ref: "#/components/schemas/Token"
x-openapi-router-controller: cic_eth.server.controllers.account_controller
/transfer_from:
post:
tags:
- Account
description:
Performs a transfer of ERC20 tokens by one address on behalf of
another address to a third party.
operationId: transfer_from
parameters:
- name: from_address
in: query
description: Ethereum address of sender
required: true
style: form
explode: true
schema:
$ref: "#/components/schemas/0xAddress"
- name: to_address
in: query
description: Ethereum address of recipient
required: true
style: form
explode: true
schema:
$ref: "#/components/schemas/0xAddress"
- name: value
in: query
description: Estimated return from conversion
required: true
style: form
explode: true
schema:
type: integer
- name: token_symbol
in: query
description: ERC20 token symbol of token to send
required: true
style: form
explode: true
schema:
type: string
- name: spender_address
in: query
description: Ethereum address of recipient
required: true
style: form
explode: true
schema:
$ref: "#/components/schemas/0xAddress"
responses:
"200":
description: OK.
content:
application/json:
schema:
$ref: "#/components/schemas/Token"
x-openapi-router-controller: cic_eth.server.controllers.account_controller
components:
schemas:
"0xAddress":
type: string
example: 0x-hex
Token:
type: object
properties:
symbol:
type: string
description: Token Symbol
address:
type: string
description: Token Address
name:
type: string
description: Token Name
decimals:
type: integer
description: Decimals
example:
symbol: 'GTF'
address: '3FF776B6f888980DEf9d4220858803f9dC5e341e'
decimals: 6
name: "Gift Token"
TokenBalance:
type: object
properties:
address:
type: string
converters:
type: array
items:
type: string
balance_network:
type: integer
balance_incoming:
type: integer
balance_outgoing:
type: integer
balance_available:
type: integer
example:
balance_network: 0
address: address
balance_incoming: 6
balance_available: 5
converters:
- converters
- converters
balance_outgoing: 1

View File

@ -0,0 +1,16 @@
import logging
import connexion
from flask_testing import TestCase
from cic_eth.server.encoder import JSONEncoder
class BaseTestCase(TestCase):
def create_app(self):
logging.getLogger('connexion.operation').setLevel('ERROR')
app = connexion.App(__name__, specification_dir='../swagger/')
app.app.json_encoder = JSONEncoder
app.add_api('swagger.yaml')
return app.app

View File

@ -0,0 +1,95 @@
# coding: utf-8
from __future__ import absolute_import
from cic_eth.server.models.model0x_address import Model0xAddress # noqa: E501
from cic_eth.server.models.token import Token # noqa: E501
from cic_eth.server.models.token_balance import TokenBalance # noqa: E501
from cic_eth.server.test import BaseTestCase
from flask import json
from six import BytesIO
class TestAccountController(BaseTestCase):
"""AccountController integration test stubs"""
def test_account_balance(self):
"""Test case for account_balance
"""
query_string = [('address', Model0xAddress()),
('token_symbol', 'token_symbol_example'),
('include_pending', True)]
response = self.client.open(
'/balance',
method='GET',
query_string=query_string)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_create_account_post(self):
"""Test case for create_account_post
"""
query_string = [('password', 'password_example'),
('register', true)]
response = self.client.open(
'/create_account',
method='POST',
query_string=query_string)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_list_transactions(self):
"""Test case for list_transactions
"""
query_string = [('address', Model0xAddress()),
('limit', 10)]
response = self.client.open(
'/transactions',
method='GET',
query_string=query_string)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_transfer(self):
"""Test case for transfer
"""
query_string = [('from_address', Model0xAddress()),
('to_address', Model0xAddress()),
('value', 56),
('token_symbol', 'token_symbol_example')]
response = self.client.open(
'/transfer',
method='POST',
query_string=query_string)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
def test_transfer_from(self):
"""Test case for transfer_from
"""
query_string = [('from_address', Model0xAddress()),
('to_address', Model0xAddress()),
('value', 56),
('token_symbol', 'token_symbol_example'),
('spender_address', Model0xAddress())]
response = self.client.open(
'/transfer_from',
method='POST',
query_string=query_string)
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
if __name__ == '__main__':
import unittest
unittest.main()

View File

@ -0,0 +1,29 @@
# coding: utf-8
from __future__ import absolute_import
from flask import json
from six import BytesIO
from cic_eth.server.models.token import Token # noqa: E501
from cic_eth.server.test import BaseTestCase
class TestTokenController(BaseTestCase):
"""TokenController integration test stubs"""
def test_get_default_token(self):
"""Test case for get_default_token
"""
response = self.client.open(
'/token',
method='GET')
self.assert200(response,
'Response body is : ' + response.data.decode('utf-8'))
if __name__ == '__main__':
import unittest
unittest.main()

View File

@ -0,0 +1,32 @@
# coding: utf-8
import sys
if sys.version_info < (3, 7):
import typing
def is_generic(klass):
""" Determine whether klass is a generic class """
return type(klass) == typing.GenericMeta
def is_dict(klass):
""" Determine whether klass is a Dict """
return klass.__extra__ == dict
def is_list(klass):
""" Determine whether klass is a List """
return klass.__extra__ == list
else:
def is_generic(klass):
""" Determine whether klass is a generic class """
return hasattr(klass, '__origin__')
def is_dict(klass):
""" Determine whether klass is a Dict """
return klass.__origin__ == dict
def is_list(klass):
""" Determine whether klass is a List """
return klass.__origin__ == list

View File

@ -0,0 +1,142 @@
import datetime
import six
import typing
from cic_eth.server import type_util
def _deserialize(data, klass):
"""Deserializes dict, list, str into an object.
:param data: dict, list or str.
:param klass: class literal, or string of class name.
:return: object.
"""
if data is None:
return None
if klass in six.integer_types or klass in (float, str, bool, bytearray):
return _deserialize_primitive(data, klass)
elif klass == object:
return _deserialize_object(data)
elif klass == datetime.date:
return deserialize_date(data)
elif klass == datetime.datetime:
return deserialize_datetime(data)
elif type_util.is_generic(klass):
if type_util.is_list(klass):
return _deserialize_list(data, klass.__args__[0])
if type_util.is_dict(klass):
return _deserialize_dict(data, klass.__args__[1])
else:
return deserialize_model(data, klass)
def _deserialize_primitive(data, klass):
"""Deserializes to primitive type.
:param data: data to deserialize.
:param klass: class literal.
:return: int, long, float, str, bool.
:rtype: int | long | float | str | bool
"""
try:
value = klass(data)
except UnicodeEncodeError:
value = six.u(data)
except TypeError:
value = data
return value
def _deserialize_object(value):
"""Return an original value.
:return: object.
"""
return value
def deserialize_date(string):
"""Deserializes string to date.
:param string: str.
:type string: str
:return: date.
:rtype: date
"""
try:
from dateutil.parser import parse
return parse(string).date()
except ImportError:
return string
def deserialize_datetime(string):
"""Deserializes string to datetime.
The string should be in iso8601 datetime format.
:param string: str.
:type string: str
:return: datetime.
:rtype: datetime
"""
try:
from dateutil.parser import parse
return parse(string)
except ImportError:
return string
def deserialize_model(data, klass):
"""Deserializes list or dict to model.
:param data: dict, list.
:type data: dict | list
:param klass: class literal.
:return: model object.
"""
instance = klass()
if not instance.swagger_types:
return data
for attr, attr_type in six.iteritems(instance.swagger_types):
if data is not None \
and instance.attribute_map[attr] in data \
and isinstance(data, (list, dict)):
value = data[instance.attribute_map[attr]]
setattr(instance, attr, _deserialize(value, attr_type))
return instance
def _deserialize_list(data, boxed_type):
"""Deserializes a list and its elements.
:param data: list to deserialize.
:type data: list
:param boxed_type: class literal.
:return: deserialized list.
:rtype: list
"""
return [_deserialize(sub_data, boxed_type)
for sub_data in data]
def _deserialize_dict(data, boxed_type):
"""Deserializes a dict and its elements.
:param data: dict to deserialize.
:type data: dict
:param boxed_type: class literal.
:return: deserialized dict.
:rtype: dict
"""
return {k: _deserialize(v, boxed_type)
for k, v in six.iteritems(data)}

View File

@ -6,3 +6,6 @@ uwsgi==2.0.19.1
graphene>=2.0
flask>=2.0.2
Flask-GraphQL>=2.0.1
connexion[swagger-ui] == 2.6.0
python_dateutil == 2.6.0
setuptools >= 21.0.0

View File

@ -214,7 +214,7 @@ services:
set -a
if [[ -f /tmp/cic/config/env_reset ]]; then source /tmp/cic/config/env_reset; fi
set +a
python /root/cic_eth/graphql/app.py -vv
python /root/cic_eth/server/__main__.py
cic-eth-tracker:
image: ${DEV_DOCKER_REGISTRY:-registry.gitlab.com/grassrootseconomics/cic-internal-integration}/cic-eth:${TAG:-latest}