Skip to content

S
SMART
Commit c94d22b5 authored by Alfonso Orta's avatar Alfonso Orta
Browse files
Options

Merge branch 'develop' into 'staging' 

Exception Handling for Coverage Tests

See merge request !29
parents 8423945d 2f7e09be
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 1258 additions and 959 deletions
src/data-hub/role-stage-discovery-microservice/app/test_coverage.py
View file @ c94d22b5
...@@ -4,90 +4,129 @@ for path in ['../', './']: ...@@ -4,90 +4,129 @@ for path in ['../', './']:
sys.path.insert(1, path) sys.path.insert(1, path)
try:
class TestCoverage(unittest.TestCase): class TestCoverage(unittest.TestCase):
def test_init_main(self): def test_init_main(self):
# python -m unittest discover try:
from db.entities import Cluster # python -m unittest discover
from db.entities import Cluster
from datetime import date, datetime
import json from datetime import date, datetime
import json
# add modules folder to interpreter path
import sys # add modules folder to interpreter path
import os import sys
modules_path = '../../../modules/' import os
if os.path.exists(modules_path): modules_path = '../../../modules/'
sys.path.insert(1, modules_path) if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
### init logging ###
import logging ### init logging ###
LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s') import logging
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
LOGGER = logging.getLogger(__name__)
#############################
import connexion #############################
from security import swagger_util import connexion
from pathlib import Path from security import swagger_util
from env_info import is_running_locally, get_resources_path from pathlib import Path
from flask import request from env_info import is_running_locally, get_resources_path
from flask import redirect from flask import request
import main #error when importing main, ModuleNotFoundError: No module named 'security' from flask import redirect
#exec(open('main.py').read()) except Exception as e:
print ("Exception found:")
def test_init_run_clustering(self): print (e)
import sys try:
import os import main #error when importing main, ModuleNotFoundError: No module named 'security'
modules_path = '../../../modules/' #exec(open('main.py').read())
if os.path.exists(modules_path): except Exception as e:
sys.path.insert(1, modules_path) print ("Exception found:")
print (e)
import json
from db.entities import Layer, Cluster def test_init_run_clustering(self):
from typing import List, Dict, Tuple, Any try:
from db.repository import Repository import sys
from processing.clustering import Clusterer, ClusterResult import os
modules_path = '../../../modules/'
import run_clustering if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
def test_init_run_node(self):
import sys import json
import os from db.entities import Layer, Cluster
modules_path = '../../../modules/' from typing import List, Dict, Tuple, Any
if os.path.exists(modules_path): from db.repository import Repository
sys.path.insert(1, modules_path) from processing.clustering import Clusterer, ClusterResult
except Exception as e:
import json print ("Exception found:")
import urllib3 print (e)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
try:
import processing.fetching.fetching as f import run_clustering
import run_node_fetching except Exception as e:
print ("Exception found:")
def test_init_run_similarity(self): print (e)
import processing.similarityFiles.similarityMain as SimilarityCalc
from db.repository import Repository def test_init_run_node(self):
try:
import run_similarity_calc import sys
import os
def test_init_run_time(self): modules_path = '../../../modules/'
import sys if os.path.exists(modules_path):
import os sys.path.insert(1, modules_path)
modules_path = '../../../modules/'
if os.path.exists(modules_path): import json
sys.path.insert(1, modules_path) import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
import json except Exception as e:
from datetime import datetime, date print ("Exception found:")
from db.repository import Repository print (e)
from db.entities import ClusterSet, Cluster, Layer, TimeSlice
from typing import Tuple, Dict, Any, List try:
import processing.fetching.fetching as f
import run_time_slicing import run_node_fetching
except Exception as e:
print ("Exception found:")
if __name__ == '__main__': print (e)
unittest.main()
def test_init_run_similarity(self):
try:
import processing.similarityFiles.similarityMain as SimilarityCalc
from db.repository import Repository
import run_similarity_calc
except Exception as e:
print ("Exception found:")
print (e)
def test_init_run_time(self):
try:
import sys
import os
modules_path = '../../../modules/'
if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
import json
from datetime import datetime, date
from db.repository import Repository
from db.entities import ClusterSet, Cluster, Layer, TimeSlice
from typing import Tuple, Dict, Any, List
except Exception as e:
print ("Exception found:")
print (e)
try:
import run_time_slicing
except Exception as e:
print ("Exception found:")
print (e)
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
\ No newline at end of file
src/data-hub/role-stage-discovery-microservice/app/tests/test_cluster.py
View file @ c94d22b5
import unittest try:
import sys import unittest
for path in ['../', './']: import sys
sys.path.insert(1, path) for path in ['../', './']:
sys.path.insert(1, path)
# python -m unittest discover # python -m unittest discover
from db.entities import Cluster from db.entities import Cluster
from datetime import date, datetime from datetime import date, datetime
import json import json
class TestCluster(unittest.TestCase): class TestCluster(unittest.TestCase):
def test_init_Cluster(self): def test_init_Cluster(self):
c = Cluster('debug', 'debug-table1', 'layer1', 1, [1, 2, 3]) try:
c = Cluster('debug', 'debug-table1', 'layer1', 1, [1, 2, 3])
self.assertEqual('debug', c.use_case) self.assertEqual('debug', c.use_case)
self.assertEqual('debug-table1', c.use_case_table) self.assertEqual('debug-table1', c.use_case_table)
self.assertEqual(1, c.cluster_label) self.assertEqual(1, c.cluster_label)
self.assertEqual([1, 2, 3], c.nodes) self.assertEqual([1, 2, 3], c.nodes)
except Exception as e:
print(e)
if __name__ == '__main__':
if __name__ == '__main__': unittest.main()
unittest.main() except Exception as e:
print ("Exception found:")
print (e)
src/data-hub/role-stage-discovery-microservice/app/tests/test_cluster_result.py
View file @ c94d22b5
import unittest try:
import sys import unittest
for path in ['../', './']: import sys
sys.path.insert(1, path) for path in ['../', './']:
sys.path.insert(1, path)
# python -m unittest discover
from processing.clustering import ClusterResultConverter, ClusterResult # python -m unittest discover
from typing import List, Dict, Any from processing.clustering import ClusterResultConverter, ClusterResult
from typing import List, Dict, Any
class TestClusterResult(unittest.TestCase):
converter:ClusterResultConverter = None class TestClusterResult(unittest.TestCase):
converter:ClusterResultConverter = None
def setUp(self):
self.converter = ClusterResultConverter() def setUp(self):
try:
def test_result_undefined_feature(self): self.converter = ClusterResultConverter()
cluster_groups = self._get_some_cluster_groups_1d() except Exception as e:
cluster_res = self.converter.convert_to_cluster_results( print (e)
cluster_groups=cluster_groups,
features=[] def test_result_undefined_feature(self):
) try:
cluster_groups = self._get_some_cluster_groups_1d()
self.assert_correct_cluster_result_len(cluster_groups, cluster_res) cluster_res = self.converter.convert_to_cluster_results(
self.assert_correct_cluster_result_labels(['n.a.','n.a.','n.a.'], cluster_res) cluster_groups=cluster_groups,
features=[]
def test_result_1d_feature(self): )
cluster_groups = self._get_some_cluster_groups_1d()
cluster_res = self.converter.convert_to_cluster_results( self.assert_correct_cluster_result_len(cluster_groups, cluster_res)
cluster_groups=cluster_groups, self.assert_correct_cluster_result_labels(['n.a.','n.a.','n.a.'], cluster_res)
features=['v'] except Exception as e:
) print (e)
self.assert_correct_cluster_result_len(cluster_groups, cluster_res) def test_result_1d_feature(self):
self.assert_correct_cluster_result_labels(['-1.0 -- 1.0','10.0 -- 11.0','2.0 -- 2.0'], cluster_res) try:
cluster_groups = self._get_some_cluster_groups_1d()
def test_result_2d_features(self): cluster_res = self.converter.convert_to_cluster_results(
cluster_groups = self._get_some_cluster_groups_2d() cluster_groups=cluster_groups,
cluster_res = self.converter.convert_to_cluster_results( features=['v']
cluster_groups=cluster_groups, )
features=['v', 'u']
)
self.assert_correct_cluster_result_len(cluster_groups, cluster_res)
self.assert_correct_cluster_result_len(cluster_groups, cluster_res) self.assert_correct_cluster_result_labels(['-1.0 -- 1.0','10.0 -- 11.0','2.0 -- 2.0'], cluster_res)
self.assert_correct_cluster_result_labels([str((0.0,0.0)), str((10.5,10.5)), str((2.0,2.0)), str((3.0,6.0))], cluster_res)
except Exception as e:
print (e)
#region Custom Assertions
def test_result_2d_features(self):
def assert_correct_cluster_result_len(self, expected: 'original dict of lists', actual: Dict[Any, ClusterResult]):
self.assertEqual(len(expected), len(actual)) try:
for i in range(len(expected)): cluster_groups = self._get_some_cluster_groups_2d()
self.assertEqual(len(expected[i]), len(actual[i].nodes)) cluster_res = self.converter.convert_to_cluster_results(
self.assertEqual(expected[i], actual[i].nodes) cluster_groups=cluster_groups,
features=['v', 'u']
def assert_correct_cluster_result_labels(self, expected: List[str], actual: Dict[Any, ClusterResult]): )
self.assertEqual(len(expected), len(actual))
for i in range(len(expected)): self.assert_correct_cluster_result_len(cluster_groups, cluster_res)
self.assertEqual(expected[i], actual[i].label) self.assert_correct_cluster_result_labels([str((0.0,0.0)), str((10.5,10.5)), str((2.0,2.0)), str((3.0,6.0))], cluster_res)
#endregion Custom Assertions except Exception as e:
print (e)
#region helper methods
def _get_some_cluster_groups_1d(self):
return { #region Custom Assertions
0: [{'v':'0'}, {'v':'1'}, {'v':'-1'}],
1: [{'v':'10'}, {'v':'11'}], def assert_correct_cluster_result_len(self, expected: 'original dict of lists', actual: Dict[Any, ClusterResult]):
2: [{'v':'2'}], try:
} self.assertEqual(len(expected), len(actual))
for i in range(len(expected)):
def _get_some_cluster_groups_2d(self): self.assertEqual(len(expected[i]), len(actual[i].nodes))
return { self.assertEqual(expected[i], actual[i].nodes)
0: [{'v':'0', 'u':'0'}, {'v':'1', 'u':'1'}, {'v':'-1', 'u':'-1'}],
1: [{'v':'10', 'u':'10'}, {'v':'11', 'u':'11'}], except Exception as e:
2: [{'v':'2', 'u':'2'}], print (e)
3: [{'v':'7', 'u':'7'}, {'v':'5', 'u':'3'}, {'v':'-3', 'u':'8'}],
} def assert_correct_cluster_result_labels(self, expected: List[str], actual: Dict[Any, ClusterResult]):
#endregion helper methods try:
self.assertEqual(len(expected), len(actual))
if __name__ == '__main__': for i in range(len(expected)):
unittest.main() self.assertEqual(expected[i], actual[i].label)
\ No newline at end of file except Exception as e:
print (e)
#endregion Custom Assertions
#region helper methods
def _get_some_cluster_groups_1d(self):
return {
0: [{'v':'0'}, {'v':'1'}, {'v':'-1'}],
1: [{'v':'10'}, {'v':'11'}],
2: [{'v':'2'}],
}
def _get_some_cluster_groups_2d(self):
return {
0: [{'v':'0', 'u':'0'}, {'v':'1', 'u':'1'}, {'v':'-1', 'u':'-1'}],
1: [{'v':'10', 'u':'10'}, {'v':'11', 'u':'11'}],
2: [{'v':'2', 'u':'2'}],
3: [{'v':'7', 'u':'7'}, {'v':'5', 'u':'3'}, {'v':'-3', 'u':'8'}],
}
#endregion helper methods
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
\ No newline at end of file
src/data-hub/semantic-linking-microservice/app/test_coverage.py
View file @ c94d22b5
...@@ -4,71 +4,112 @@ for path in ['../', './']: ...@@ -4,71 +4,112 @@ for path in ['../', './']:
sys.path.insert(1, path) sys.path.insert(1, path)
try:
class TestCoverage(unittest.TestCase): class TestCoverage(unittest.TestCase):
def test_init_main(self): def test_init_main(self):
# add modules folder to interpreter path try:
import sys # add modules folder to interpreter path
import os import sys
import prance import os
from pathlib import Path import prance
modules_path = '../../../modules/' from pathlib import Path
if os.path.exists(modules_path): modules_path = '../../../modules/'
sys.path.insert(1, modules_path) if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
# init logging to file
import logging # init logging to file
LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s') import logging
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
LOGGER = logging.getLogger(__name__)
#################################
import connexion #################################
from security import swagger_util import connexion
from env_info import is_running_locally, get_resources_path from security import swagger_util
from messaging.ReconnectingMessageManager import ReconnectingMessageManager from env_info import is_running_locally, get_resources_path
from messaging.MessageHandler import MessageHandler from messaging.ReconnectingMessageManager import ReconnectingMessageManager
from flask import request from messaging.MessageHandler import MessageHandler
from flask import redirect from flask import request
from flask import redirect
# init message handler
from db.repository import Repository # init message handler
from db.repository import Repository
import main except Exception as e:
print ("Exception found:")
def test_routes(self): print (e)
from routes import debug try:
from routes import layers
from routes import nodes import main
def test_messaging(self): except Exception as e:
import network_constants as netconst print ("Exception found:")
from security.token_manager import TokenManager print (e)
from db.entities import Layer
def test_routes(self):
import json try:
import requests from routes import debug
from typing import Dict, List except Exception as e:
from threading import Thread print ("Exception found:")
print (e)
import logging
from messaging import MessageHandler try:
from routes import layers
def test_db(self): except Exception as e:
import network_constants as netconst print ("Exception found:")
from database.MongoRepositoryBase import MongoRepositoryBase print (e)
from db.entities import Layer
try:
import pymongo from routes import nodes
import json except Exception as e:
from typing import List, Dict print ("Exception found:")
print (e)
# init logging to file
import logging def test_messaging(self):
try:
from db import repository import network_constants as netconst
from db.entities import layer from security.token_manager import TokenManager
from db.entities import Layer
if __name__ == '__main__': import json
unittest.main() import requests
from typing import Dict, List
from threading import Thread
import logging
except Exception as e:
print ("Exception found:")
print (e)
try:
from messaging import MessageHandler
except Exception as e:
print ("Exception found:")
print (e)
def test_db(self):
try:
import network_constants as netconst
from database.MongoRepositoryBase import MongoRepositoryBase
from db.entities import Layer
import pymongo
import json
from typing import List, Dict
# init logging to file
import logging
except Exception as e:
print ("Exception found:")
print (e)
try:
from db import repository
from db.entities import layer
except Exception as e:
print ("Exception found:")
print (e)
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
\ No newline at end of file
src/participation-hub/business-logic-microservice/app/test_coverage.py
View file @ c94d22b5
...@@ -4,91 +4,145 @@ for path in ['../', './']: ...@@ -4,91 +4,145 @@ for path in ['../', './']:
sys.path.insert(1, path) sys.path.insert(1, path)
try:
class TestCoverage(unittest.TestCase): class TestCoverage(unittest.TestCase):
def test_init_main(self): def test_init_main(self):
# python -m unittest discover try:
# add modules folder to interpreter path # python -m unittest discover
import sys # add modules folder to interpreter path
import os import sys
from pathlib import Path import os
from typing import Dict, Any from pathlib import Path
from typing import Dict, Any
modules_path = '../../../modules/'
if os.path.exists(modules_path): modules_path = '../../../modules/'
sys.path.insert(1, modules_path) if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
# load swagger config
import connexion # load swagger config
from security import swagger_util import connexion
from env_info import is_running_locally, get_resources_path from security import swagger_util
from flask import request from env_info import is_running_locally, get_resources_path
from flask import redirect from flask import request
from flask import redirect
app = connexion.App(__name__, specification_dir='configs/')
app = connexion.App(__name__, specification_dir='configs/')
from db.entities.layer_adapter import LayerAdapter
import main from db.entities.layer_adapter import LayerAdapter
except Exception as e:
def test_db_main(self): print ("Exception found:")
import network_constants as netconst print (e)
from database.MongoRepositoryBase import MongoRepositoryBase
from db.entities import layer_adapter try:
from db.entities import table import main
from db.entities import use_case except Exception as e:
print ("Exception found:")
import pymongo print (e)
import json
from typing import List, Dict def test_db_main(self):
try:
from db import repository import network_constants as netconst
from db import table_repository from database.MongoRepositoryBase import MongoRepositoryBase
from db import use_case_repository from db.entities import layer_adapter
from db.entities import table
from db.entities import use_case
def test_routes(self):
from routes import layer import pymongo
from routes import tables import json
from routes import use_case from typing import List, Dict
except Exception as e:
def test_services(self): print ("Exception found:")
from services import layer_adapter_service print (e)
def test_use_case_scripts(self): try:
import network_constants as nc from db import repository
from security.token_manager import TokenManager from db import table_repository
import requests from db import use_case_repository
from typing import List except Exception as e:
from _add_use_case_scripts import requestPost print ("Exception found:")
####### print (e)
#from _add_use_case_scripts.bank-app import add_bank_app_schema ##eror not importing? invalid folder name?
#from _add_use_case_scripts.bank-app.tables import add_bank_app_schema def test_routes(self):
try:
from _add_use_case_scripts.car_sharing import add_carsharing_schema from routes import layer
from _add_use_case_scripts.car_sharing.tables import add_car except Exception as e:
from _add_use_case_scripts.car_sharing.tables import add_hash print ("Exception found:")
from _add_use_case_scripts.car_sharing.tables import add_offer print (e)
from _add_use_case_scripts.car_sharing.tables import add_publication try:
from _add_use_case_scripts.car_sharing.tables import add_travel from routes import tables
from _add_use_case_scripts.car_sharing.tables import add_user except Exception as e:
print ("Exception found:")
from _add_use_case_scripts.crowd_journalism import add_crowdjournalism_schema print (e)
from _add_use_case_scripts.crowd_journalism.tables import add_classification try:
from _add_use_case_scripts.crowd_journalism.tables import add_event from routes import use_case
from _add_use_case_scripts.crowd_journalism.tables import add_purchase except Exception as e:
from _add_use_case_scripts.crowd_journalism.tables import add_tag print ("Exception found:")
from _add_use_case_scripts.crowd_journalism.tables import add_video print (e)
from _add_use_case_scripts.debug import add_debug_schema def test_services(self):
from _add_use_case_scripts.debug.tables import add_pizza_table try:
from services import layer_adapter_service
#from _add_use_case_scripts.smart-energy import add_smart_energy_schema except Exception as e:
#from _add_use_case_scripts.smart-energy.tables import add_smart_energy print ("Exception found:")
print (e)
from _add_use_case_scripts.vialog import add_vialog_schema
from _add_use_case_scripts.vialog.tables import add_user def test_use_case_scripts(self):
from _add_use_case_scripts.vialog.tables import add_video try:
import network_constants as nc
if __name__ == '__main__': from security.token_manager import TokenManager
unittest.main() import requests
from typing import List
from _add_use_case_scripts import requestPost
except Exception as e:
print ("Exception found:")
print (e)
#######
#from _add_use_case_scripts.bank-app import add_bank_app_schema ##eror not importing? invalid folder name?
#from _add_use_case_scripts.bank-app.tables import add_bank_app_schema
try:
from _add_use_case_scripts.car_sharing import add_carsharing_schema
from _add_use_case_scripts.car_sharing.tables import add_car
from _add_use_case_scripts.car_sharing.tables import add_hash
from _add_use_case_scripts.car_sharing.tables import add_offer
from _add_use_case_scripts.car_sharing.tables import add_publication
from _add_use_case_scripts.car_sharing.tables import add_travel
from _add_use_case_scripts.car_sharing.tables import add_user
except Exception as e:
print ("Exception found:")
print (e)
try:
from _add_use_case_scripts.crowd_journalism import add_crowdjournalism_schema
from _add_use_case_scripts.crowd_journalism.tables import add_classification
from _add_use_case_scripts.crowd_journalism.tables import add_event
from _add_use_case_scripts.crowd_journalism.tables import add_purchase
from _add_use_case_scripts.crowd_journalism.tables import add_tag
from _add_use_case_scripts.crowd_journalism.tables import add_video
except Exception as e:
print ("Exception found:")
print (e)
try:
from _add_use_case_scripts.debug import add_debug_schema
from _add_use_case_scripts.debug.tables import add_pizza_table
except Exception as e:
print ("Exception found:")
print (e)
#from _add_use_case_scripts.smart-energy import add_smart_energy_schema
#from _add_use_case_scripts.smart-energy.tables import add_smart_energy
try:
from _add_use_case_scripts.vialog import add_vialog_schema
from _add_use_case_scripts.vialog.tables import add_user
from _add_use_case_scripts.vialog.tables import add_video
except Exception as e:
print ("Exception found:")
print (e)
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
\ No newline at end of file
src/rest-gateway/app/test_coverage.py
View file @ c94d22b5
...@@ -2,114 +2,173 @@ import unittest ...@@ -2,114 +2,173 @@ import unittest
import sys import sys
for path in ['../', './']: for path in ['../', './']:
sys.path.insert(1, path) sys.path.insert(1, path)
try:
class TestCoverage(unittest.TestCase): class TestCoverage(unittest.TestCase):
def test_init_main(self): def test_init_main(self):
print("Entered test main") try:
# python -m unittest discover print("Entered test main")
# add modules folder to interpreter path # python -m unittest discover
import sys # add modules folder to interpreter path
import os import sys
import prance import os
from pathlib import Path import prance
from typing import Dict, Any from pathlib import Path
from typing import Dict, Any
modules_path = '../../modules/'
if os.path.exists(modules_path): modules_path = '../../modules/'
sys.path.insert(1, modules_path) if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
# init logging to file
import logging # init logging to file
LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s') import logging
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
LOGGER = logging.getLogger(__name__)
################################# except Exception as e:
print ("Exception found:")
import connexion print (e)
from security import swagger_util #################################
from env_info import is_running_locally, get_resources_path try:
from flask import request import connexion
from flask import redirect from security import swagger_util
from env_info import is_running_locally, get_resources_path
import main #something in main is causing an infinite loop (probably an async task/ listener) from flask import request
print("Finished test main") from flask import redirect
except Exception as e:
def test_database(self): print ("Exception found:")
print("Entered test db") print (e)
import network_constants as netconst try:
from database.MongoRepositoryBase import MongoRepositoryBase import main #something in main is causing an infinite loop (probably an async task/ listener)
except Exception as e:
import pymongo print ("Exception found:")
import json print (e)
from db.entities.user import User print("Finished test main")
from typing import List
from db import repository def test_database(self):
try:
print("Finished test db") print("Entered test db")
import network_constants as netconst
def test_services(self): from database.MongoRepositoryBase import MongoRepositoryBase
print("Entered test services")
from functools import wraps import pymongo
from flask import g, request, redirect, url_for import json
from db.entities.user import User
# global imports (dont't worry, red is normal) from typing import List
from db.repository import Repository except Exception as e:
from db.entities.user import User print ("Exception found:")
from services.user_service import UserService print (e)
from env_info import get_resources_path
try:
import jwt from db import repository
from datetime import datetime, timedelta except Exception as e:
from typing import Dict print ("Exception found:")
import bcrypt print (e)
from services import login_wrapper print("Finished test db")
from services import token_service
from services import user_service def test_services(self):
print("Finished test services") try:
print("Entered test services")
def test_routes(self): from functools import wraps
print("Entered test routes") from flask import g, request, redirect, url_for
from flask import request, Response
from messaging.ReconnectingMessageManager import ReconnectingMessageManager # global imports (dont't worry, red is normal)
import json from db.repository import Repository
from flask import request from db.entities.user import User
# global imports (dont't worry, red is normal) from services.user_service import UserService
from db.entities.user import User from env_info import get_resources_path
from services.user_service import UserService
from services.login_wrapper import login_required import jwt
from services.token_service import TokenService from datetime import datetime, timedelta
import bcrypt from typing import Dict
import jwt import bcrypt
except Exception as e:
print ("Exception found:")
print (e)
from routes import user
#from routes import blockchain_trace #message_sender in blockchain_trace is causing an infinite loop (probabily an async task//listener) try:
from routes import debug from services import login_wrapper
print("Finished test routes") except Exception as e:
print ("Exception found:")
def test_add_users(self): print (e)
print("Entered test users")
# add modules folder to interpreter path try:
import sys from services import token_service
import os except Exception as e:
import json print ("Exception found:")
import prance print (e)
from pathlib import Path try:
from typing import Dict, Any from services import user_service
except Exception as e:
modules_path = '../../modules/' print ("Exception found:")
if os.path.exists(modules_path): print (e)
sys.path.insert(1, modules_path) print("Finished test services")
from services.user_service import UserService def test_routes(self):
from env_info import is_running_locally, get_resources_path try:
import add_users print("Entered test routes")
print("Finished test users") from flask import request, Response
from messaging.ReconnectingMessageManager import ReconnectingMessageManager
import json
from flask import request
if __name__ == '__main__': # global imports (dont't worry, red is normal)
unittest.main() 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
except Exception as e:
print ("Exception found:")
print (e)
try:
from routes import user
except Exception as e:
print ("Exception found:")
print (e)
#from routes import blockchain_trace #message_sender in blockchain_trace is causing an infinite loop (probabily an async task//listener)
try:
from routes import debug
except Exception as e:
print ("Exception found:")
print (e)
print("Finished test routes")
def test_add_users(self):
try:
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
except Exception as e:
print ("Exception found:")
print (e)
try:
import add_users
except Exception as e:
print ("Exception found:")
print (e)
print("Finished test users")
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
\ No newline at end of file
src/transaction-hub-in/trace-retrieval-microservice/app/test_coverage.py
View file @ c94d22b5
...@@ -3,83 +3,146 @@ import sys ...@@ -3,83 +3,146 @@ import sys
for path in ['../', './']: for path in ['../', './']:
sys.path.insert(1, path) sys.path.insert(1, path)
try:
class TestCoverage(unittest.TestCase): class TestCoverage(unittest.TestCase):
def test_init_main(self): def test_init_main(self):
# python -m unittest discover try:
# add modules folder to interpreter path # python -m unittest discover
import sys # add modules folder to interpreter path
import os import sys
import prance import os
from pathlib import Path import prance
from pathlib import Path
modules_path = '../../../modules/'
if os.path.exists(modules_path): modules_path = '../../../modules/'
sys.path.insert(1, modules_path) if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
# init logging to file
import logging # init logging to file
LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s') import logging
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) LOG_FORMAT = ('%(levelname) -5s %(asctime)s %(name)s:%(funcName) -35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
LOGGER = logging.getLogger(__name__)
############################# except Exception as e:
import connexion print ("Exception found:")
from security import swagger_util print (e)
from env_info import is_running_locally, get_resources_path
#############################
from database.repository import Repository try:
from messaging.MessageHandler import MessageHandler import connexion
from messaging.ReconnectingMessageManager import ReconnectingMessageManager from security import swagger_util
from messaging.rest_fetcher import RestFetcher from env_info import is_running_locally, get_resources_path
from flask import request except Exception as e:
from flask import redirect print ("Exception found:")
import main print (e)
def test_database(self): try:
# global imports (dont't worry, red is normal) from database.repository import Repository
from typing import List, Dict except Exception as e:
import network_constants as netconst print ("Exception found:")
from database.MongoRepositoryBase import MongoRepositoryBase print (e)
from database.entities.transaction import Transaction
try:
import pymongo from messaging.MessageHandler import MessageHandler
import json except Exception as e:
import time print ("Exception found:")
print (e)
from database import repository
try:
def test_messaging(self): from messaging.ReconnectingMessageManager import ReconnectingMessageManager
except Exception as e:
from security.token_manager import TokenManager print ("Exception found:")
import network_constants print (e)
from database.entities.transaction import Transaction
from database.repository import Repository try:
from messaging.rest_fetcher import RestFetcher from messaging.rest_fetcher import RestFetcher
from flask import request
import json from flask import redirect
import hashlib except Exception as e:
import logging print ("Exception found:")
import requests print (e)
try:
from typing import Dict import main
from typing import List except Exception as e:
print ("Exception found:")
from messaging import MessageHandler print (e)
from messaging import rest_fetcher
def test_database(self):
def test_routes(self): # global imports (dont't worry, red is normal)
#global imports try:
from database.entities.transaction import Transaction from typing import List, Dict
from database.repository import Repository import network_constants as netconst
from database.MongoRepositoryBase import MongoRepositoryBase
import json from database.entities.transaction import Transaction
from flask import Response, request
import pymongo
import json
from routes import transactions import time
except Exception as e:
print ("Exception found:")
print (e)
if __name__ == '__main__': try:
unittest.main() from database import repository
except Exception as e:
print ("Exception found:")
print (e)
def test_messaging(self):
try:
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
except Exception as e:
print ("Exception found:")
print (e)
try:
from messaging import MessageHandler
except Exception as e:
print ("Exception found:")
print (e)
try:
from messaging import rest_fetcher
except Exception as e:
print ("Exception found:")
print (e)
def test_routes(self):
#global imports
try:
from database.entities.transaction import Transaction
from database.repository import Repository
except Exception as e:
print ("Exception found:")
print (e)
try:
import json
from flask import Response, request
except Exception as e:
print ("Exception found:")
print (e)
try:
from routes import transactions
except Exception as e:
print ("Exception found:")
print (e)
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
\ No newline at end of file
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