From 535e0dc1340ce57c0bfeed8d1ce530111cf41063 Mon Sep 17 00:00:00 2001 From: Alex Shatov Date: Thu, 22 Mar 2018 15:52:14 -0400 Subject: 2.4.1 better step-timer + audit + unit-tests - improved step-timer due to unit tests -- fixed events -- better logging - audit - collect list of package thru subprocess pip freeze - unit tests coverage 76% Change-Id: Ib1cb5f687612ecf18aa7414b1ff7dbf5774345b4 Signed-off-by: Alex Shatov Issue-ID: DCAEGEN2-389 --- tests/test_step_timer.py | 206 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 206 insertions(+) create mode 100644 tests/test_step_timer.py (limited to 'tests/test_step_timer.py') diff --git a/tests/test_step_timer.py b/tests/test_step_timer.py new file mode 100644 index 0000000..fe2d1c1 --- /dev/null +++ b/tests/test_step_timer.py @@ -0,0 +1,206 @@ +# ============LICENSE_START======================================================= +# Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved. +# ================================================================================ +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ============LICENSE_END========================================================= +# +# ECOMP is a trademark and service mark of AT&T Intellectual Property. + +"""test of the step_timer""" + +import json +import logging +import sys +import time +from datetime import datetime + +from policyhandler.config import Config +from policyhandler.step_timer import StepTimer + +Config.load_from_file() + +class MockTimer(object): + """testing step_timer""" + logger = logging.getLogger("policy_handler.unit_test.step_timer") + + INIT = "init" + NEXT = "next" + STARTED = "started" + PAUSED = "paused" + STOPPING = "stopping" + STOPPED = "stopped" + + def __init__(self, name, interval): + """step_timer test settings""" + self.name = name or "step_timer" + self.interval = interval or 5 + self.step_timer = None + self.status = None + self.run_counter = 0 + self.status_ts = datetime.now() + self.exe_ts = None + self.exe_interval = None + self.set_status(MockTimer.INIT) + + def __enter__(self): + """constructor""" + return self + + def __exit__(self, exc_type, exc_value, traceback): + """destructor""" + self.stop_timer() + + def on_time(self, *args, **kwargs): + """timer event""" + self.exe_ts = datetime.now() + self.exe_interval = (self.exe_ts - self.status_ts).total_seconds() + MockTimer.logger.info("run on_time[%s] (%s, %s) in %s for %s", + self.run_counter, json.dumps(args), json.dumps(kwargs), + self.exe_interval, self.get_status()) + time.sleep(3) + MockTimer.logger.info("done on_time[%s] (%s, %s) in %s for %s", + self.run_counter, json.dumps(args), json.dumps(kwargs), + self.exe_interval, self.get_status()) + + def verify_last_event(self): + """assertions needs to be in the main thread""" + if self.exe_interval is None: + MockTimer.logger.info("not executed: %s", self.get_status()) + return + + MockTimer.logger.info("verify exe %s for %s", self.exe_interval, self.get_status()) + assert self.exe_interval >= self.interval + assert self.exe_interval < 2 * self.interval + MockTimer.logger.info("success %s", self.get_status()) + + def run_timer(self): + """create and start the step_timer""" + if self.step_timer: + self.step_timer.next() + self.set_status(MockTimer.NEXT) + return + + self.step_timer = StepTimer( + self.name, self.interval, MockTimer.on_time, + MockTimer.logger, + self + ) + self.step_timer.start() + self.set_status(MockTimer.STARTED) + + def pause_timer(self): + """pause step_timer""" + if self.step_timer: + self.step_timer.pause() + self.set_status(MockTimer.PAUSED) + + def stop_timer(self): + """stop step_timer""" + if self.step_timer: + self.set_status(MockTimer.STOPPING) + self.step_timer.stop() + self.step_timer.join() + self.step_timer = None + self.set_status(MockTimer.STOPPED) + + def set_status(self, status): + """set the status of the timer""" + if status in [MockTimer.NEXT, MockTimer.STARTED]: + self.run_counter += 1 + + self.status = status + now = datetime.now() + time_step = (now - self.status_ts).total_seconds() + self.status_ts = now + MockTimer.logger.info("%s: %s", time_step, self.get_status()) + + def get_status(self): + """string representation""" + status = "{0}[{1}] {2} in {3} from {4} last exe {5}".format( + self.status, self.run_counter, self.name, self.interval, + str(self.status_ts), str(self.exe_ts) + ) + if self.step_timer: + return "{0}: {1}".format(status, self.step_timer.get_status()) + return status + +def test_step_timer(): + """test step_timer""" + MockTimer.logger.info("============ test_step_timer =========") + with MockTimer("step_timer", 5) as step_timer: + step_timer.run_timer() + time.sleep(1) + step_timer.verify_last_event() + + time.sleep(1 + step_timer.interval) + step_timer.verify_last_event() + + step_timer.pause_timer() + time.sleep(2) + step_timer.verify_last_event() + + step_timer.pause_timer() + time.sleep(2) + step_timer.verify_last_event() + + step_timer.run_timer() + time.sleep(3 * step_timer.interval) + step_timer.verify_last_event() + + step_timer.run_timer() + time.sleep(3 * step_timer.interval) + step_timer.verify_last_event() + +def test_interrupt_step_timer(): + """test step_timer""" + MockTimer.logger.info("============ test_interrupt_step_timer =========") + with MockTimer("step_timer", 5) as step_timer: + step_timer.run_timer() + time.sleep(1) + step_timer.verify_last_event() + + step_timer.pause_timer() + time.sleep(2 + step_timer.interval) + step_timer.verify_last_event() + + step_timer.pause_timer() + time.sleep(2) + step_timer.verify_last_event() + + step_timer.pause_timer() + time.sleep(2) + step_timer.verify_last_event() + + step_timer.run_timer() + time.sleep(2) + step_timer.verify_last_event() + + step_timer.run_timer() + time.sleep(2 + step_timer.interval) + step_timer.verify_last_event() + + step_timer.run_timer() + time.sleep(2) + step_timer.verify_last_event() + + step_timer.pause_timer() + time.sleep(2) + step_timer.verify_last_event() + + step_timer.run_timer() + time.sleep(2) + step_timer.verify_last_event() + + step_timer.run_timer() + time.sleep(3 * step_timer.interval) + step_timer.verify_last_event() -- cgit 1.2.3-korg