Commit 4cd488de authored by Manuel's avatar Manuel

fixed testcases to run with the testing.py script

parent 383b7388
......@@ -11,6 +11,7 @@ Use command line argument '-w' to run on windows.
'''
PY = 'py' if (len(sys.argv) > 1 and sys.argv[1] == '-w') else 'python3.7' # use -w to run on windows
ROOT = pathlib.Path(__file__).parent.parent.absolute()
TESTS_FOLDER_NAME = os.path.normpath("/tests")
......
......@@ -21,7 +21,8 @@ from messaging.ReconnectingMessageManager import ReconnectingMessageManager
from messaging.MessageHandler import MessageHandler
# init message handler
message_handler = MessageHandler()
from db.repository import Repository
message_handler = MessageHandler(Repository())
def message_received_callback(channel, method, properties, body):
message_handler.handle_generic(body)
......
import network_constants as netconst
from security.token_manager import TokenManager
from db.entities import Layer
from db.repository import Repository
import json
import requests
......@@ -12,8 +11,8 @@ import logging
LOGGER = logging.getLogger(__name__)
class MessageHandler:
def __init__(self):
self._repository = Repository()
def __init__(self, repository):
self._repository = repository
def handle_generic(self, body):
LOGGER.info(f"Received message: {body}")
......
......@@ -2,14 +2,39 @@ import unittest
import manage_sys_paths
import json
from db.repository import Repository
from db.entities.layer import Layer
from messaging.MessageHandler import MessageHandler
from messaging.DummyMessageManager import DummyMessageManager as DummyMessageSender
from typing import List
class DummyMongoRepo:
'''Dummy class to be used for testing the MessageHandler'''
last_trace = None
def insert_trace(self, trace):
self.last_trace = trace
def delete_layer(self, layer):
pass
def add_layer(self, layer):
pass
def get_layers_for_use_case_and_table(self, use_case, table):
return [
Layer(
layer_info={
"layer_name": "pizza_layer_test",
"properties": ["UniqueID"],
"total_properties": ["UniqueID"],
"use_case": "debug",
"table": "pizza"
}
)
]
def add_layer_nodes(self, nodes: List):
pass
class Test_Pipeline(unittest.TestCase):
handler = None
......@@ -19,7 +44,7 @@ class Test_Pipeline(unittest.TestCase):
def setUp(self):
self.repo = DummyMongoRepo()
self.msg_sender = DummyMessageSender.get_instance()
self.handler = MessageHandler()
self.handler = MessageHandler(self.repo)
def _buildTraceMessage(self):
return {
......
import unittest
# import manage_sys_paths
import manage_sys_paths
from db.entities.layer_adapter import LayerAdapter
......
[
{
"type": "blockchain-transaction",
"ApplicationType": "debug",
"docType": "order",
"orderId": 1,
"pizza": 1,
"price": 12.0,
"restaurant": 1
},
{
"type": "blockchain-transaction",
"ApplicationType": "debug",
"docType": "pizza",
"id": 1,
"name": "Diavolo",
"dough": {
"type": "wheat",
"spinach": false
},
"toppings": [
{
"name": "Tomato Sauce",
"price": 1.00
},
{
"name": "Cheese",
"price": 0.50
},
{
"name": "Chilli Oil",
"price": 0.50
},
{
"name": "Peppers",
"price": 1.50
}
]
}
]
import unittest
import manage_sys_paths
from messaging.DummyMessageManager import DummyMessageManager
# init dummy message manager so no connection to rabbitmq is established
_ = DummyMessageManager.get_instance()
import routes.blockchain_trace as blockchain_trace
from env_info import get_resources_path
from security.token_manager import TokenManager
import network_constants
import json
import requests
class Test_BlockchainTrace(unittest.TestCase):
def test_isBlockchainTraceValid_validInputAndTypes(self):
with open(f"{get_resources_path()}/pizza_transactions.json") as f:
transactions = json.loads(f.read())
jwt = TokenManager.getInstance().getToken()
for transaction in transactions:
url = f'https://{network_constants.REST_GATEWAY_HOSTNAME}:{network_constants.REST_GATEWAY_REST_PORT}/api/trace'
response = requests.post(
url,
verify = False,
proxies = { "http":None, "https":None },
headers = {"Authorization": f"Bearer {jwt}"},
json = transaction
)
print(f"{url} {response.status_code}")
if response.status_code != 201:
print(f"server responded with {response.status_code}: {response.text}")
self.assertTrue(response.status_code==201, "Should accept the new trace.")
if __name__ == '__main__':
unittest.main()
\ No newline at end of file
......@@ -76,11 +76,6 @@ class Test_MessageHandler(unittest.TestCase):
}
}
return json.dumps(message_values)
def test_HandlePizzaMessage(self):
msg = self._get_pizza_message()
_ = self.handler.handle_blockchain_transaction(json.loads(msg)["content"])
if __name__ == '__main__':
unittest.main()
\ 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