Commit 612f26e0 authored by Joanne Hugé's avatar Joanne Hugé

promise/plugin: add ORS promises and tests

parent 5f477994
import time
from .util import get_json_log_latest_timestamp
from zope.interface import implementer
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
@implementer(interface.IPromise)
class RunPromise(GenericPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=1)
self.amarisoft_stats_log = self.getConfig('amarisoft-stats-log')
self.stats_period = int(self.getConfig('stats-period'))
def sense(self):
latest_timestamp = get_json_log_latest_timestamp(self.amarisoft_stats_log)
delta = time.time() - latest_timestamp
if delta > self.stats_period * 2:
self.logger.error("Latest entry from amarisoft statistics log too "\
"old (%s seconds old)" % (delta,))
else:
self.logger.info("Latest entry from amarisoft statistics is "\
"%s seconds old" % (delta,))
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=1, failure_amount=1)
from .util import get_json_log_data_interval
from .util import JSONPromise
from zope.interface import implementer
from slapos.grid.promise import interface
@implementer(interface.IPromise)
class RunPromise(JSONPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=1)
self.amarisoft_stats_log = self.getConfig('amarisoft-stats-log')
self.stats_period = int(self.getConfig('stats-period'))
self.min_rxtx_delay_threshold = float(self.getConfig('min-rxtx-delay', 0))
self.testing = self.getConfig('testing') == "True"
def sense(self):
if self.testing:
self.logger.info("skipping promise")
return
data_list = get_json_log_data_interval(self.amarisoft_stats_log, self.stats_period * 5)
min_rxtx_delay_it = map(lambda x: float(x['rf']['rxtx_delay_min']), data_list)
if not min_rxtx_delay_it:
self.logger.error("No TX/RX diff data available")
else:
min_rxtx_delay = min(min_rxtx_delay_it)
if min_rxtx_delay < self.min_rxtx_delay_threshold:
self.logger.error("The minimum available time %s (ms) for radio front end processing is lower than the threshold %s (ms)." % (min_rxtx_delay, self.min_rxtx_delay_threshold))
else:
self.logger.info("The minimum %s (ms) available time for radio front end processing is OK" % (min_rxtx_delay,))
self.json_logger.info("Minimum available time for radio front end processing (ms)",
extra={'data': {'min_rxtx_delay': min_rxtx_delay}})
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=1, failure_amount=1)
import errno
import json
import logging
import os
from dateutil import parser
from .util import JSONPromise
from .util import tail_file
from zope.interface import implementer
from slapos.grid.promise import interface
@implementer(interface.IPromise)
class RunPromise(JSONPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=1)
self.amarisoft_rf_info_log = self.getConfig('amarisoft-rf-info-log')
# self.stats_period = int(self.getConfig('stats-period'))
self.testing = self.getConfig('testing') == "True"
def sense(self):
if self.testing:
self.logger.info("skipping promise")
return
last_line = tail_file(self.amarisoft_rf_info_log)
if "CPRI" not in last_line:
self.logger.info("No CPRI feature")
else:
if "HW" in last_line and "SW" in last_line:
self.logger.info("CPRI locked")
else:
self.logger.error("HW or SW missing")
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=1, failure_amount=1)
from zope.interface import implementer
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
@implementer(interface.IPromise)
class RunPromise(GenericPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=2)
def sense(self):
ifname = self.getConfig('ifname')
testing = self.getConfig('testing') == "True"
if testing:
self.logger.info("skipping promise")
return
f = open('/sys/class/net/%s/operstate' % ifname, 'r')
if f.read() == 'up\n':
self.logger.info("%s is up", ifname)
else:
self.logger.error("%s is down", ifname)
f.close()
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=3, failure_amount=2)
import errno
import json
import logging
import os
from dateutil import parser
from .util import iter_logrotate_file_handle
from .util import iter_reverse_lines
from .util import JSONPromise
from zope.interface import implementer
from slapos.grid.promise import interface
@implementer(interface.IPromise)
class RunPromise(JSONPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=1)
self.netconf_log = self.getConfig('netconf-log')
self.testing = self.getConfig('testing') == "True"
def sense(self):
if self.testing:
self.logger.info("skipping promise")
return
for f in iter_logrotate_file_handle(self.netconf_log, 'rb'):
for line in iter_reverse_lines(f):
l = json.loads(line.decode().replace("'", '"'))
alarm_notif = l.get('data', {}).get('notification', {}).get('alarm-notif', None)
if alarm_notif and alarm_notif['fault-id'] == '102':
if alarm_notif['is-cleared'] == 'false':
affected_objects = alarm_notif.get('affected-objects', {})
self.logger.error('Loss of Frame (LOF) alarm is on, affected objects are: %s', affected_objects)
self.json_logger.info("Affected objects", extra={'data': affected_objects})
else:
self.logger.info('Loss of Frame (LOF) alarm is off')
return
self.logger.info('No Loss of Frame (LOF) alarm received')
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=1, failure_amount=1)
import errno
import json
import logging
import os
from dateutil import parser
from .util import iter_logrotate_file_handle
from .util import iter_reverse_lines
from .util import JSONPromise
from zope.interface import implementer
from slapos.grid.promise import interface
@implementer(interface.IPromise)
class RunPromise(JSONPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=1)
self.netconf_log = self.getConfig('netconf-log')
self.testing = self.getConfig('testing') == "True"
def sense(self):
if self.testing:
self.logger.info("skipping promise")
return
for f in iter_logrotate_file_handle(self.netconf_log, 'rb'):
for line in iter_reverse_lines(f):
l = json.loads(line.decode().replace("'", '"'))
alarm_notif = l.get('data', {}).get('notification', {}).get('alarm-notif', None)
if alarm_notif and alarm_notif['fault-id'] == '101':
if alarm_notif['is-cleared'] == 'false':
affected_objects = alarm_notif.get('affected-objects', {})
self.logger.error('RSSI Imbalance alarm & RX Diversity Lost alarm is on, affected objects are: %s', affected_objects)
self.json_logger.info("Affected objects", extra={'data': affected_objects})
else:
self.logger.info('RSSI Imbalance alarm & RX Diversity Lost alarm is off')
return
self.logger.info('No RSSI Imbalance alarm & RX Diversity Lost alarm received')
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=1, failure_amount=1)
import errno
import json
import logging
import os
from dateutil import parser
from .util import iter_logrotate_file_handle
from .util import iter_reverse_lines
from .util import JSONPromise
from zope.interface import implementer
from slapos.grid.promise import interface
@implementer(interface.IPromise)
class RunPromise(JSONPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=1)
self.netconf_log = self.getConfig('netconf-log')
self.testing = self.getConfig('testing') == "True"
def sense(self):
if self.testing:
self.logger.info("skipping promise")
return
for f in iter_logrotate_file_handle(self.netconf_log, 'rb'):
for line in iter_reverse_lines(f):
l = json.loads(line.decode().replace("'", '"'))
alarm_notif = l.get('data', {}).get('notification', {}).get('alarm-notif', None)
if alarm_notif and alarm_notif['fault-id'] == '9':
if alarm_notif['is-cleared'] == 'false':
affected_objects = alarm_notif.get('affected-objects', {})
self.logger.error('VSWR alarm is on, affected objects are: %s', affected_objects)
self.json_logger.info("Affected objects", extra={'data': affected_objects})
else:
self.logger.info('VSWR alarm is off')
return
self.logger.info('No VSWR alarm received')
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=1, failure_amount=1)
from .util import get_json_log_data_interval
from .util import JSONPromise
from zope.interface import implementer
from slapos.grid.promise import interface
@implementer(interface.IPromise)
class RunPromise(JSONPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=1)
self.testing = self.getConfig('testing') == "True"
self.amarisoft_stats_log = self.getConfig('amarisoft-stats-log')
self.stats_period = int(self.getConfig('stats-period'))
self.max_rx_sample_db = float(self.getConfig('max-rx-sample-db'))
def sense(self):
data_list = get_json_log_data_interval(self.amarisoft_stats_log, self.stats_period * 2)
max_rx_list = []
saturated = False
for rx_antenna_list in map(lambda x: x['samples']['rx'], data_list):
rx_list = map(lambda x: float(x['max']), rx_antenna_list)
if not max_rx_list:
max_rx_list = list(rx_list)
for i, rx in enumerate(rx_list):
max_rx_list[i] = max(max_rx_list[i], rx)
if rx >= self.max_rx_sample_db:
saturated = True
self.json_logger.info("RX maximum sample values (dB)", extra={'data': max_rx_list})
if not max_rx_list:
self.logger.error("No RX samples data available")
elif saturated:
self.logger.error("RX antenna saturated, please lower rx_gain")
else:
self.logger.info("No saturation detected on RX antenna")
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=1, failure_amount=1)
import subprocess
from zope.interface import implementer
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
@implementer(interface.IPromise)
class RunPromise(GenericPromise):
def __init__(self, config):
super(RunPromise, self).__init__(config)
self.setPeriodicity(minute=1)
def sense(self):
testing = self.getConfig('testing') == "True"
sdr = self.getConfig('sdr')
if testing:
self.logger.info("skipping promise")
return
try:
out = subprocess.check_output([
sdr + '/sdr_util', '-c', '0', 'version'], stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
if e.returncode == 1 and \
("DMA channel is already opened" in e.output.decode() or \
"Device or resource busy" in e.output.decode()):
self.logger.info("eNB is using /dev/sdr0")
return
self.logger.error("eNB is not using /dev/sdr0")
def test(self):
"""
Called after sense() if the instance is still converging.
Returns success or failure based on sense results.
In this case, fail if the previous sensor result is negative.
"""
return self._test(result_count=1, failure_amount=1)
def anomaly(self):
"""
Called after sense() if the instance has finished converging.
Returns success or failure based on sense results.
Failure signals the instance has diverged.
In this case, fail if two out of the last three results are negative.
"""
return self._anomaly(result_count=1, failure_amount=1)
# -*- coding: utf-8 -*-
##############################################################################
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import os
from datetime import datetime
from datetime import timedelta
from slapos.grid.promise import PromiseError
from slapos.promise.plugin.check_amarisoft_stats_log import RunPromise
from . import TestPromisePluginMixin
class TestCheckAmarisoftStatsLog(TestPromisePluginMixin):
promise_name = "check-amarisoft-stats-log.py"
def setUp(self):
super(TestCheckAmarisoftStatsLog, self).setUp()
self.amarisoft_stats_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'amarisoft_stats.json.log')
with open(self.amarisoft_stats_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {}}
{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {}}
{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
(datetime.now() - timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
))
def writePromise(self, **kw):
super(TestCheckAmarisoftStatsLog, self).writePromise(self.promise_name,
"from %s import %s\nextra_config_dict = %r\n"
% (RunPromise.__module__, RunPromise.__name__, kw))
def test_promise_success(self):
self.writePromise(**{
'amarisoft-stats-log': self.amarisoft_stats_log,
'stats-period': 10,
})
self.configureLauncher()
self.launcher.run()
def test_promise_fail(self):
self.writePromise(**{
'amarisoft-stats-log': self.amarisoft_stats_log,
'stats-period': 1,
})
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
if __name__ == '__main__':
unittest.main()
# -*- coding: utf-8 -*-
##############################################################################
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import os
from datetime import datetime
from datetime import timedelta
from slapos.grid.promise import PromiseError
from slapos.promise.plugin.check_baseband_latency import RunPromise
from . import TestPromisePluginMixin
class TestCheckBasebandLatency(TestPromisePluginMixin):
promise_name = "check-baseband-latency.py"
def setUp(self):
super(TestCheckBasebandLatency, self).setUp()
self.amarisoft_stats_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'amarisoft_stats.json.log')
with open(self.amarisoft_stats_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {'rf': {'rxtx_delay_min': %f}}}
{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {'rf': {'rxtx_delay_min': %f}}}
{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {'rf': {'rxtx_delay_min': %f}}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3], 7.0,
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3], 2.0,
(datetime.now() - timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3], 5.0,
))
def writePromise(self, **kw):
super(TestCheckBasebandLatency, self).writePromise(self.promise_name,
"from %s import %s\nextra_config_dict = %r\n"
% (RunPromise.__module__, RunPromise.__name__, kw))
def test_promise_success(self):
self.writePromise(**{
'amarisoft-stats-log': self.amarisoft_stats_log,
'stats-period': 100,
'min-rxtx-delay': 0,
})
self.configureLauncher()
self.launcher.run()
def test_promise_fail(self):
self.writePromise(**{
'amarisoft-stats-log': self.amarisoft_stats_log,
'stats-period': 100,
'min-rxtx-delay': 3,
})
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
if __name__ == '__main__':
unittest.main()
# -*- coding: utf-8 -*-
##############################################################################
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import mock
import os
import time
from datetime import datetime
from datetime import timedelta
from slapos.grid.promise import PromiseError
from slapos.promise.plugin.check_cpri_lock import RunPromise
from . import TestPromisePluginMixin
class TestCheckCpriLock(TestPromisePluginMixin):
promise_name = "check-cpri-lock.py"
def setUp(self):
super(TestCheckCpriLock, self).setUp()
def writePromise(self, **kw):
super(TestCheckCpriLock, self).writePromise(self.promise_name,
"from %s import %s\nextra_config_dict = %r\n"
% (RunPromise.__module__, RunPromise.__name__, kw))
def test_promise_success(self):
self.amarisoft_rf_info_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'amarisoft_rf_info.json.log')
if os.path.exists(self.amarisoft_rf_info_log):
os.remove(self.amarisoft_rf_info_log)
with open(self.amarisoft_rf_info_log, 'w+') as f:
f.write(
"""
{"time": "%s", "log_level": "INFO", "message": "RF info", "data": {'rf_info': "CPRI: x16 HW SW"}}
""" % ((datetime.now() - timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S")[:-3],))
self.writePromise(**{
'amarisoft-rf-info-log': self.amarisoft_rf_info_log,
# 'stats-period': 100,
})
self.configureLauncher()
self.launcher.run()
def test_promise_fail(self):
self.amarisoft_rf_info_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'amarisoft_rf_info.json.log')
if os.path.exists(self.amarisoft_rf_info_log):
os.remove(self.amarisoft_rf_info_log)
with open(self.amarisoft_rf_info_log, 'w') as f:
f.write(
"""
{"time": "%s", "log_level": "INFO", "message": "RF info", "data": {'rf_info': "CPRI: x16\\n"}}
""" % ((datetime.now() - timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S")[:-3],))
self.writePromise(**{
'amarisoft-rf-info-log': self.amarisoft_rf_info_log,
# 'stats-period': 100,
})
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
if __name__ == '__main__':
unittest.main()
# -*- coding: utf-8 -*-
##############################################################################
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import os
import time
from datetime import datetime
from datetime import timedelta
from slapos.grid.promise import PromiseError
from slapos.promise.plugin.check_lopcomm_lof import RunPromise
from . import TestPromisePluginMixin
class TestCheckLopcommLOFSuccess(TestPromisePluginMixin):
promise_name = "check-lopcomm-lof.py"
def setUp(self):
super(TestCheckLopcommLOFSuccess, self).setUp()
self.netconf_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'netconf.json.log')
def writePromise(self, **kw):
super(TestCheckLopcommLOFSuccess, self).writePromise(self.promise_name,
"from %s import %s\nextra_config_dict = %r\n"
% (RunPromise.__module__, RunPromise.__name__, kw))
def test_promise_success(self):
with open(self.netconf_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '102', 'fault-source': 'Radio Module', 'affected-objects': {'name': 'Radio Module'}, 'fault-severity': 'CRITICAL', 'is-cleared': 'false', 'fault-text': 'LOF Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}
{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '102', 'fault-source': 'Radio Module', 'affected-objects': {'name': 'Radio Module'}, 'fault-severity': 'CRITICAL', 'is-cleared': 'true', 'fault-text': 'LOF Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
))
self.writePromise(**{
'netconf-log': self.netconf_log,
})
self.configureLauncher()
self.launcher.run()
def test_promise_fail(self):
with open(self.netconf_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '102', 'fault-source': 'Radio Module', 'affected-objects': {'name': 'Radio Module'}, 'fault-severity': 'CRITICAL', 'is-cleared': 'true', 'fault-text': 'LOF Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}
{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '102', 'fault-source': 'Radio Module', 'affected-objects': {'name': 'Radio Module'}, 'fault-severity': 'CRITICAL', 'is-cleared': 'false', 'fault-text': 'LOF Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
))
self.writePromise(**{
'netconf-log': self.netconf_log,
})
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
self.assertEqual("Loss of Frame (LOF) alarm is on, affected objects are: {'name': 'Radio Module'}", self.getPromiseResult(self.promise_name)['result']['message'])
# -*- coding: utf-8 -*-
##############################################################################
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import os
import time
from datetime import datetime
from datetime import timedelta
from slapos.grid.promise import PromiseError
from slapos.promise.plugin.check_lopcomm_rssi import RunPromise
from . import TestPromisePluginMixin
class TestCheckLopcommRSSISuccess(TestPromisePluginMixin):
promise_name = "check-lopcomm-rssi.py"
def setUp(self):
super(TestCheckLopcommRSSISuccess, self).setUp()
self.netconf_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'netconf.json.log')
def writePromise(self, **kw):
super(TestCheckLopcommRSSISuccess, self).writePromise(self.promise_name,
"from %s import %s\nextra_config_dict = %r\n"
% (RunPromise.__module__, RunPromise.__name__, kw))
def test_promise_success(self):
with open(self.netconf_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '101', 'fault-source': 'RF Module', 'affected-objects': {'name': 'RF Module'}, 'fault-severity': 'MINOR', 'is-cleared': 'false', 'fault-text': 'RSSI Imbalance alarm & RX Diversity Lost Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}
{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '101', 'fault-source': 'RF Module', 'affected-objects': {'name': 'RF Module'}, 'fault-severity': 'MINOR', 'is-cleared': 'true', 'fault-text': 'RSSI Imbalance alarm & RX Diversity Lost Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
))
self.writePromise(**{
'netconf-log': self.netconf_log,
})
self.configureLauncher()
self.launcher.run()
def test_promise_fail(self):
with open(self.netconf_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '101', 'fault-source': 'RF Module', 'affected-objects': {'name': 'RF Module'}, 'fault-severity': 'MINOR', 'is-cleared': 'true', 'fault-text': 'RSSI Imbalance alarm & RX Diversity Lost Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}
{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '101', 'fault-source': 'RF Module', 'affected-objects': {'name': 'RF Module'}, 'fault-severity': 'MINOR', 'is-cleared': 'false', 'fault-text': 'RSSI Imbalance alarm & RX Diversity Lost Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
))
self.writePromise(**{
'netconf-log': self.netconf_log,
})
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
self.assertEqual("RSSI Imbalance alarm & RX Diversity Lost alarm is on, affected objects are: {'name': 'RF Module'}", self.getPromiseResult(self.promise_name)['result']['message'])
# -*- coding: utf-8 -*-
##############################################################################
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import os
import time
from datetime import datetime
from datetime import timedelta
from slapos.grid.promise import PromiseError
from slapos.promise.plugin.check_lopcomm_vswr import RunPromise
from . import TestPromisePluginMixin
class TestCheckLopcommVSWRSuccess(TestPromisePluginMixin):
promise_name = "check-lopcomm-vswr.py"
def setUp(self):
super(TestCheckLopcommVSWRSuccess, self).setUp()
self.netconf_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'netconf.json.log')
def writePromise(self, **kw):
super(TestCheckLopcommVSWRSuccess, self).writePromise(self.promise_name,
"from %s import %s\nextra_config_dict = %r\n"
% (RunPromise.__module__, RunPromise.__name__, kw))
def test_promise_success(self):
with open(self.netconf_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '9', 'fault-source': 'Antport1', 'affected-objects': {'name': 'Antport1'}, 'fault-severity': 'MAJOR', 'is-cleared': 'false', 'fault-text': 'PA 1 VSWR Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}
{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '9', 'fault-source': 'Antport1', 'affected-objects': {'name': 'Antport1'}, 'fault-severity': 'MAJOR', 'is-cleared': 'true', 'fault-text': 'PA 1 VSWR Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
))
self.writePromise(**{
'netconf-log': self.netconf_log,
})
self.configureLauncher()
self.launcher.run()
def test_promise_fail(self):
with open(self.netconf_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '9', 'fault-source': 'Antport1', 'affected-objects': {'name': 'Antport1'}, 'fault-severity': 'MAJOR', 'is-cleared': 'true', 'fault-text': 'PA 1 VSWR Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}
{"time": "%s", "log_level": "INFO", "message": "", "data": {'notification': {'@xmlns': 'urn:ietf:params:xml:ns:netconf:notification:1.0', 'eventTime': '1970-01-05T00:38:50Z', 'alarm-notif': {'@xmlns': 'urn:o-ran:fm:1.0', 'fault-id': '9', 'fault-source': 'Antport1', 'affected-objects': {'name': 'Antport1'}, 'fault-severity': 'MAJOR', 'is-cleared': 'false', 'fault-text': 'PA 1 VSWR Alarm', 'event-time': '1970-01-05T00:38:50Z'}}}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3],
))
self.writePromise(**{
'netconf-log': self.netconf_log,
})
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
self.assertEqual("VSWR alarm is on, affected objects are: {'name': 'Antport1'}", self.getPromiseResult(self.promise_name)['result']['message'])
# -*- coding: utf-8 -*-
##############################################################################
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import os
from datetime import datetime
from datetime import timedelta
from slapos.grid.promise import PromiseError
from slapos.promise.plugin.check_rx_saturated import RunPromise
from . import TestPromisePluginMixin
class TestCheckRXSaturated(TestPromisePluginMixin):
promise_name = "check-rx-saturated.py"
def setUp(self):
super(TestCheckRXSaturated, self).setUp()
self.amarisoft_stats_log = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'amarisoft_stats.json.log')
with open(self.amarisoft_stats_log, 'w+') as f:
f.write("""{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {'samples': {'rx': [{'max': %f}, {'max': %f}]}}}
{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {'samples': {'rx': [{'max': %f}, {'max': %f}]}}}
{"time": "%s", "log_level": "INFO", "message": "Samples stats", "data": {'samples': {'rx': [{'max': %f}, {'max': %f}]}}}""" % (
(datetime.now() - timedelta(seconds=25)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3], -5.0, -6.0,
(datetime.now() - timedelta(seconds=15)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3], -2.0, -3.0,
(datetime.now() - timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S,%f")[:-3], -9.0, -7.0,
))
def writePromise(self, **kw):
super(TestCheckRXSaturated, self).writePromise(self.promise_name,
"from %s import %s\nextra_config_dict = %r\n"
% (RunPromise.__module__, RunPromise.__name__, kw))
def test_promise_success(self):
self.writePromise(**{
'amarisoft-stats-log': self.amarisoft_stats_log,
'stats-period': 10,
'max-rx-sample-db': 0.0,
})
self.configureLauncher()
self.launcher.run()
def test_promise_fail(self):
self.writePromise(**{
'amarisoft-stats-log': self.amarisoft_stats_log,
'stats-period': 10,
'max-rx-sample-db': -3.0,
})
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
if __name__ == '__main__':
unittest.main()
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