Commit 7ce91e7c authored by Bogdan's avatar Bogdan

Merge branch 'support/bugfix-coverage' into develop

parents e58e2506 6cda1589
...@@ -23,31 +23,33 @@ from env_info import is_running_locally, get_resources_path ...@@ -23,31 +23,33 @@ from env_info import is_running_locally, get_resources_path
from flask import request from flask import request
from flask import redirect from flask import redirect
app = connexion.App(__name__, specification_dir='configs/')
@app.app.before_request
def before_request():
if request.url.startswith('http://'):
url = request.url.replace('http://', 'https://', 1)
code = 301
return redirect(url, code=code)
@app.route('/', methods=['GET'])
def api_root():
return redirect('/api/ui')
# SSL configuration
certificate_path = get_resources_path()
context = (os.path.normpath(f'{certificate_path}/articonf1.crt'), os.path.normpath(f'{certificate_path}/articonf1.key')) # certificate and key files
if is_running_locally():
print("Running locally...")
app.add_api(swagger_util.get_bundled_specs(Path("configs/swagger_local.yml")),
resolver = connexion.RestyResolver("cms_rest_api"))
else:
app.add_api(swagger_util.get_bundled_specs(Path("configs/swagger.yml")),
resolver = connexion.RestyResolver("cms_rest_api"))
# start app # start app
if __name__ == '__main__': if __name__ == '__main__':
app = connexion.App(__name__, specification_dir='configs/')
@app.app.before_request
def before_request():
if request.url.startswith('http://'):
url = request.url.replace('http://', 'https://', 1)
code = 301
return redirect(url, code=code)
@app.route('/', methods=['GET'])
def api_root():
return redirect('/api/ui')
# SSL configuration
certificate_path = get_resources_path()
context = (os.path.normpath(f'{certificate_path}/articonf1.crt'), os.path.normpath(f'{certificate_path}/articonf1.key')) # certificate and key files
if is_running_locally():
print("Running locally...")
app.add_api(swagger_util.get_bundled_specs(Path("configs/swagger_local.yml")),
resolver = connexion.RestyResolver("cms_rest_api"))
else:
app.add_api(swagger_util.get_bundled_specs(Path("configs/swagger.yml")),
resolver = connexion.RestyResolver("cms_rest_api"))
app.run(host='0.0.0.0', port=5000, debug=True, use_reloader=False, ssl_context=context) # disable reloader so only subscribed once to rabbitmqö app.run(host='0.0.0.0', port=5000, debug=True, use_reloader=False, ssl_context=context) # disable reloader so only subscribed once to rabbitmqö
\ No newline at end of file
import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
class TestCoverage(unittest.TestCase):
def test_init_main(self):
print("Entered test main")
# python -m unittest discover
# add modules folder to interpreter path
import sys
import os
import prance
from pathlib import Path
from typing import Dict, Any
modules_path = '../../modules/'
if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
# init logging to file
import logging
LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s')
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
LOGGER = logging.getLogger(__name__)
#################################
import connexion
from security import swagger_util
from env_info import is_running_locally, get_resources_path
from flask import request
from flask import redirect
import main #something in main is causing an infinite loop (probably an async task/ listener)
print("Finished test main")
def test_database(self):
print("Entered test db")
import network_constants as netconst
from database.MongoRepositoryBase import MongoRepositoryBase
import pymongo
import json
from db.entities.user import User
from typing import List
from db import repository
print("Finished test db")
def test_services(self):
print("Entered test services")
from functools import wraps
from flask import g, request, redirect, url_for
# global imports (dont't worry, red is normal)
from db.repository import Repository
from db.entities.user import User
from services.user_service import UserService
from env_info import get_resources_path
import jwt
from datetime import datetime, timedelta
from typing import Dict
import bcrypt
from services import login_wrapper
from services import token_service
from services import user_service
print("Finished test services")
def test_routes(self):
print("Entered test routes")
from flask import request, Response
from messaging.ReconnectingMessageManager import ReconnectingMessageManager
import json
from flask import request
# global imports (dont't worry, red is normal)
from db.entities.user import User
from services.user_service import UserService
from services.login_wrapper import login_required
from services.token_service import TokenService
import bcrypt
import jwt
from routes import user
#from routes import blockchain_trace #message_sender in blockchain_trace is causing an infinite loop (probabily an async task//listener)
from routes import debug
print("Finished test routes")
def test_add_users(self):
print("Entered test users")
# add modules folder to interpreter path
import sys
import os
import json
import prance
from pathlib import Path
from typing import Dict, Any
modules_path = '../../modules/'
if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
from services.user_service import UserService
from env_info import is_running_locally, get_resources_path
import add_users
print("Finished test users")
if __name__ == '__main__':
unittest.main()
...@@ -27,7 +27,7 @@ from flask import request ...@@ -27,7 +27,7 @@ from flask import request
from flask import redirect from flask import redirect
# init message handler # init message handler
message_handler = MessageHandler(Repository(), ReconnectingMessageManager.getInstance(), RestFetcher()) message_handler:MessageHandler = None
def message_received_callback(channel, method, properties, body): def message_received_callback(channel, method, properties, body):
message_handler.handle_generic(body) message_handler.handle_generic(body)
...@@ -59,6 +59,7 @@ else: ...@@ -59,6 +59,7 @@ else:
# start app # start app
if __name__ == '__main__': if __name__ == '__main__':
message_handler = MessageHandler(Repository(), ReconnectingMessageManager.getInstance(), RestFetcher())
message_manager = ReconnectingMessageManager.getInstance() message_manager = ReconnectingMessageManager.getInstance()
message_manager.start_consuming('inhub', 'direct', 'trace-retrieval', True, message_received_callback) message_manager.start_consuming('inhub', 'direct', 'trace-retrieval', True, message_received_callback)
......
import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
class TestCoverage(unittest.TestCase):
def test_init_main(self):
# python -m unittest discover
# add modules folder to interpreter path
import sys
import os
import prance
from pathlib import Path
modules_path = '../../../modules/'
if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
# init logging to file
import logging
LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s')
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
LOGGER = logging.getLogger(__name__)
#############################
import connexion
from security import swagger_util
from env_info import is_running_locally, get_resources_path
from database.repository import Repository
from messaging.MessageHandler import MessageHandler
from messaging.ReconnectingMessageManager import ReconnectingMessageManager
from messaging.rest_fetcher import RestFetcher
from flask import request
from flask import redirect
import main
def test_database(self):
# global imports (dont't worry, red is normal)
from typing import List, Dict
import network_constants as netconst
from database.MongoRepositoryBase import MongoRepositoryBase
from database.entities.transaction import Transaction
import pymongo
import json
import time
from database import repository
def test_messaging(self):
from security.token_manager import TokenManager
import network_constants
from database.entities.transaction import Transaction
from database.repository import Repository
from messaging.rest_fetcher import RestFetcher
import json
import hashlib
import logging
import requests
from typing import Dict
from typing import List
from messaging import MessageHandler
from messaging import rest_fetcher
def test_routes(self):
#global imports
from database.entities.transaction import Transaction
from database.repository import Repository
import json
from flask import Response, request
from routes import transactions
if __name__ == '__main__':
unittest.main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment