Skip to content

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

Refactored Test Code

parent 4c4c978a
Hide whitespace changes
Inline Side-by-side
Showing with 555 additions and 573 deletions
src/data-hub/role-stage-discovery-microservice/app/test_coverage.py
View file @ f6af0a8e
......@@ -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 @ f6af0a8e
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 @ f6af0a8e
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 @ f6af0a8e
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 @ f6af0a8e
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 @ f6af0a8e
......@@ -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 @ f6af0a8e
......@@ -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 @ f6af0a8e
......@@ -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 @ f6af0a8e
......@@ -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 @ f6af0a8e
......@@ -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 @ f6af0a8e
......@@ -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 @ f6af0a8e
......@@ -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