diff --git a/LCU/Maintenance/MDB_tools/cli/probe_mdb.py b/LCU/Maintenance/MDB_tools/cli/probe_mdb.py
index 3c165f2bf0ba56c48dcbfdee6428a08a353c3c0f..0ea1975e89cefe4deddeaca53a1c96acac2f5b95 100755
--- a/LCU/Maintenance/MDB_tools/cli/probe_mdb.py
+++ b/LCU/Maintenance/MDB_tools/cli/probe_mdb.py
@@ -21,6 +21,11 @@ CACHING_SIZE = 100
def setup_argument_parser():
+ """
+ Setup the argument parser with the possible options
+ :return: an instance of the argument parser with all the probe_mdb options
+ :rtype: argparse.ArgumentParser
+ """
parser = argparse.ArgumentParser(prog='probe_mdb', description="probe_mdb is a tool which simplify the queries to" +
" the maintenance database. Specifically, it helps"
" to gather past test output performed on a "
@@ -43,7 +48,7 @@ def setup_argument_parser():
return parser
-def perform_query(query_string, address, next_url=None):
+def query_api_for_stationtest_rtsm_results(query_string, address, next_url=None):
"""
Execute the query string to the specified address
:param query_string: the query string to be done
@@ -64,14 +69,14 @@ def perform_query(query_string, address, next_url=None):
extracted_content = parsed_content['results']
next_url = parsed_content['next']
if next_url is not None:
- return extracted_content + perform_query(query_string, address, next_url)
+ return extracted_content + query_api_for_stationtest_rtsm_results(query_string, address, next_url)
else:
return extracted_content
else:
return []
-def get_query_string_for_time_limit(from_date, to_date):
+def get_url_string_to_select_timespan(from_date, to_date):
"""
Create the query string to query the data in a certain time stamp
@@ -112,11 +117,13 @@ def get_query_string_for_time_limit(from_date, to_date):
return query
-def get_query_string_for_type(test_type):
+def get_url_string_to_select_test_type(test_type):
"""
Create the query string to probe a certain station test type choices are RTSM, STATION_TEST
:param test_type: it can be RTSM or STATION_TEST
- :return:
+ :type test_type: str
+ :return: the url for the API call
+ :rtype: str
"""
if test_type == 'RTSM':
@@ -130,11 +137,12 @@ def get_query_string_for_type(test_type):
return query
-def get_query_string_for_station_name(station_name):
+def get_url_string_to_select_station_name(station_name):
"""
Return the query string to filter the results to a specific station name
- :param station_name:
- :return:
+ :param station_name: the name of the station
+ :return: the part of the url to the API call to select a specific station
+ :rtype: str
"""
query = '&station_name={}'.format(station_name.strip(' \'\"'))
return query
@@ -149,38 +157,56 @@ class TestType:
RTSM = 'RTSM'
-def query_station_test(address, from_time, to_time, station_name=""):
+def retrieve_station_tests(address, from_time, to_time, station_name=""):
"""
Queries the station test for a given time span and a station name
+ :param address: address where the MDB REST api is hosted
+ :type address: str
+ :param from_time: date time from which interrogate the MDB REST API
+ :type from_time: str or datetime
+ :param to_time: date time to which interrogate the MDB REST API
+ :type to_time: str or datetime
+ :param station_name: name of the station to query for
+ :type station_name: str
+ :return: a dict with the results of the query to the REST API
"""
- query_string = get_query_string_for_type(TestType.STATION_TEST)
+ query_string = get_url_string_to_select_test_type(TestType.STATION_TEST)
if from_time or to_time:
- query_string += get_query_string_for_time_limit(from_time, to_time)
+ query_string += get_url_string_to_select_timespan(from_time, to_time)
if station_name and station_name is not None:
- query_string += get_query_string_for_station_name(station_name)
+ query_string += get_url_string_to_select_station_name(station_name)
- return perform_query(query_string, address)
+ return query_api_for_stationtest_rtsm_results(query_string, address)
-def query_rtsm(address, from_time, to_time, station_name=""):
- """
- Queries the station test for a given time span and a station name
+def retrieve_rtsm(address, from_time, to_time, station_name=""):
"""
- query_string = get_query_string_for_type(TestType.RTSM)
+ Queries the RTSM for a given time span and a station name
+ :param address: address where the MDB REST api is hosted
+ :type address: str
+ :param from_time: date time from which interrogate the MDB REST API
+ :type from_time: str or datetime
+ :param to_time: date time to which interrogate the MDB REST API
+ :type to_time: str or datetime
+ :param station_name: name of the station to query for
+ :type station_name: str
+ :return: a dict with the results of the query to the REST API
+ """
+ query_string = get_url_string_to_select_test_type(TestType.RTSM)
if from_time or to_time:
- query_string += get_query_string_for_time_limit(from_time, to_time)
+ query_string += get_url_string_to_select_timespan(from_time, to_time)
if station_name:
- query_string += get_query_string_for_station_name(station_name)
+ query_string += get_url_string_to_select_station_name(station_name)
- return perform_query(query_string, address)
+ return query_api_for_stationtest_rtsm_results(query_string, address)
-def reformat_values(item):
+def reformat_datetime(item):
"""
Reformats some objects to better print them out
- :param item:
- :return:
+ :param item: the object to be formatted
+ :return: the formatted string
"""
if isinstance(item, datetime):
return str(item).replace('T', ' ').replace('-', ' ')
@@ -193,11 +219,10 @@ def print_out_station_test_summary(station_test, idx):
Prints the summary of the station test
:param station_test: dict containing the results of the station test
:param idx: index of the current station test
- :return:
"""
summary_fields = ['station_name', 'station_type', 'start_time', 'end_time', 'performed_checks']
summary_header = ['Name', 'Type', 'Start', 'End', 'Checks']
- values = [reformat_values(station_test[item]) for item in summary_fields]
+ values = [reformat_datetime(station_test[item]) for item in summary_fields]
print()
print('Station test #{}'.format(idx))
table_width = 200
@@ -219,13 +244,17 @@ def sort_component_errors(component_errors):
component_type ascending
component_id ascending
:param component_errors:
- :return:
+ :return: returns the sorted component errors
"""
sorted_components = sorted(component_errors, key=itemgetter('component_type', 'component_id'))
return sorted_components
def print_out_component_errors_summary(station_test):
+ """
+ Prints out on terminal a summary of the component errors
+ :param station_test: the dict that contains the station_test
+ """
summary_fields = ['component_type', 'component_id', 'error_type']
table = beautifultable.BeautifulTable()
@@ -239,6 +268,10 @@ def print_out_component_errors_summary(station_test):
def print_out_component_errors(station_test):
+ """
+ Prints out the output of the component errors
+ :param station_test: the dict that contains the station_test
+ """
excluded_fields = {'parent', 'resourcetype', 'polymorphic_ctype', 'id', 'station_test_id', 'tile_errors',
'parent_component_error'}
summary_fields = ['component_type', 'component_id', 'error_type']
@@ -291,10 +324,11 @@ def rtsm_sort_errors(errors):
"""
Sort the rtsm errors by
rcu
+ error_type
time
:param errors: RTSM errors to be sorted
- :return:
+ :return: the sorted RTSM errors
"""
sorted_components = sorted(errors, key=itemgetter('rcu', 'error_type', 'time'))
return sorted_components
@@ -304,7 +338,6 @@ def print_out_rtsm_summary(rtsm_results):
"""
It prints out on a terminal the RTMS summary
:param rtsm_results: A list of dicts containing the RTSM results
- :return:
"""
table_width = 200
terminal = blessings.Terminal()
@@ -348,7 +381,6 @@ def print_out_rtsm(rtsm_results):
"""
Prints the results of the RTSM
:param rtsm_results:
- :return:
"""
table_width = 200
terminal = blessings.Terminal()
@@ -379,12 +411,11 @@ def print_out_rtsm(rtsm_results):
print(table)
-def print_out(station_test_results, rtsm_results):
+def print_out_rtsm_and_stationtests(station_test_results, rtsm_results):
"""
Prints out the formatted station test results and RTSM results
- :param station_test_results:
- :param rtsm_results:
- :return:
+ :param station_test_results: dict that contains the stationtests output
+ :param rtsm_results: dict that contains the rtsm tests output
"""
for i, station_test_result in enumerate(station_test_results):
@@ -399,6 +430,9 @@ def print_out(station_test_results, rtsm_results):
def probe_mdb():
+ """
+ Main entry for the script
+ """
parser = setup_argument_parser()
args = parser.parse_args()
if args is None:
@@ -412,11 +446,11 @@ def probe_mdb():
if args.last_month:
from_date = datetime.now()-timedelta(days=30)
- station_test_results = query_station_test(args.address, from_date, to_date, args.station)
+ station_test_results = retrieve_station_tests(args.address, from_date, to_date, args.station)
- rtsm_results = query_rtsm(args.address, from_date, to_date, args.station)
+ rtsm_results = retrieve_rtsm(args.address, from_date, to_date, args.station)
- print_out(station_test_results, rtsm_results)
+ print_out_rtsm_and_stationtests(station_test_results, rtsm_results)
def main():
diff --git a/LCU/Maintenance/MDB_tools/test/t_probe_mdb.py b/LCU/Maintenance/MDB_tools/test/t_probe_mdb.py
index 174e6e95330a22856b62bf33710dc7c4a8dc6423..423b3441ad3a94da21c1980741ba6ac2d2973d85 100644
--- a/LCU/Maintenance/MDB_tools/test/t_probe_mdb.py
+++ b/LCU/Maintenance/MDB_tools/test/t_probe_mdb.py
@@ -3,17 +3,18 @@ import unittest
from datetime import datetime
from mock import patch, Mock
+from output_testing import compare_output_with
-from lofar.maintenance.utils.cli.probe_mdb import get_query_string_for_time_limit, TestType, \
- get_query_string_for_type, CACHING_SIZE, get_query_string_for_station_name, query_station_test, query_rtsm, \
- reformat_values, sort_component_errors, rtsm_sort_errors, perform_query
+from lofar.maintenance.utils.cli.probe_mdb import get_url_string_to_select_timespan, TestType, \
+ get_url_string_to_select_test_type, CACHING_SIZE, get_url_string_to_select_station_name, retrieve_station_tests, retrieve_rtsm, \
+ reformat_datetime, sort_component_errors, rtsm_sort_errors, query_api_for_stationtest_rtsm_results
logger = logging.getLogger(__name__)
class TESTProbeMDB(unittest.TestCase):
@patch('lofar.maintenance.utils.cli.probe_mdb.requests')
- def test_perform_query_success(self, requests_mock):
+ def test_query_api_for_stationtest_rtsm_results_success(self, requests_mock):
address = "testaddress"
query_string = "test_query_string"
@@ -22,7 +23,7 @@ class TESTProbeMDB(unittest.TestCase):
results=[dict(key='value')],
next=None)
requests_mock.get.return_value = response
- result = perform_query(query_string, address)
+ result = query_api_for_stationtest_rtsm_results(query_string, address)
requests_mock.get.assert_called()
requests_mock.get.assert_called_with('testaddress/test_query_string')
@@ -30,7 +31,7 @@ class TESTProbeMDB(unittest.TestCase):
self.assertEqual(result, [dict(key='value')])
@patch('lofar.maintenance.utils.cli.probe_mdb.requests')
- def test_perform_query_success_next_url(self, requests_mock):
+ def test_query_api_for_stationtest_rtsm_success_next_url(self, requests_mock):
address = "testaddress"
query_string = "test_query_string"
next_url = "next_url"
@@ -41,7 +42,7 @@ class TESTProbeMDB(unittest.TestCase):
next=None)]
requests_mock.get.return_value = response
- response = perform_query(query_string, address)
+ response = query_api_for_stationtest_rtsm_results(query_string, address)
requests_mock.get.assert_called()
requests_mock.get.assert_called_with('next_url')
@@ -49,220 +50,220 @@ class TESTProbeMDB(unittest.TestCase):
self.assertEqual(response, [dict(key='value'), dict(key='value')])
@patch('lofar.maintenance.utils.cli.probe_mdb.requests')
- def test_perform_query_error_returns_empty_list(self, requests_mock):
+ def test_query_api_for_stationtest_rtsm_error_returns_empty_list(self, requests_mock):
address = "testaddress"
query_string = "test_query_string"
response = Mock(status_code=300)
requests_mock.get.return_value = response
- response = perform_query(query_string, address)
+ response = query_api_for_stationtest_rtsm_results(query_string, address)
requests_mock.get.assert_called()
self.assertEqual(response, [])
- def test_get_query_string_from_time_limit_from_date_success(self):
+ def test_get_url_string_to_select_timespan_from_date_success(self):
from_date = "2018-12-13"
- response = get_query_string_for_time_limit(from_date, None)
+ response = get_url_string_to_select_timespan(from_date, None)
self.assertEqual(response, '&start_time__year__gte=2018&start_time__month__gte=12&start_time__day__gte=13')
from_date = datetime(2018, 12, 13)
- response = get_query_string_for_time_limit(from_date, None)
+ response = get_url_string_to_select_timespan(from_date, None)
self.assertEqual(response, '&start_time__year__gte=2018&start_time__month__gte=12&start_time__day__gte=13')
- def test_get_query_string_from_time_limit_wrong_string_format(self):
+ def test_get_url_string_to_select_timespan_wrong_string_format(self):
from_date = "2018-12-13-21"
with self.assertRaises(ValueError):
- get_query_string_for_time_limit(from_date, None)
+ get_url_string_to_select_timespan(from_date, None)
- def test_get_query_string_from_time_limit_to_date_success(self):
+ def test_get_url_string_to_select_timespan_to_date_success(self):
to_date = "2018-12-13"
- response = get_query_string_for_time_limit(None, to_date)
+ response = get_url_string_to_select_timespan(None, to_date)
self.assertEqual(response, '&start_time__year__lte=2018&start_time__month__lte=12&start_time__day__lte=13')
to_date = datetime(2018, 12, 13)
- response = get_query_string_for_time_limit(None, to_date)
+ response = get_url_string_to_select_timespan(None, to_date)
self.assertEqual(response, '&start_time__year__lte=2018&start_time__month__lte=12&start_time__day__lte=13')
- def test_get_query_string_to_time_limit_wrong_string_format(self):
+ def test_get_url_string_to_select_timespan_wrong_string_format(self):
to_date = "2018-12-13-21"
with self.assertRaises(ValueError):
- get_query_string_for_time_limit(None, to_date)
+ get_url_string_to_select_timespan(None, to_date)
def test_get_query_string_none_specified(self):
- self.assertEqual(get_query_string_for_time_limit(None, None), '')
+ self.assertEqual(get_url_string_to_select_timespan(None, None), '')
def test_get_query_string_both_specified(self):
date = "2018-12-13"
expected_result = '&start_time__year__gte=2018&start_time__month__gte=12&start_time__day__gte=13' + \
'&start_time__year__lte=2018&start_time__month__lte=12&start_time__day__lte=13'
- self.assertEqual(get_query_string_for_time_limit(date, date), expected_result)
+ self.assertEqual(get_url_string_to_select_timespan(date, date), expected_result)
def test_get_query_string_per_type_rtsm(self):
- result = get_query_string_for_type(TestType.RTSM)
+ result = get_url_string_to_select_test_type(TestType.RTSM)
self.assertEqual(result, 'rtsm/?page_size={}'.format(CACHING_SIZE))
def test_get_query_string_per_type_station_test(self):
- result = get_query_string_for_type(TestType.STATION_TEST)
+ result = get_url_string_to_select_test_type(TestType.STATION_TEST)
self.assertEqual(result, 'stationtests/?page_size={}'.format(CACHING_SIZE))
def test_get_query_string_per_type_wrong_selection(self):
with self.assertRaises(ValueError):
- get_query_string_for_type("WRONG")
+ get_url_string_to_select_test_type("WRONG")
- def test_get_query_string_for_station_name(self):
+ def test_get_url_string_to_select_station_name(self):
station_name = 'JIMMY'
- result = get_query_string_for_station_name(station_name)
+ result = get_url_string_to_select_station_name(station_name)
self.assertEqual('&station_name={}'.format(station_name), result)
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_type')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_time_limit')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_station_name')
- @patch('lofar.maintenance.utils.cli.probe_mdb.perform_query')
- def test_query_station_test_all_set(self, perform_query_mock,
- get_query_string_for_station_name_mock,
- get_query_string_for_time_limit_mock,
- get_query_string_for_type_mock):
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_test_type')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_timespan')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_station_name')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.query_api_for_stationtest_rtsm_results')
+ def test_retrieve_station_tests_all_set(self, query_api_for_stationtest_rtsm_results_mock,
+ get_url_string_to_select_station_name_mock,
+ get_url_string_to_select_timespan_mock,
+ get_url_string_to_select_test_type_mock):
from_time = "2013-12-12"
to_time = "2013-12-11"
address = "test_address"
station_name = "WILLY"
- query_station_test(address, from_time, to_time, station_name=station_name)
-
- get_query_string_for_station_name_mock.assert_called()
- get_query_string_for_time_limit_mock.assert_called()
- get_query_string_for_type_mock.assert_called()
- perform_query_mock.assert_called()
-
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_type')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_time_limit')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_station_name')
- @patch('lofar.maintenance.utils.cli.probe_mdb.perform_query')
- def test_query_station_test_time_not_set(self, perform_query_mock,
- get_query_string_for_station_name_mock,
- get_query_string_for_time_limit_mock,
- get_query_string_for_type_mock):
+ retrieve_station_tests(address, from_time, to_time, station_name=station_name)
+
+ get_url_string_to_select_station_name_mock.assert_called()
+ get_url_string_to_select_timespan_mock.assert_called()
+ get_url_string_to_select_test_type_mock.assert_called()
+ query_api_for_stationtest_rtsm_results_mock.assert_called()
+
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_test_type')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_timespan')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_station_name')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.query_api_for_stationtest_rtsm_results')
+ def test_retrieve_station_tests_time_not_set(self, query_api_for_stationtest_rtsm_results_mock,
+ get_url_string_to_select_station_name_mock,
+ get_url_string_to_select_timespan_mock,
+ get_url_string_to_select_test_type_mock):
from_time = None
to_time = None
address = "test_address"
station_name = "WILLY"
- query_station_test(address, from_time, to_time, station_name=station_name)
-
- get_query_string_for_station_name_mock.assert_called()
- get_query_string_for_time_limit_mock.assert_not_called()
- get_query_string_for_type_mock.assert_called()
- perform_query_mock.assert_called()
-
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_type')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_time_limit')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_station_name')
- @patch('lofar.maintenance.utils.cli.probe_mdb.perform_query')
- def test_query_station_test_station_name_not_set(self, perform_query_mock,
- get_query_string_for_station_name_mock,
- get_query_string_for_time_limit_mock,
- get_query_string_for_type_mock):
+ retrieve_station_tests(address, from_time, to_time, station_name=station_name)
+
+ get_url_string_to_select_station_name_mock.assert_called()
+ get_url_string_to_select_timespan_mock.assert_not_called()
+ get_url_string_to_select_test_type_mock.assert_called()
+ query_api_for_stationtest_rtsm_results_mock.assert_called()
+
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_test_type')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_timespan')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_station_name')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.query_api_for_stationtest_rtsm_results')
+ def test_retrieve_station_tests_station_name_not_set(self, query_api_for_stationtest_rtsm_results_mock,
+ get_url_string_to_select_station_name_mock,
+ get_url_string_to_select_timespan_mock,
+ get_url_string_to_select_test_type_mock):
from_time = "2013-05-01"
to_time = "2013-05-01"
address = "test_address"
- query_station_test(address, from_time, to_time)
+ retrieve_station_tests(address, from_time, to_time)
- get_query_string_for_station_name_mock.assert_not_called()
- get_query_string_for_time_limit_mock.assert_called()
- get_query_string_for_type_mock.assert_called()
- perform_query_mock.assert_called()
+ get_url_string_to_select_station_name_mock.assert_not_called()
+ get_url_string_to_select_timespan_mock.assert_called()
+ get_url_string_to_select_test_type_mock.assert_called()
+ query_api_for_stationtest_rtsm_results_mock.assert_called()
- get_query_string_for_station_name_mock.reset()
- get_query_string_for_time_limit_mock.reset()
- get_query_string_for_type_mock.reset()
- perform_query_mock.reset()
+ get_url_string_to_select_station_name_mock.reset()
+ get_url_string_to_select_timespan_mock.reset()
+ get_url_string_to_select_test_type_mock.reset()
+ query_api_for_stationtest_rtsm_results_mock.reset()
from_time = "2013-05-01"
to_time = "2013-05-01"
address = "test_address"
- query_station_test(address, from_time, to_time, station_name="")
-
- get_query_string_for_station_name_mock.assert_not_called()
- get_query_string_for_time_limit_mock.assert_called()
- get_query_string_for_type_mock.assert_called()
- perform_query_mock.assert_called()
-
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_type')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_time_limit')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_station_name')
- @patch('lofar.maintenance.utils.cli.probe_mdb.perform_query')
- def test_query_rtsm_all_set(self, perform_query_mock,
- get_query_string_for_station_name_mock,
- get_query_string_for_time_limit_mock,
- get_query_string_for_type_mock):
+ retrieve_station_tests(address, from_time, to_time, station_name="")
+
+ get_url_string_to_select_station_name_mock.assert_not_called()
+ get_url_string_to_select_timespan_mock.assert_called()
+ get_url_string_to_select_test_type_mock.assert_called()
+ query_api_for_stationtest_rtsm_results_mock.assert_called()
+
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_test_type')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_timespan')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_station_name')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.query_api_for_stationtest_rtsm_results')
+ def test_retrieve_rtsm_all_set(self, query_api_for_stationtest_rtsm_results_mock,
+ get_url_string_to_select_station_name_mock,
+ get_url_string_to_select_timespan_mock,
+ get_url_string_to_select_test_type_mock):
from_time = "2013-12-12"
to_time = "2013-12-11"
address = "test_address"
station_name = "WILLY"
- query_rtsm(address, from_time, to_time, station_name=station_name)
-
- get_query_string_for_station_name_mock.assert_called()
- get_query_string_for_time_limit_mock.assert_called()
- get_query_string_for_type_mock.assert_called()
- perform_query_mock.assert_called()
-
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_type')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_time_limit')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_station_name')
- @patch('lofar.maintenance.utils.cli.probe_mdb.perform_query')
- def test_query_rtsm_time_not_set(self, perform_query_mock,
- get_query_string_for_station_name_mock,
- get_query_string_for_time_limit_mock,
- get_query_string_for_type_mock):
+ retrieve_rtsm(address, from_time, to_time, station_name=station_name)
+
+ get_url_string_to_select_station_name_mock.assert_called()
+ get_url_string_to_select_timespan_mock.assert_called()
+ get_url_string_to_select_test_type_mock.assert_called()
+ query_api_for_stationtest_rtsm_results_mock.assert_called()
+
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_test_type')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_timespan')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_station_name')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.query_api_for_stationtest_rtsm_results')
+ def test_retrieve_rtsm_time_not_set(self, query_api_for_stationtest_rtsm_results_mock,
+ get_url_string_to_select_station_name_mock,
+ get_url_string_to_select_timespan_mock,
+ get_url_string_to_select_test_type_mock):
from_time = None
to_time = None
address = "test_address"
station_name = "WILLY"
- query_rtsm(address, from_time, to_time, station_name=station_name)
-
- get_query_string_for_station_name_mock.assert_called()
- get_query_string_for_time_limit_mock.assert_not_called()
- get_query_string_for_type_mock.assert_called()
- perform_query_mock.assert_called()
-
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_type')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_time_limit')
- @patch('lofar.maintenance.utils.cli.probe_mdb.get_query_string_for_station_name')
- @patch('lofar.maintenance.utils.cli.probe_mdb.perform_query')
- def test_query_rtsm_station_name_not_set(self, perform_query_mock,
- get_query_string_for_station_name_mock,
- get_query_string_for_time_limit_mock,
- get_query_string_for_type_mock):
+ retrieve_rtsm(address, from_time, to_time, station_name=station_name)
+
+ get_url_string_to_select_station_name_mock.assert_called()
+ get_url_string_to_select_timespan_mock.assert_not_called()
+ get_url_string_to_select_test_type_mock.assert_called()
+ query_api_for_stationtest_rtsm_results_mock.assert_called()
+
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_test_type')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_timespan')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.get_url_string_to_select_station_name')
+ @patch('lofar.maintenance.utils.cli.probe_mdb.query_api_for_stationtest_rtsm_results')
+ def test_retrieve_rtsm_station_name_not_set(self, query_api_for_stationtest_rtsm_results_mock,
+ get_url_string_to_select_station_name_mock,
+ get_url_string_to_select_timespan_mock,
+ get_url_string_to_select_test_type_mock):
from_time = "2013-05-01"
to_time = "2013-05-01"
address = "test_address"
- query_rtsm(address, from_time, to_time)
+ retrieve_rtsm(address, from_time, to_time)
- get_query_string_for_station_name_mock.assert_not_called()
- get_query_string_for_time_limit_mock.assert_called()
- get_query_string_for_type_mock.assert_called()
- perform_query_mock.assert_called()
+ get_url_string_to_select_station_name_mock.assert_not_called()
+ get_url_string_to_select_timespan_mock.assert_called()
+ get_url_string_to_select_test_type_mock.assert_called()
+ query_api_for_stationtest_rtsm_results_mock.assert_called()
- get_query_string_for_station_name_mock.reset()
- get_query_string_for_time_limit_mock.reset()
- get_query_string_for_type_mock.reset()
- perform_query_mock.reset()
+ get_url_string_to_select_station_name_mock.reset()
+ get_url_string_to_select_timespan_mock.reset()
+ get_url_string_to_select_test_type_mock.reset()
+ query_api_for_stationtest_rtsm_results_mock.reset()
from_time = "2013-05-01"
to_time = "2013-05-01"
address = "test_address"
- query_rtsm(address, from_time, to_time, station_name="")
+ retrieve_rtsm(address, from_time, to_time, station_name="")
- get_query_string_for_station_name_mock.assert_not_called()
- get_query_string_for_time_limit_mock.assert_called()
- get_query_string_for_type_mock.assert_called()
- perform_query_mock.assert_called()
+ get_url_string_to_select_station_name_mock.assert_not_called()
+ get_url_string_to_select_timespan_mock.assert_called()
+ get_url_string_to_select_test_type_mock.assert_called()
+ query_api_for_stationtest_rtsm_results_mock.assert_called()
def test_reformat_values(self):
date = datetime(2013, 12, 12, 11, 11, 11)
- self.assertEqual(reformat_values(date), "2013 12 12 11:11:11")
+ self.assertEqual(reformat_datetime(date), "2013 12 12 11:11:11")
date = '20/07/2012'
- self.assertEqual(reformat_values(date), date)
+ self.assertEqual(reformat_datetime(date), date)
def test_sort_component_errors(self):
component_errors = [dict(component_type="HBA", component_id=13),
@@ -300,7 +301,6 @@ class TESTProbeMDB(unittest.TestCase):
sorted_component_errors = rtsm_sort_errors(rtsm_component_errors)
self.assertEqual(sorted_component_errors, expected_result)
-
if __name__ == '__main__':
logging.basicConfig(format="%(asctime)s %(levelname)s: %(message)s", level=logging.DEBUG)
unittest.main()