x
/
fuel-ccp-tests
Code Issues Proposed changes

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
This commit is contained in:
Artem Panchenko 2016-09-09 10:55:15 +03:00
parent 6c2a2dc642
commit 9d89dd98f1
3 changed files with 95 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
fuel_ccp_tests/fixtures/common_fixtures.py
View File

@ -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

70
fuel_ccp_tests/helpers/log_step.py Normal file
View File

@ -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<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
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__))

86
fuel_ccp_tests/tests/system/test_netchecker.py
View File

@ -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)