Skip to content

S
SMART
Commit 3d79e088 authored by Bogdan's avatar Bogdan
Browse files
Options

Error Handling

parent a919d413
Hide whitespace changes
Inline Side-by-side
Showing with 596 additions and 434 deletions
src/data-hub/role-stage-discovery-microservice/app/test_coverage.py
View file @ 3d79e088
...@@ -4,90 +4,119 @@ for path in ['../', './']: ...@@ -4,90 +4,119 @@ 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:
pass
def test_init_run_clustering(self):
try:
import sys import main #error when importing main, ModuleNotFoundError: No module named 'security'
import os #exec(open('main.py').read())
modules_path = '../../../modules/' except:
if os.path.exists(modules_path): pass
sys.path.insert(1, modules_path)
def test_init_run_clustering(self):
import json try:
from db.entities import Layer, Cluster import sys
from typing import List, Dict, Tuple, Any import os
from db.repository import Repository modules_path = '../../../modules/'
from processing.clustering import Clusterer, ClusterResult if os.path.exists(modules_path):
sys.path.insert(1, modules_path)
import run_clustering
import json
def test_init_run_node(self): from db.entities import Layer, Cluster
import sys from typing import List, Dict, Tuple, Any
import os from db.repository import Repository
modules_path = '../../../modules/' from processing.clustering import Clusterer, ClusterResult
if os.path.exists(modules_path): except:
sys.path.insert(1, modules_path) pass
import json try:
import urllib3 import run_clustering
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) except:
pass
import processing.fetching.fetching as f
import run_node_fetching def test_init_run_node(self):
try:
def test_init_run_similarity(self): import sys
import processing.similarityFiles.similarityMain as SimilarityCalc import os
from db.repository import Repository modules_path = '../../../modules/'
if os.path.exists(modules_path):
import run_similarity_calc sys.path.insert(1, modules_path)
def test_init_run_time(self): import json
import sys import urllib3
import os urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
modules_path = '../../../modules/' except:
if os.path.exists(modules_path): pass
sys.path.insert(1, modules_path)
try:
import json import processing.fetching.fetching as f
from datetime import datetime, date import run_node_fetching
from db.repository import Repository except:
from db.entities import ClusterSet, Cluster, Layer, TimeSlice pass
from typing import Tuple, Dict, Any, List
def test_init_run_similarity(self):
import run_time_slicing try:
import processing.similarityFiles.similarityMain as SimilarityCalc
from db.repository import Repository
if __name__ == '__main__':
unittest.main() import run_similarity_calc
except:
pass
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:
pass
try:
import run_time_slicing
except:
pass
if __name__ == '__main__':
unittest.main()
except:
pass
\ No newline at end of file
src/data-hub/semantic-linking-microservice/app/test_coverage.py
View file @ 3d79e088
...@@ -4,71 +4,102 @@ for path in ['../', './']: ...@@ -4,71 +4,102 @@ 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): try:
# add modules folder to interpreter path def test_init_main(self):
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 try:
def test_routes(self): import main
from routes import debug
from routes import layers except:
from routes import nodes pass
except:
def test_messaging(self): pass
import network_constants as netconst
from security.token_manager import TokenManager def test_routes(self):
from db.entities import Layer try:
from routes import debug
import json except:
import requests pass
from typing import Dict, List
from threading import Thread try:
from routes import layers
import logging except:
from messaging import MessageHandler pass
def test_db(self): try:
import network_constants as netconst from routes import nodes
from database.MongoRepositoryBase import MongoRepositoryBase except:
from db.entities import Layer pass
import pymongo def test_messaging(self):
import json try:
from typing import List, Dict import network_constants as netconst
from security.token_manager import TokenManager
# init logging to file from db.entities import Layer
import logging
import json
from db import repository import requests
from db.entities import layer from typing import Dict, List
from threading import Thread
if __name__ == '__main__': import logging
unittest.main() except:
pass
try:
from messaging import MessageHandler
except:
pass
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:
pass
try:
from db import repository
from db.entities import layer
except:
pass
if __name__ == '__main__':
unittest.main()
except:
pass
\ No newline at end of file
src/participation-hub/business-logic-microservice/app/test_coverage.py
View file @ 3d79e088
...@@ -4,91 +4,131 @@ for path in ['../', './']: ...@@ -4,91 +4,131 @@ 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:
def test_db_main(self): pass
import network_constants as netconst
from database.MongoRepositoryBase import MongoRepositoryBase try:
from db.entities import layer_adapter import main
from db.entities import table except:
from db.entities import use_case pass
import pymongo def test_db_main(self):
import json try:
from typing import List, Dict import network_constants as netconst
from database.MongoRepositoryBase import MongoRepositoryBase
from db import repository from db.entities import layer_adapter
from db import table_repository from db.entities import table
from db import use_case_repository from db.entities import use_case
import pymongo
def test_routes(self): import json
from routes import layer from typing import List, Dict
from routes import tables except:
from routes import use_case pass
def test_services(self): try:
from services import layer_adapter_service from db import repository
from db import table_repository
def test_use_case_scripts(self): from db import use_case_repository
import network_constants as nc except:
from security.token_manager import TokenManager pass
import requests
from typing import List
from _add_use_case_scripts import requestPost def test_routes(self):
####### try:
from routes import layer
#from _add_use_case_scripts.bank-app import add_bank_app_schema ##eror not importing? invalid folder name? except:
#from _add_use_case_scripts.bank-app.tables import add_bank_app_schema pass
try:
from _add_use_case_scripts.car_sharing import add_carsharing_schema from routes import tables
from _add_use_case_scripts.car_sharing.tables import add_car except:
from _add_use_case_scripts.car_sharing.tables import add_hash pass
from _add_use_case_scripts.car_sharing.tables import add_offer try:
from _add_use_case_scripts.car_sharing.tables import add_publication from routes import use_case
from _add_use_case_scripts.car_sharing.tables import add_travel except:
from _add_use_case_scripts.car_sharing.tables import add_user pass
from _add_use_case_scripts.crowd_journalism import add_crowdjournalism_schema def test_services(self):
from _add_use_case_scripts.crowd_journalism.tables import add_classification try:
from _add_use_case_scripts.crowd_journalism.tables import add_event from services import layer_adapter_service
from _add_use_case_scripts.crowd_journalism.tables import add_purchase except:
from _add_use_case_scripts.crowd_journalism.tables import add_tag pass
from _add_use_case_scripts.crowd_journalism.tables import add_video
def test_use_case_scripts(self):
from _add_use_case_scripts.debug import add_debug_schema try:
from _add_use_case_scripts.debug.tables import add_pizza_table import network_constants as nc
from security.token_manager import TokenManager
#from _add_use_case_scripts.smart-energy import add_smart_energy_schema import requests
#from _add_use_case_scripts.smart-energy.tables import add_smart_energy from typing import List
from _add_use_case_scripts import requestPost
from _add_use_case_scripts.vialog import add_vialog_schema except:
from _add_use_case_scripts.vialog.tables import add_user pass
from _add_use_case_scripts.vialog.tables import add_video #######
if __name__ == '__main__': #from _add_use_case_scripts.bank-app import add_bank_app_schema ##eror not importing? invalid folder name?
unittest.main() #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:
pass
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:
pass
try:
from _add_use_case_scripts.debug import add_debug_schema
from _add_use_case_scripts.debug.tables import add_pizza_table
except:
pass
#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:
pass
if __name__ == '__main__':
unittest.main()
except:
pass
\ No newline at end of file
src/rest-gateway/app/test_coverage.py
View file @ 3d79e088
...@@ -2,114 +2,145 @@ import unittest ...@@ -2,114 +2,145 @@ 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:
pass
import connexion #################################
from security import swagger_util try:
from env_info import is_running_locally, get_resources_path import connexion
from flask import request from security import swagger_util
from flask import redirect from env_info import is_running_locally, get_resources_path
from flask import request
import main #something in main is causing an infinite loop (probably an async task/ listener) from flask import redirect
print("Finished test main") except: pass
try:
def test_database(self): import main #something in main is causing an infinite loop (probably an async task/ listener)
print("Entered test db") except: pass
import network_constants as netconst print("Finished test main")
from database.MongoRepositoryBase import MongoRepositoryBase
def test_database(self):
import pymongo try:
import json print("Entered test db")
from db.entities.user import User import network_constants as netconst
from typing import List from database.MongoRepositoryBase import MongoRepositoryBase
from db import repository
import pymongo
print("Finished test db") import json
from db.entities.user import User
def test_services(self): from typing import List
print("Entered test services") except: pass
from functools import wraps
from flask import g, request, redirect, url_for try:
from db import repository
# global imports (dont't worry, red is normal) except: pass
from db.repository import Repository
from db.entities.user import User print("Finished test db")
from services.user_service import UserService
from env_info import get_resources_path def test_services(self):
try:
import jwt print("Entered test services")
from datetime import datetime, timedelta from functools import wraps
from typing import Dict from flask import g, request, redirect, url_for
import bcrypt
# global imports (dont't worry, red is normal)
from services import login_wrapper from db.repository import Repository
from services import token_service from db.entities.user import User
from services import user_service from services.user_service import UserService
print("Finished test services") from env_info import get_resources_path
def test_routes(self): import jwt
print("Entered test routes") from datetime import datetime, timedelta
from flask import request, Response from typing import Dict
from messaging.ReconnectingMessageManager import ReconnectingMessageManager import bcrypt
import json except: pass
from flask import request
# global imports (dont't worry, red is normal) try:
from db.entities.user import User from services import login_wrapper
from services.user_service import UserService except: pass
from services.login_wrapper import login_required
from services.token_service import TokenService try:
import bcrypt from services import token_service
import jwt except: pass
try:
from services import user_service
except: pass
from routes import user print("Finished test services")
#from routes import blockchain_trace #message_sender in blockchain_trace is causing an infinite loop (probabily an async task//listener)
from routes import debug def test_routes(self):
print("Finished test routes") try:
print("Entered test routes")
def test_add_users(self): from flask import request, Response
print("Entered test users") from messaging.ReconnectingMessageManager import ReconnectingMessageManager
# add modules folder to interpreter path import json
import sys from flask import request
import os # global imports (dont't worry, red is normal)
import json from db.entities.user import User
import prance from services.user_service import UserService
from pathlib import Path from services.login_wrapper import login_required
from typing import Dict, Any from services.token_service import TokenService
import bcrypt
modules_path = '../../modules/' import jwt
if os.path.exists(modules_path): except: pass
sys.path.insert(1, modules_path)
try:
from services.user_service import UserService from routes import user
from env_info import is_running_locally, get_resources_path except: pass
import add_users
print("Finished test users") #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
if __name__ == '__main__': except: pass
unittest.main() 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: pass
try:
import add_users
except: pass
print("Finished test users")
if __name__ == '__main__':
unittest.main()
except:
pass
\ No newline at end of file
src/transaction-hub-in/trace-retrieval-microservice/app/test_coverage.py
View file @ 3d79e088
...@@ -3,83 +3,114 @@ import sys ...@@ -3,83 +3,114 @@ 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: pass
import connexion
from security import swagger_util #############################
from env_info import is_running_locally, get_resources_path try:
import connexion
from database.repository import Repository from security import swagger_util
from messaging.MessageHandler import MessageHandler from env_info import is_running_locally, get_resources_path
from messaging.ReconnectingMessageManager import ReconnectingMessageManager except: pass
from messaging.rest_fetcher import RestFetcher
from flask import request try:
from flask import redirect from database.repository import Repository
import main except: pass
def test_database(self): try:
# global imports (dont't worry, red is normal) from messaging.MessageHandler import MessageHandler
from typing import List, Dict except: pass
import network_constants as netconst
from database.MongoRepositoryBase import MongoRepositoryBase try:
from database.entities.transaction import Transaction from messaging.ReconnectingMessageManager import ReconnectingMessageManager
except: pass
import pymongo
import json try:
import time from messaging.rest_fetcher import RestFetcher
from flask import request
from database import repository from flask import redirect
except: pass
def test_messaging(self): try:
import main
from security.token_manager import TokenManager except: pass
import network_constants
from database.entities.transaction import Transaction def test_database(self):
from database.repository import Repository # global imports (dont't worry, red is normal)
from messaging.rest_fetcher import RestFetcher try:
from typing import List, Dict
import json import network_constants as netconst
import hashlib from database.MongoRepositoryBase import MongoRepositoryBase
import logging from database.entities.transaction import Transaction
import requests
import pymongo
from typing import Dict import json
from typing import List import time
except: pass
from messaging import MessageHandler try:
from messaging import rest_fetcher from database import repository
except: pass
def test_routes(self):
#global imports def test_messaging(self):
from database.entities.transaction import Transaction try:
from database.repository import Repository from security.token_manager import TokenManager
import network_constants
import json from database.entities.transaction import Transaction
from flask import Response, request from database.repository import Repository
from messaging.rest_fetcher import RestFetcher
from routes import transactions import json
import hashlib
import logging
import requests
if __name__ == '__main__':
unittest.main() from typing import Dict
from typing import List
except: pass
try:
from messaging import MessageHandler
except: pass
try:
from messaging import rest_fetcher
except: pass
def test_routes(self):
#global imports
try:
from database.entities.transaction import Transaction
from database.repository import Repository
except: pass
try:
import json
from flask import Response, request
except: pass
try:
from routes import transactions
except: pass
if __name__ == '__main__':
unittest.main()
except: pass
\ 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