Skip to content
Snippets Groups Projects
Commit 154000e6 authored by Nikolay Stanchev's avatar Nikolay Stanchev
Browse files

Adds tests to check if the execute_pipeline endpoint opens a separate process in the correct way

parent d08e1359
No related branches found
No related tags found
No related merge requests found
......@@ -24,7 +24,7 @@
from json import dumps
import pytest
from unittest.mock import patch
from unittest.mock import patch, Mock
from pyramid import testing
from clmcservice.graphapi.views import GraphAPI
from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound
......@@ -495,13 +495,21 @@ class TestGraphAPI(object):
error_raised = True
assert error_raised, error_msg
@patch('clmcservice.graphapi.views.Popen')
@patch('clmcservice.graphapi.views.uuid4')
def test_execute_pipeline_graph(self, uuid_mock):
def test_execute_pipeline_graph(self, uuid_mock, popen_mock):
"""
Tests the functionality to start a pipeline script executing the graph API workflow - build, query, delete
"""
# mock the behaviour of the uuid function
uuid_mock.return_value = "monitor_test_uuid1"
# mock the behaviour of the Popen class
popen_intance_mock = Mock()
popen_intance_mock.pid = Mock(return_value=111)
popen_mock.return_value = popen_intance_mock
service_functions = dict(nginx={"measurement_name": "nginx", "response_time_field": "mean(avg_processing_time)",
"request_size_field": "mean(avg_request_size)", "response_size_field": "mean(avg_response_size)"},
minio={"measurement_name": "minio_http", "response_time_field": "mean(total_processing_time)/mean(total_requests_count)",
......@@ -515,7 +523,8 @@ class TestGraphAPI(object):
response = GraphAPI(request).execute_graph_pipeline()
assert response == {"uuid": uuid_mock.return_value, "database": "test_sfc"}
assert False, "Implement tests for checking that an actual process is started."
popen_mock.assert_called_with("graph_pipeline.sh", body) # assert that the graph pipeline script is ran with the JSON config that was received in the request
popen_intance_mock.pid.assert_called() # assert that the process ID attribute was called and saved
@staticmethod
def check_exist_relationship(relationships_tuple, graph, uuid):
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment