From 9d89dd98f145c259731ff3cbe31501458207a095 Mon Sep 17 00:00:00 2001 From: Artem Panchenko Date: Fri, 9 Sep 2016 10:55:15 +0300 Subject: [PATCH] Create 'show_step' fixture to log tests steps Added new fixture 'show_step' which provides an access to the 'log_step' helper, used to log tests steps specified in the function doc string. Change-Id: I814684e4b5d8ba3e73ef6dd461d671f25f9b1c6c --- fuel_ccp_tests/fixtures/common_fixtures.py | 9 ++ fuel_ccp_tests/helpers/log_step.py | 70 +++++++++++++++ .../tests/system/test_netchecker.py | 86 ++++--------------- 3 files changed, 95 insertions(+), 70 deletions(-) create mode 100644 fuel_ccp_tests/helpers/log_step.py diff --git a/fuel_ccp_tests/fixtures/common_fixtures.py b/fuel_ccp_tests/fixtures/common_fixtures.py index 1813b83..11ea217 100644 --- a/fuel_ccp_tests/fixtures/common_fixtures.py +++ b/fuel_ccp_tests/fixtures/common_fixtures.py @@ -18,8 +18,10 @@ import time import pytest from fuel_ccp_tests import logger +from fuel_ccp_tests.helpers import log_step from fuel_ccp_tests.helpers import utils + LOG = logger.logger @@ -64,3 +66,10 @@ def pytest_runtest_teardown(item): foot = "\n" + "<" * 5 + "#" * 30 + "[ {} ]" + "#" * 30 + ">" * 5 foot = foot.format(finish_step) LOG.info(foot) + + +@pytest.fixture(scope='function') +def show_step(request): + def _show_step(step_number): + return log_step.log_step(request.function, step_number) + return _show_step diff --git a/fuel_ccp_tests/helpers/log_step.py b/fuel_ccp_tests/helpers/log_step.py new file mode 100644 index 0000000..f5d26f0 --- /dev/null +++ b/fuel_ccp_tests/helpers/log_step.py @@ -0,0 +1,70 @@ +# Copyright 2016 Mirantis, Inc. +# +# 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. + +import re + +from fuel_ccp_tests import logger + + +LOG = logger.logger + + +def parse_test_doc(docstring): + test_case = {} + parse_regex = re.compile(r'(?P^(.*\S.*\n)+)+' + r'(?P<empty_line1>\s*\n)' + r'\s*Scenario:\s*\n(?P<scenario>(.+\n)+)' + r'(?P<empty_line2>\s*(\n|$))?' + r'(\s*Duration:\s+(?P<duration>\d+).*\n)?') + doc_match = re.match(parse_regex, docstring) + + if not doc_match: + LOG.error("Can't parse test docstring, unknown format!") + return test_case + + test_case['title'] = re.sub(r'[\n\s]+', # replace multiple spaces and + ' ', # line breaks by single space + doc_match.group('title') + ).strip() + + test_case['steps'] = [] + for raw_step in re.split(r'\s+\d+\.\s*', doc_match.group('scenario')): + if not raw_step: + # start or end of the string + continue + test_case['steps'].append( + re.sub(r'[\n\s]+', # replace multiple spaces and + ' ', # line breaks by single space + raw_step + ).strip() + ) + + # TODO(apanchenko): now it works only with 'seconds' + duration = doc_match.group('duration') or 1000 + test_case['duration'] = int(duration) + return test_case + + +def log_step(func, step_num): + if not func.__doc__: + LOG.error("Can't show step #{0}: docstring for method {1} not " + "found!".format(step_num, func.__name__)) + test_case_steps = parse_test_doc(func.__doc__)['steps'] + try: + LOG.info(" *** [STEP#{0}] {1} ***".format( + step_num, + test_case_steps[step_num - 1])) + except IndexError: + LOG.error("Can't show step #{0}: docstring for method {1} does't " + "contain it!".format(step_num, func.__name__)) diff --git a/fuel_ccp_tests/tests/system/test_netchecker.py b/fuel_ccp_tests/tests/system/test_netchecker.py index 9ed0618..1198f24 100644 --- a/fuel_ccp_tests/tests/system/test_netchecker.py +++ b/fuel_ccp_tests/tests/system/test_netchecker.py @@ -17,7 +17,6 @@ import pytest import random import time import os -import re import yaml from devops.helpers.helpers import wait, wait_pass @@ -186,61 +185,10 @@ class TestFuelCCPNetChecker(SystemBaseTest, TestFuelCCPNetCheckerMixin): 'iptables -D FORWARD -p tcp --dport 8081 -j DROP', node_name=slave_node) - @staticmethod - def parse_test_doc(docstring): - test_case = {} - parse_regex = re.compile(r'(?P<title>^(.*\S.*\n)+)+' - r'(?P<empty_line1>\s*\n)' - r'\s*Scenario:\s*\n(?P<scenario>(.+\n)+)' - r'(?P<empty_line2>\s*(\n|$))?' - r'(\s*Duration:\s+(?P<duration>\d+).*\n)?') - doc_match = re.match(parse_regex, docstring) - - if not doc_match: - LOG.error("Can't parse test docstring, unknown format!") - return test_case - - test_case['title'] = re.sub(r'[\n\s]+', # replace multiple spaces and - ' ', # line breaks by single space - doc_match.group('title') - ).strip() - - test_case['steps'] = [] - for raw_step in re.split(r'\s+\d+\.\s*', doc_match.group('scenario')): - if not raw_step: - # start or end of the string - continue - test_case['steps'].append( - re.sub(r'[\n\s]+', # replace multiple spaces and - ' ', # line breaks by single space - raw_step - ).strip() - ) - - # TODO(apanchenko): now it works only with 'seconds' - duration = doc_match.group('duration') or 1000 - test_case['duration'] = int(duration) - return test_case - - @staticmethod - def show_step(func, step_num): - if not func.__doc__: - LOG.error("Can't show step #{0}: docstring for method {1} not " - "found!".format(step_num, func.__name__)) - test_case_steps = TestFuelCCPNetChecker.parse_test_doc( - func.__doc__)['steps'] - try: - LOG.info(" *** [STEP#{0}] {1} ***".format( - step_num, - test_case_steps[step_num - 1])) - except IndexError: - LOG.error("Can't show step #{0}: docstring for method {1} does't " - "contain it!".format(step_num, func.__name__)) - @pytest.mark.fail_snapshot @pytest.mark.snapshot_needed @pytest.mark.revert_snapshot(ext.SNAPSHOT.k8s_deployed) - def test_k8s_netchecker_calico(self, underlay, k8scluster): + def test_k8s_netchecker_calico(self, underlay, k8scluster, show_step): """Test for deploying an k8s environment with Calico and check connectivity between its networks @@ -267,67 +215,65 @@ class TestFuelCCPNetChecker(SystemBaseTest, TestFuelCCPNetCheckerMixin): Duration: 600 seconds """ - me = self.test_k8s_netchecker_calico - # STEP #1 - self.show_step(me, 1) + show_step(1) k8sclient = k8scluster.api # STEP #2 - self.show_step(me, 2) + show_step(2) k8scluster.create_registry() # STEP #3 - self.show_step(me, 3) + show_step(3) self.dir_upload(underlay, host='master', source=settings.NETCHECKER_SERVER_DIR, destination='/tmp/mcp-netchecker-server') # STEP #4 - self.show_step(me, 4) + show_step(4) self.build_netchecker(underlay, stype='server', source_dir='/tmp/mcp-netchecker-server') # STEP #5 - self.show_step(me, 5) + show_step(5) self.push_netchecker(underlay, stype='server') # STEP #6 - self.show_step(me, 6) + show_step(6) self.dir_upload(underlay, host='master', source=settings.NETCHECKER_AGENT_DIR, destination='/tmp/mcp-netchecker-agent') # STEP #7 - self.show_step(me, 7) + show_step(7) self.build_netchecker(underlay, stype='agent', source_dir='/tmp/mcp-netchecker-agent') # STEP #8 - self.show_step(me, 8) + show_step(8) self.push_netchecker(underlay, stype='agent') # STEP #9 - self.show_step(me, 9) + show_step(9) self.start_netchecker_server(k8sclient=k8sclient) self.wait_netchecker_running(underlay, timeout=240) # STEP #10 - self.show_step(me, 10) + show_step(10) self.start_netchecker_agent(underlay, k8sclient) # STEP #11 # currently agents need some time to start reporting to the server - self.show_step(me, 11) + show_step(11) time.sleep(120) self.check_network(underlay, k8sclient, works=True) # STEP #12 - self.show_step(me, 12) + show_step(12) target_slave = self.get_random_slave(underlay) # stop netchecker-server @@ -351,17 +297,17 @@ class TestFuelCCPNetChecker(SystemBaseTest, TestFuelCCPNetCheckerMixin): self.wait_netchecker_running(underlay, timeout=240) # STEP #13 - self.show_step(me, 13) + show_step(13) # currently agents need some time to start reporting to the server time.sleep(120) self.check_network(underlay, k8sclient, works=False) # STEP #14 - self.show_step(me, 14) + show_step(14) self.unblock_traffic_on_slave(underlay, target_slave) # STEP #15 - self.show_step(me, 15) + show_step(15) # currently agents need some time to start reporting to the server time.sleep(240) self.check_network(underlay, k8sclient, works=True)