From e8941ef6cbdf4b8aba7bd66c472a7acb0aa0a4a1 Mon Sep 17 00:00:00 2001 From: Tomas Babej Date: Thu, 26 Sep 2013 13:18:59 +0200 Subject: ipatests: Create util module for ipatests Part of: https://fedorahosted.org/freeipa/ticket/3834 --- ipatests/test_integration/util.py | 60 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 ipatests/test_integration/util.py (limited to 'ipatests/test_integration') diff --git a/ipatests/test_integration/util.py b/ipatests/test_integration/util.py new file mode 100644 index 00000000..1a1bb3fc --- /dev/null +++ b/ipatests/test_integration/util.py @@ -0,0 +1,60 @@ +# Authors: +# Tomas Babej +# +# Copyright (C) 2013 Red Hat +# see file 'COPYING' for use and warranty information +# +# 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 3 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, see . + +import time + + +def run_repeatedly(host, command, assert_zero_rc=True, test=None, + timeout=30, **kwargs): + """ + Runs command on host repeatedly until it's finished successfully (returns + 0 exit code and its stdout passes the test function). + + Returns True if the command was executed succesfully, False otherwise. + + This method accepts additional kwargs and passes these arguments + to the actual run_command method. + """ + + time_waited = 0 + time_step = 2 + + # Check that the test is a function + if test: + assert callable(test) + + while(time_waited <= timeout): + result = host.run_command(command, raiseonerr=False, **kwargs) + + return_code_ok = not assert_zero_rc or (result.returncode == 0) + test_ok = not test or test(result.stdout_text) + + if return_code_ok and test_ok: + # Command successful + return True + else: + # Command not successful + time.sleep(time_step) + time_waited += time_step + + raise AssertionError("Command: {cmd} repeatedly failed {times} times, " + "exceeding the timeout of {timeout} seconds." + .format(cmd=' '.join(command), + times=timeout / time_step, + timeout=timeout)) -- cgit