Skip to content

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

Merge branch 'develop' into 'staging' 

Develop

See merge request !34
parents 3c486da0 d670155e
Hide whitespace changes
Inline Side-by-side
Showing with 579 additions and 584 deletions
bin/testing.py
View file @ 108a70c8
......@@ -4,7 +4,6 @@ import importlib.util
import pathlib
import shutil
import re
'''
This script searches for all 'tests/' directories and executes all tests
by cd'ing into the dir and executing unittest discover.
......@@ -12,16 +11,21 @@ It additionally installs all dependencies from a '../requirements.txt' via pip.
Use command line argument '-w' to run on windows.
'''
PY = sys.argv[2] if (len(sys.argv) > 1 and sys.argv[1] == '-py') else 'python' # use -py to use your own python command
PY = sys.argv[2] if (len(sys.argv) > 1 and sys.argv[1] == '-py') else 'python3.7' # use -py to use your own python command
ROOT = pathlib.Path(__file__).parent.parent.absolute()
REPORTS = ROOT / 'reports'
TESTS_FOLDER_NAME = os.path.normpath("/tests")
print("Creating VENV")
os.system(f"{PY} -m venv venv")
PY = f"~/smart/venv/bin/{PY}"
print("\nSearching for tests at the path: "+ str(ROOT))
count = 0
resultCodeList = []
coverage_paths_set = set()
microservice_coverage_paths_set = set()
for (dirname, dirs, files) in os.walk(ROOT):
#I assume all the tests are placed in a folder named "tests"
......@@ -37,8 +41,8 @@ for (dirname, dirs, files) in os.walk(ROOT):
exit_val = os.system(f"{PY} -m pip install -r ../requirements.txt") # install pip dependencies
#resultCodeList.append(exit_val)
#exit_val = os.system(f"{PY} -m unittest discover") # execute the tests
exit_val = os.system(f"python3.7 -m coverage run --append --omit=*/site-packages*,*/dist-packages* -m unittest discover") #TEST CODE COVERAGE
coverage_paths_set.add(os.path.normpath(dirname))
exit_val = os.system(f"{PY} -m coverage run --append --omit=*/site-packages*,*/dist-packages* -m unittest discover") #TEST CODE COVERAGE
microservice_coverage_paths_set.add(os.path.normpath(dirname))
resultCodeList.append(exit_val) #once per folder i.e if 3 tests are in a folder and crash, there will be just one exit val
except Exception as e:
......@@ -56,8 +60,8 @@ for (dirname, dirs, files) in os.walk(ROOT):
#cur_dir = os.path(dirname).parent()
os.chdir(cur_dir)
print(f"Executing coverage test in {cur_dir}")
exit_val = os.system(f"python3.7 -m coverage run --append --omit=*/site-packages* -m unittest discover")
coverage_paths_set.add(os.path.normpath(cur_dir))
exit_val = os.system(f"{PY} -m coverage run --append --omit=*/site-packages* -m unittest discover")
microservice_coverage_paths_set.add(os.path.normpath(cur_dir))
except Exception as e:
print(e)
......@@ -83,7 +87,7 @@ except Exception as e:
print("Combinging coverages")
counter = 0
for path in coverage_paths_set:
for path in microservice_coverage_paths_set:
try:
path += '/.coverage'
original = os.path.normpath( path )
......@@ -96,11 +100,20 @@ for path in coverage_paths_set:
except Exception as e:
print(e)
continue
try:
coverage_xml_path = os.path.normpath( str(REPORTS) + '/coverage.xml')
os.remove(coverage_xml_path)
#coverage_html_path = os.path.normpath( str(REPORTS) + '/htmlcov' )
#os.rmdir(coverage_html_path)
except Exception as e:
print(e)
print("Generating Combined report")
os.system("python3.7 -m coverage combine")
os.system("python3.7 -m coverage xml")
os.system("python3.7 -m coverage html") #if you want to generate the html as well
os.system(f"{PY} -m coverage combine")
os.system(f"{PY} -m coverage xml")
os.system(f"{PY} -m coverage html") #if you want to generate the html as well
firstError = -1
i = 0
......
src/data-hub/role-stage-discovery-microservice/app/test_coverage.py
View file @ 108a70c8
......@@ -3,7 +3,9 @@ import sys
for path in ['../', './']:
sys.path.insert(1, path)
#####################################
### Don't include for test report ###
#####################################
try:
class TestCoverage(unittest.TestCase):
def test_init_main(self):
......
src/data-hub/role-stage-discovery-microservice/app/tests/test_cluster.py
View file @ 108a70c8
try:
import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
# python -m unittest discover
from db.entities import Cluster
import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
from datetime import date, datetime
import json
# python -m unittest discover
from db.entities import Cluster
from datetime import date, datetime
import json
class TestCluster(unittest.TestCase):
def test_init_Cluster(self):
try:
c = Cluster('debug', 'debug-table1', 'layer1', 1, [1, 2, 3])
self.assertEqual('debug', c.use_case)
self.assertEqual('debug-table1', c.use_case_table)
self.assertEqual(1, c.cluster_label)
self.assertEqual([1, 2, 3], c.nodes)
except Exception as e:
print(e)
class TestCluster(unittest.TestCase):
def test_init_Cluster(self):
c = Cluster('debug', 'debug-table1', 'layer1', 1, [1, 2, 3])
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
self.assertEqual('debug', c.use_case)
self.assertEqual('debug-table1', c.use_case_table)
self.assertEqual(1, c.cluster_label)
self.assertEqual([1, 2, 3], c.nodes)
if __name__ == '__main__':
unittest.main()
src/data-hub/role-stage-discovery-microservice/app/tests/test_cluster_result.py
View file @ 108a70c8
try:
import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
# python -m unittest discover
from processing.clustering import ClusterResultConverter, ClusterResult
from typing import List, Dict, Any
class TestClusterResult(unittest.TestCase):
converter:ClusterResultConverter = None
def setUp(self):
try:
self.converter = ClusterResultConverter()
except Exception as e:
print (e)
def test_result_undefined_feature(self):
try:
cluster_groups = self._get_some_cluster_groups_1d()
cluster_res = self.converter.convert_to_cluster_results(
cluster_groups=cluster_groups,
features=[]
)
self.assert_correct_cluster_result_len(cluster_groups, cluster_res)
self.assert_correct_cluster_result_labels(['n.a.','n.a.','n.a.'], cluster_res)
except Exception as e:
print (e)
def test_result_1d_feature(self):
try:
cluster_groups = self._get_some_cluster_groups_1d()
cluster_res = self.converter.convert_to_cluster_results(
cluster_groups=cluster_groups,
features=['v']
)
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)
except Exception as e:
print (e)
def test_result_2d_features(self):
try:
cluster_groups = self._get_some_cluster_groups_2d()
cluster_res = self.converter.convert_to_cluster_results(
cluster_groups=cluster_groups,
features=['v', 'u']
)
self.assert_correct_cluster_result_len(cluster_groups, 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)
import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
# python -m unittest discover
from processing.clustering import ClusterResultConverter, ClusterResult
from typing import List, Dict, Any
class TestClusterResult(unittest.TestCase):
converter:ClusterResultConverter = None
def setUp(self):
self.converter = ClusterResultConverter()
def test_result_undefined_feature(self):
cluster_groups = self._get_some_cluster_groups_1d()
cluster_res = self.converter.convert_to_cluster_results(
cluster_groups=cluster_groups,
features=[]
)
self.assert_correct_cluster_result_len(cluster_groups, cluster_res)
self.assert_correct_cluster_result_labels(['n.a.','n.a.','n.a.'], cluster_res)
def test_result_1d_feature(self):
cluster_groups = self._get_some_cluster_groups_1d()
cluster_res = self.converter.convert_to_cluster_results(
cluster_groups=cluster_groups,
features=['v']
)
#region Custom Assertions
def assert_correct_cluster_result_len(self, expected: 'original dict of lists', actual: Dict[Any, ClusterResult]):
try:
self.assertEqual(len(expected), len(actual))
for i in range(len(expected)):
self.assertEqual(len(expected[i]), len(actual[i].nodes))
self.assertEqual(expected[i], actual[i].nodes)
except Exception as e:
print (e)
def assert_correct_cluster_result_labels(self, expected: List[str], actual: Dict[Any, ClusterResult]):
try:
self.assertEqual(len(expected), len(actual))
for i in range(len(expected)):
self.assertEqual(expected[i], actual[i].label)
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
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)
def test_result_2d_features(self):
cluster_groups = self._get_some_cluster_groups_2d()
cluster_res = self.converter.convert_to_cluster_results(
cluster_groups=cluster_groups,
features=['v', 'u']
)
self.assert_correct_cluster_result_len(cluster_groups, 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)
#region Custom Assertions
def assert_correct_cluster_result_len(self, expected: 'original dict of lists', actual: Dict[Any, ClusterResult]):
self.assertEqual(len(expected), len(actual))
for i in range(len(expected)):
self.assertEqual(len(expected[i]), len(actual[i].nodes))
self.assertEqual(expected[i], actual[i].nodes)
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.assertEqual(expected[i], actual[i].label)
#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()
\ No newline at end of file
src/data-hub/role-stage-discovery-microservice/app/tests/test_clusterer.py
View file @ 108a70c8
try:
import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
# python -m unittest discover
from processing.clustering import Clusterer, ClusterResult
import numpy as np
from typing import List, Dict, Any
class TestClusterer(unittest.TestCase):
clusterer:Clusterer = None
# python -m unittest discover
from processing.clustering import Clusterer, ClusterResult
import numpy as np
from typing import List, Dict, Any
def setUp(self):
self.clusterer = Clusterer(min_points=2)
class TestClusterer(unittest.TestCase):
clusterer:Clusterer = None
#region _extract_features
def setUp(self):
self.clusterer = Clusterer(min_points=2)
def test_extract_features_emptyDataset_noResults(self):
features = self.clusterer._extract_features(dataset=[], features=['test'])
np.testing.assert_equal(np.asarray([]), features)
#region _extract_features
def test_extract_features_emptyFeatures_singleEmptyResult(self):
features = self.clusterer._extract_features(dataset=[{'a':1, 'b':2}], features=[])
np.testing.assert_equal(np.asarray([[]]), features)
def test_extract_features_emptyDataset_noResults(self):
features = self.clusterer._extract_features(dataset=[], features=['test'])
np.testing.assert_equal(np.asarray([]), features)
def test_extract_features_singleFeature_Projection(self):
features = self.clusterer._extract_features(dataset=[{'a':1, 'b':2}], features=['a'])
np.testing.assert_equal(np.asarray([[1]]), features)
def test_extract_features_emptyFeatures_singleEmptyResult(self):
features = self.clusterer._extract_features(dataset=[{'a':1, 'b':2}], features=[])
np.testing.assert_equal(np.asarray([[]]), features)
def test_extract_features_singleFeature_Projection_2(self):
features = self.clusterer._extract_features(dataset=[{'a':1, 'b':2}, {'a':3, 'b':4}], features=['a'])
np.testing.assert_equal(np.asarray([[1], [3]]), features)
def test_extract_features_multFeature_Projection(self):
features = self.clusterer._extract_features(dataset=[{'a':0, 'b':2, 'c':4}, {'a':1, 'b':3, 'c':5}], features=['a','c'])
np.testing.assert_equal(np.asarray([[0,4], [1,5]]), features)
def test_extract_features_singleFeature_Projection(self):
features = self.clusterer._extract_features(dataset=[{'a':1, 'b':2}], features=['a'])
np.testing.assert_equal(np.asarray([[1]]), features)
def test_extract_features_singleFeature_Projection_2(self):
features = self.clusterer._extract_features(dataset=[{'a':1, 'b':2}, {'a':3, 'b':4}], features=['a'])
np.testing.assert_equal(np.asarray([[1], [3]]), features)
#endregion _extract_features
def test_extract_features_multFeature_Projection(self):
features = self.clusterer._extract_features(dataset=[{'a':0, 'b':2, 'c':4}, {'a':1, 'b':3, 'c':5}], features=['a','c'])
np.testing.assert_equal(np.asarray([[0,4], [1,5]]), features)
#endregion _extract_features
#region create_labels
#region create_labels
def test_create_labels_noneInput_noneOutput(self):
labels = self.clusterer.create_labels(None)
self.assertEqual(None, labels)
def test_create_labels_noneInput_noneOutput(self):
labels = self.clusterer.create_labels(None)
self.assertEqual(None, labels)
def test_create_labels_emptyInput_emptyOutput(self):
labels = self.clusterer.create_labels([])
self.assertEqual([], labels)
def test_create_labels_singleInput_noise(self):
clusterer = Clusterer(min_points=1)
def test_create_labels_emptyInput_emptyOutput(self):
labels = self.clusterer.create_labels([])
self.assertEqual([], labels)
def test_create_labels_singleInput_noise(self):
clusterer = Clusterer(min_points=1)
features = clusterer._extract_features(dataset=[self.location(1,2)], features=self.get_location_features())
labels = clusterer.create_labels(features)
features = clusterer._extract_features(dataset=[self.location(1,2)], features=self.get_location_features())
labels = clusterer.create_labels(features)
self.assertEqual(1, len(labels))
self.assertEqual(-1, labels[0])
self.assertEqual(1, len(labels))
self.assertEqual(-1, labels[0])
def test_create_labels_tooSmallInputForMinPtsHyperparameter_error(self):
clusterer = Clusterer(min_points=3)
def test_create_labels_tooSmallInputForMinPtsHyperparameter_error(self):
clusterer = Clusterer(min_points=3)
features = clusterer._extract_features(dataset=[self.location(1,2), self.location(1,2)], features=self.get_location_features())
with self.assertRaises(ValueError):
# Fails because (min_pts > |input elements|)
clusterer.create_labels(features)
features = clusterer._extract_features(dataset=[self.location(1,2), self.location(1,2)], features=self.get_location_features())
with self.assertRaises(ValueError):
# Fails because (min_pts > |input elements|)
clusterer.create_labels(features)
def test_create_labels_nearInputs_singleCluster(self):
locations = [self.location(1,2), self.location(2,2)]
def test_create_labels_nearInputs_singleCluster(self):
locations = [self.location(1,2), self.location(2,2)]
features = self.clusterer._extract_features(dataset=locations, features=self.get_location_features())
labels = self.clusterer.create_labels(features)
features = self.clusterer._extract_features(dataset=locations, features=self.get_location_features())
labels = self.clusterer.create_labels(features)
self.assertEqual(2, len(labels))
self.assertEqual(labels[0], labels[1])
self.assertEqual(2, len(labels))
self.assertEqual(labels[0], labels[1])
def test_create_labels_nearInputs_twoClusters(self):
locations = [self.location(1,2), self.location(2,2), self.location(20,20), self.location(20,23)]
def test_create_labels_nearInputs_twoClusters(self):
locations = [self.location(1,2), self.location(2,2), self.location(20,20), self.location(20,23)]
features = self.clusterer._extract_features(dataset=locations, features=self.get_location_features())
labels = self.clusterer.create_labels(features)
features = self.clusterer._extract_features(dataset=locations, features=self.get_location_features())
labels = self.clusterer.create_labels(features)
self.assertEqual(4, len(labels))
self.assertEqual(labels[0], labels[1])
self.assertEqual(labels[2], labels[3])
self.assertNotEqual(labels[0], labels[2])
self.assertEqual(4, len(labels))
self.assertEqual(labels[0], labels[1])
self.assertEqual(labels[2], labels[3])
self.assertNotEqual(labels[0], labels[2])
#endregion create_labels
#endregion create_labels
#region label_dataset
#region label_dataset
def test_label_dataset_NoneLocations_NoException(self):
self.clusterer.label_dataset(None, [])
def test_label_dataset_NoneLocations_NoException(self):
self.clusterer.label_dataset(None, [])
def test_label_dataset_NoneLabels_NoException(self):
self.clusterer.label_dataset([], None)
def test_label_dataset_NoneLabels_NoException(self):
self.clusterer.label_dataset([], None)
def test_label_dataset_emptyInput_emptyOutput(self):
locations = []
self.clusterer.label_dataset(locations, [])
self.assertEqual(0, len(locations))
def test_label_dataset_emptyInput_emptyOutput(self):
locations = []
self.clusterer.label_dataset(locations, [])
self.assertEqual(0, len(locations))
def test_label_dataset_diffInputLengths_ValueError_1(self):
with self.assertRaises(ValueError):
self.clusterer.label_dataset([], [1])
def test_label_dataset_diffInputLengths_ValueError_1(self):
with self.assertRaises(ValueError):
self.clusterer.label_dataset([], [1])
def test_label_dataset_diffInputLengths_ValueError_2(self):
with self.assertRaises(ValueError):
self.clusterer.label_dataset([self.location(1,2)], [])
def test_label_dataset_diffInputLengths_ValueError_2(self):
with self.assertRaises(ValueError):
self.clusterer.label_dataset([self.location(1,2)], [])
def test_label_dataset_multInput_correctlyLabeled(self):
locations = [self.location(1,2), self.location(2,2), self.location(20,20)]
labels = [17,2,20]
def test_label_dataset_multInput_correctlyLabeled(self):
locations = [self.location(1,2), self.location(2,2), self.location(20,20)]
labels = [17,2,20]
self.clusterer.label_dataset(locations, labels)
self.clusterer.label_dataset(locations, labels)
self.assertEqual(3, len(locations))
self.assertHaveLabelsAsNewKey(locations, labels)
self.assertEqual(3, len(locations))
self.assertHaveLabelsAsNewKey(locations, labels)
#endregion label_dataset
#endregion label_dataset
#region cluster_dataset
#region cluster_dataset
def test_cluster_dataset_locationsMultInput_correctlyLabeled(self):
locations = [self.location(1,2), self.location(2,2), self.location(20,20), self.location(20,21)]
labels = [0,0,1,1]
exp_res = {0:locations[0:2], 1:locations[2:4]}
def test_cluster_dataset_locationsMultInput_correctlyLabeled(self):
locations = [self.location(1,2), self.location(2,2), self.location(20,20), self.location(20,21)]
labels = [0,0,1,1]
exp_res = {0:locations[0:2], 1:locations[2:4]}
res = self.clusterer.cluster_dataset(locations, self.get_location_features())
res = self.clusterer.cluster_dataset(locations, self.get_location_features())
self.assertHaveLabelsAsNewKey(locations, labels)
self.assertClusteringResult(exp_res, res)
self.assertHaveLabelsAsNewKey(locations, labels)
self.assertClusteringResult(exp_res, res)
def test_cluster_dataset_timesMultInput_correctlyLabeled(self):
times = [self.time(123), self.time(128), self.time(223), self.time(225)]
labels = [0,0,1,1]
exp_res = {0:times[0:2], 1:times[2:4]}
def test_cluster_dataset_timesMultInput_correctlyLabeled(self):
times = [self.time(123), self.time(128), self.time(223), self.time(225)]
labels = [0,0,1,1]
exp_res = {0:times[0:2], 1:times[2:4]}
res = self.clusterer.cluster_dataset(times, self.get_time_features())
self.assertHaveLabelsAsNewKey(times, labels)
self.assertClusteringResult(exp_res, res)
res = self.clusterer.cluster_dataset(times, self.get_time_features())
self.assertHaveLabelsAsNewKey(times, labels)
self.assertClusteringResult(exp_res, res)
def test_cluster_dataset_locationsMultInput_correctlyLabeled_2(self):
clusterer = Clusterer(3)
locations = [
self.location(1,2), self.location(2,2), self.location(2,2),
self.location(20,20), self.location(20,21), self.location(20,20),
self.location(50,50),
self.location(50,1), self.location(50,2), self.location(50,-1)
]
labels = [0,0,0,1,1,1,-1,2,2,2]
exp_res = {0:locations[0:3], 1:locations[3:6], -1:locations[6:7], 2:locations[7:10]}
def test_cluster_dataset_locationsMultInput_correctlyLabeled_2(self):
clusterer = Clusterer(3)
locations = [
self.location(1,2), self.location(2,2), self.location(2,2),
self.location(20,20), self.location(20,21), self.location(20,20),
self.location(50,50),
self.location(50,1), self.location(50,2), self.location(50,-1)
]
labels = [0,0,0,1,1,1,-1,2,2,2]
exp_res = {0:locations[0:3], 1:locations[3:6], -1:locations[6:7], 2:locations[7:10]}
res = clusterer.cluster_dataset(locations, self.get_location_features())
res = clusterer.cluster_dataset(locations, self.get_location_features())
self.assertHaveLabelsAsNewKey(locations, labels)
self.assertClusteringResult(exp_res, res)
self.assertHaveLabelsAsNewKey(locations, labels)
self.assertClusteringResult(exp_res, res)
#endregion cluster_dataset
#endregion cluster_dataset
#region helper methods
#region helper methods
def location(self, lat, long_) -> dict:
return {'latitude': lat, 'longitude':long_}
def location(self, lat, long_) -> dict:
return {'latitude': lat, 'longitude':long_}
def get_location_features(self):
return ['latitude', 'longitude']
def get_location_features(self):
return ['latitude', 'longitude']
def time(self, ts) -> dict:
return {'timestamp': ts}
def time(self, ts) -> dict:
return {'timestamp': ts}
def get_time_features(self):
return ['timestamp']
def get_time_features(self):
return ['timestamp']
def assertHaveLabelsAsNewKey(self, locations, labels):
self.assertEqual(len(labels), len(locations))
def assertHaveLabelsAsNewKey(self, locations, labels):
self.assertEqual(len(labels), len(locations))
for i in range(len(locations)):
self.assertEqual(labels[i], locations[i]['cluster_label'])
for i in range(len(locations)):
self.assertEqual(labels[i], locations[i]['cluster_label'])
def assertClusteringResult(self, expected: Dict[Any, List], actual: Dict[Any, ClusterResult]):
self.assertEqual(len(expected), len(actual))
def assertClusteringResult(self, expected: Dict[Any, List], actual: Dict[Any, ClusterResult]):
self.assertEqual(len(expected), len(actual))
for k in expected.keys():
if k not in actual:
self.fail(f"Cluster key ({k}, {type(k)}) not in result.")
self.assertListEqual(expected[k], actual[k].nodes)
#endregion helper methods
for k in expected.keys():
if k not in actual:
self.fail(f"Cluster key ({k}, {type(k)}) not in result.")
self.assertListEqual(expected[k], actual[k].nodes)
#endregion helper methods
if __name__ == '__main__':
unittest.main()
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
src/data-hub/role-stage-discovery-microservice/app/tests/test_similarity.py
View file @ 108a70c8
try:
import unittest
import unittest
import sys
for path in ['../', './', '../../../modules/', '../../../../modules']:
sys.path.insert(1, path)
import sys
for path in ['../', './', '../../../modules/', '../../../../modules']:
sys.path.insert(1, path)
from db.entities.connected_node import NodeC
from db.entities.connected_cluster import ClusterC
from db.entities.connected_layer import LayerC
from db.entities.connected_node import NodeC
from db.entities.connected_cluster import ClusterC
from db.entities.connected_layer import LayerC
import math
import datetime
from typing import Dict
from typing import Dict
from processing.similarityFiles.dataInput import *
from processing.similarityFiles.calculateWeights import *
from processing.similarityFiles.calculateSimilarity import *
from processing.similarityFiles.miscFunctions import *
import math
import datetime
from typing import Dict
from typing import Dict
from processing.similarityFiles.dataInput import *
from processing.similarityFiles.calculateWeights import *
from processing.similarityFiles.calculateSimilarity import *
from processing.similarityFiles.miscFunctions import *
import json
import json
class TestSimilarity(unittest.TestCase):
'''Tests the similarity calculation which works without object orientation.'''
class TestSimilarity(unittest.TestCase):
'''Tests the similarity calculation which works without object orientation.'''
def test_integration_similarityCalculation(self):
'''
Only for testing, can be deleted at any time.\n
Served as a testing example to make sure the computations are correct
'''
def test_integration_calculateSimilarity_ClustersDict_CorrectValue(self):
'''
Only for testing, can be deleted at any time.\n
Served as a testing example to make sure the computations are correct
'''
limitNrNodes = 100000
limitNrNodes = 100000
layerDict = dict()
("Creating Connected_cluster dict and similarity dict")
inputLayerLocation=getTestLocationLayerData()
inputLayerTime=getTestTimeLayerData()
inputLayerPrice=getTestPriceLayerData()
layerDict = populateWithNewNodesSingleLayer(inputLayerLocation,layerDict,limitNrNodes)
layerDict = populateWithNewNodesSingleLayer(inputLayerTime,layerDict,limitNrNodes)
layerDict = populateWithNewNodesSingleLayer(inputLayerPrice,layerDict,limitNrNodes)
layerDict = dict()
("Creating Connected_cluster dict and similarity dict")
inputLayerLocation=getTestLocationLayerData()
inputLayerTime=getTestTimeLayerData()
inputLayerPrice=getTestPriceLayerData()
layerDict = populateWithNewNodesSingleLayer(inputLayerLocation,layerDict,limitNrNodes)
layerDict = populateWithNewNodesSingleLayer(inputLayerTime,layerDict,limitNrNodes)
layerDict = populateWithNewNodesSingleLayer(inputLayerPrice,layerDict,limitNrNodes)
layerDict = calculateWeights(layerDict)
similarityDict = calculateSimilarity(layerDict)
("Asserting if the calculated values are true")
layerDict = calculateWeights(layerDict)
similarityDict = calculateSimilarity(layerDict)
("Asserting if the calculated values are true")
#assert x == "hello"
#assert x == "hello"
# SYNTAX:
#similarityDict[(clusterLabel1,clusterLabel2,layerOfTheClusters)][layerToWhichTheClustersAreCompared] == ExpectedSimilarityValue
#checking if the cluster "1" and cluster "2" from the "Location" layer have the sqrt(2) similarity values for when compared with 'Price'and 'Time'layers
self.assertEqual(similarityDict[(1,2,'Location')]['Price'], math.sqrt(2))
self.assertEqual(similarityDict[(1,2,'Location')]['Time'], math.sqrt(2))
self.assertEqual(similarityDict[(1,3,'Location')]['Price'], math.sqrt(10))
self.assertEqual(similarityDict[(1,3,'Location')]['Time'], math.sqrt(16))
self.assertEqual(similarityDict[(2,3,'Location')]['Price'], math.sqrt(4))
self.assertEqual(similarityDict[(2,3,'Location')]['Time'], math.sqrt(10))
self.assertEqual(similarityDict[(4,5,'Time')]['Location'], math.sqrt(19))
self.assertEqual(similarityDict[(4,5,'Time')]['Price'], math.sqrt(26))
self.assertEqual(similarityDict[(6,7,'Price')]['Location'], math.sqrt(3))
self.assertEqual(similarityDict[(6,7,'Price')]['Time'], math.sqrt(8))
# assert similarityDict[(1,2,'Location')]
("Test Passed Succesfully")
return layerDict
# SYNTAX:
#similarityDict[(clusterLabel1,clusterLabel2,layerOfTheClusters)][layerToWhichTheClustersAreCompared] == ExpectedSimilarityValue
#checking if the cluster "1" and cluster "2" from the "Location" layer have the sqrt(2) similarity values for when compared with 'Price'and 'Time'layers
self.assertEqual(similarityDict[(1,2,'Location')]['Price'], math.sqrt(2))
self.assertEqual(similarityDict[(1,2,'Location')]['Time'], math.sqrt(2))
self.assertEqual(similarityDict[(1,3,'Location')]['Price'], math.sqrt(10))
self.assertEqual(similarityDict[(1,3,'Location')]['Time'], math.sqrt(16))
self.assertEqual(similarityDict[(2,3,'Location')]['Price'], math.sqrt(4))
self.assertEqual(similarityDict[(2,3,'Location')]['Time'], math.sqrt(10))
self.assertEqual(similarityDict[(4,5,'Time')]['Location'], math.sqrt(19))
self.assertEqual(similarityDict[(4,5,'Time')]['Price'], math.sqrt(26))
self.assertEqual(similarityDict[(6,7,'Price')]['Location'], math.sqrt(3))
self.assertEqual(similarityDict[(6,7,'Price')]['Time'], math.sqrt(8))
# assert similarityDict[(1,2,'Location')]
("Test Passed Succesfully")
return layerDict
def getTestLocationLayerData():
inputLayerLocation = [
{
"cluster_label": 1,
"layer_name": "Location",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "a",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "aa",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "aaa",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aaaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
},
{
"cluster_label": 2,
"layer_name": "Location",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156b",
"TravelPrice": 15,
"UniqueID": "b",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "bb",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "bbb",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "bb",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
},
{
"cluster_label": 3,
"layer_name": "Location",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156c",
"TravelPrice": 15,
"UniqueID": "c",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "c",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "cc",
"UserID": "asdf"
}
]
}
]
return inputLayerLocation
def getTestTimeLayerData():
inputLayerTime = [
def getTestLocationLayerData():
inputLayerLocation = [
{
"cluster_label": 4,
"layer_name": "Time",
"cluster_label": 1,
"layer_name": "Location",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "a",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
......@@ -227,37 +95,19 @@ try:
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "aaaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "b",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
"UniqueID": "aa",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "bb",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
"UniqueID": "aaa",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
......@@ -265,50 +115,20 @@ try:
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "bbb",
"UniqueID": "aaaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
},
{
"cluster_label": 5,
"layer_name": "Time",
"cluster_label": 2,
"layer_name": "Location",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156f",
"TravelPrice": 15,
"UniqueID": "a",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "c",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
}
]
return inputLayerTime
def getTestPriceLayerData():
inputLayerPrice = [
{
"cluster_label": 6,
"layer_name": "Price",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156h",
"TravelID": "5e57ec9159bc0668543f156b",
"TravelPrice": 15,
"UniqueID": "b",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
......@@ -317,27 +137,19 @@ try:
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "bb",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "aaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
"UniqueID": "bbb",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
......@@ -345,61 +157,245 @@ try:
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aaaa",
"UniqueID": "bb",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
},
{
"cluster_label": 7,
"layer_name": "Price",
"cluster_label": 3,
"layer_name": "Location",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156g",
"TravelPrice": 15,
"UniqueID": "a",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelID": "5e57ec9159bc0668543f156c",
"TravelPrice": 15,
"UniqueID": "aa",
"UniqueID": "c",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "b",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
"UniqueID": "c",
"UserID": "asdf"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelID": "asd",
"TravelPrice": 15,
"UniqueID": "c",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
"UniqueID": "cc",
"UserID": "asdf"
}
]
}
]
return inputLayerPrice
return inputLayerLocation
def getTestTimeLayerData():
inputLayerTime = [
{
"cluster_label": 4,
"layer_name": "Time",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "a",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aaaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "b",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "bb",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "bbb",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
},
{
"cluster_label": 5,
"layer_name": "Time",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156f",
"TravelPrice": 15,
"UniqueID": "a",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "c",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
}
]
return inputLayerTime
def getTestPriceLayerData():
inputLayerPrice = [
{
"cluster_label": 6,
"layer_name": "Price",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156h",
"TravelPrice": 15,
"UniqueID": "b",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "bb",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aaaa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
},
{
"cluster_label": 7,
"layer_name": "Price",
"nodes": [
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156g",
"TravelPrice": 15,
"UniqueID": "a",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "aa",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "b",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
},
{
"Finished_time": 1576631193265951,
"Latitude_Destination": -5.973257,
"Longitude_Destination": 37.416316,
"TravelID": "5e57ec9159bc0668543f156d",
"TravelPrice": 15,
"UniqueID": "c",
"UserID": "a95075f5042b1b27060080156d87fe34ec7e712c"
}
]
}
]
return inputLayerPrice
if __name__ == '__main__':
unittest.main()
except Exception as e:
print ("Exception found:")
print (e)
\ No newline at end of file
if __name__ == '__main__':
unittest.main()
\ No newline at end of file
src/data-hub/semantic-linking-microservice/app/test_coverage.py
View file @ 108a70c8
......@@ -3,7 +3,9 @@ import sys
for path in ['../', './']:
sys.path.insert(1, path)
#####################################
### Don't include for test report ###
#####################################
try:
class TestCoverage(unittest.TestCase):
def test_init_main(self):
......
src/data-hub/semantic-linking-microservice/app/tests/test_pipeline.py
View file @ 108a70c8
......@@ -67,8 +67,8 @@ class Test_Pipeline(unittest.TestCase):
}
}
}
def testTraceProcessing(self):
#original name testTraceProcessing
def test_handle_new_trace_newTraceMsg_correctlyInserted(self):
msg = self._buildTraceMessage()
self.handler.handle_new_trace(msg["content"])
self.assertEqual(len(self.handler._repository.layernodes),1)
......
src/participation-hub/business-logic-microservice/app/test_coverage.py
View file @ 108a70c8
......@@ -3,7 +3,9 @@ import sys
for path in ['../', './']:
sys.path.insert(1, path)
#####################################
### Don't include for test report ###
#####################################
try:
class TestCoverage(unittest.TestCase):
def test_init_main(self):
......
src/participation-hub/business-logic-microservice/app/tests/test_layer_adapter.py
View file @ 108a70c8
......@@ -4,7 +4,7 @@ from db.entities.layer_adapter import LayerAdapter
class Test_Layer_Adapter(unittest.TestCase):
def test_valid_adapter(self):
def test_LayerAdapter_newLayerAdapterObj_validInstantiation(self):
adapter1 = LayerAdapter("layer1", "use_case", "table", ["a", "c"], ["a"])
print(adapter1.to_serializable_dict)
......
src/rest-gateway/app/test_coverage.py
View file @ 108a70c8
......@@ -2,6 +2,11 @@ import unittest
import sys
for path in ['../', './']:
sys.path.insert(1, path)
#####################################
### Don't include for test report ###
#####################################
try:
class TestCoverage(unittest.TestCase):
def test_init_main(self):
......
src/transaction-hub-in/trace-retrieval-microservice/app/test_coverage.py
View file @ 108a70c8
......@@ -3,6 +3,10 @@ import sys
for path in ['../', './']:
sys.path.insert(1, path)
#####################################
### Don't include for test report ###
#####################################
try:
class TestCoverage(unittest.TestCase):
def test_init_main(self):
......
src/transaction-hub-in/trace-retrieval-microservice/app/tests/test_MessageHandler.py
View file @ 108a70c8
......@@ -170,7 +170,7 @@ class Test_MessageHandler(unittest.TestCase):
self.assertEqual('semantic-linking', self.msg_sender.last_message['key'])
self.assertEqual('new-trace', json.loads(self.msg_sender.last_message['msg'])["type"])
def test_handleblockchain_duplicateTrace(self):
def test_handleBlockchainTransaction_duplicateTrace_oneTransAddedToDuplicateRepo(self):
msg = self._get_valid_message()
msg2 = self._get_valid_message()
msg = eval(msg)
......@@ -179,7 +179,7 @@ class Test_MessageHandler(unittest.TestCase):
self.handler.handle_blockchain_transaction(msg2['content'])
self.assertEqual(len(self.repo.added_transactions),len(self.repo.duplicated_transactions))
def test_handleblockchain_duplicateTraceDifferentTable(self):
def test_handleBlockchainTransaction_duplicateTraceDifferentTable_bothTransactionsAddedAsUnique(self):
msg = self._get_valid_message()
msg2 = self._get_valid_message2()
msg = eval(msg)
......@@ -188,7 +188,7 @@ class Test_MessageHandler(unittest.TestCase):
self.handler.handle_blockchain_transaction(msg2['content'])
self.assertEqual(len(self.repo.added_transactions),2)
def test_handleblockchain_duplicateTraceDifferentUseCase(self):
def test_handleBlockchainTransaction_duplicateTraceDifferentUseCase_bothTransactionsAddedAsUnique(self):
msg = self._get_valid_message()
msg2 = self._get_valid_message3()
msg = eval(msg)
......
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