github/open-cas-linux
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Enable logging mechanism and example of logger usage in one test

Browse Source
This commit is contained in:
Katarzyna Lapinska 2019-10-23 11:03:38 +02:00
parent 06b3e69c0f
commit 35d52caf74
5 changed files with 116 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
test/functional/.gitignore vendored
View File

@ -1 +1,2 @@
plugins/
results/

9
test/functional/pytest.ini
View File

@ -1,9 +1,2 @@
[pytest]
log_cli = 1
log_cli_level = INFO
log_cli_format = %(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)
log_cli_date_format=%Y-%m-%d %H:%M:%S
log_file = pytest.log
log_file_level = INFO
log_file_format = %(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)
log_file_date_format=%Y-%m-%d %H:%M:%S
addopts = -s

2
test/functional/test-framework

@ -1 +1 @@
Subproject commit 465f9670c0ae15f09aa2b8a775cb0b47a34a51fe
Subproject commit 5667b7bbf9bb60eedefb20b0625d1b9f32f214f2

49
test/functional/tests/cli/test_cli_start_stop.py
View File

@ -4,39 +4,47 @@
#
import logging
import pytest
from api.cas import casadm, casadm_parser
from tests.conftest import base_prepare
from core.test_run import TestRun
from storage_devices.disk import DiskType
from test_utils.size import Unit, Size
LOGGER = logging.getLogger(__name__)
@pytest.mark.parametrize("shortcut", [True, False])
@pytest.mark.parametrize('prepare_and_cleanup',
[{"core_count": 0, "cache_count": 1, "cache_type": "optane"}, ],
indirect=True)
def test_cli_start_stop_default_value(prepare_and_cleanup, shortcut):
prepare()
cache_device = next(
disk for disk in TestRun.dut.disks if disk.disk_type == DiskType.optane)
cache_device.create_partitions([Size(500, Unit.MebiByte)])
cache_device = cache_device.partitions[0]
casadm.start_cache(cache_device, shortcut=shortcut, force=True)
with TestRun.LOGGER.step("Prepare devices"):
cache_device = next(
disk for disk in TestRun.dut.disks if disk.disk_type == DiskType.optane)
cache_device.create_partitions([Size(500, Unit.MebiByte)])
cache_device = cache_device.partitions[0]
caches = casadm_parser.get_caches()
assert len(caches) == 1
assert caches[0].cache_device.system_path == cache_device.system_path
with TestRun.LOGGER.step("Start cache"):
casadm.start_cache(cache_device, shortcut=shortcut, force=True)
casadm.stop_cache(cache_id=caches[0].cache_id, shortcut=shortcut)
with TestRun.LOGGER.step("Check if cache started successfully"):
caches = casadm_parser.get_caches()
if len(caches) != 1:
TestRun.LOGGER.error(f"There is wrong caches count found in OS: {len(caches)}")
if caches[0].cache_device.system_path != cache_device.system_path:
TestRun.LOGGER.error(f"Cache started using wrong device: "
f"{caches[0].cache_device.system_path}. "
f"Should be {cache_device.system_path}")
output = casadm.list_caches(shortcut=shortcut)
caches = casadm_parser.get_caches()
assert len(caches) == 0
assert output.stdout == "No caches running"
with TestRun.LOGGER.step("Stop cache"):
casadm.stop_cache(cache_id=caches[0].cache_id, shortcut=shortcut)
with TestRun.LOGGER.step("Check if cache stopped properly"):
output = casadm.list_caches(shortcut=shortcut)
caches = casadm_parser.get_caches()
if len(caches) != 0:
TestRun.LOGGER.error(f"There is wrong caches count found in OS: {len(caches)}. "
f"Should be 0.")
if output.stdout != "No caches running":
TestRun.LOGGER.error("There is no 'No caches running' info in casadm -L output")
@pytest.mark.parametrize("shortcut", [True, False])
@ -44,7 +52,6 @@ def test_cli_start_stop_default_value(prepare_and_cleanup, shortcut):
[{"core_count": 1, "cache_count": 1, "cache_type": "optane"}],
indirect=True)
def test_cli_add_remove_default_value(prepare_and_cleanup, shortcut):
prepare()
cache_device = next(
disk for disk in TestRun.dut.disks if disk.disk_type == DiskType.optane)
cache_device.create_partitions([Size(500, Unit.MebiByte)])
@ -70,7 +77,3 @@ def test_cli_add_remove_default_value(prepare_and_cleanup, shortcut):
caches = casadm_parser.get_caches()
assert len(caches) == 0
assert output.stdout == "No caches running"
def prepare():
base_prepare()

129
test/functional/tests/conftest.py
View File

@ -7,16 +7,20 @@ import pytest
import os
import sys
import yaml
import traceback
from IPy import IP
from connection.ssh_executor import SshExecutor
sys.path.append(os.path.join(os.path.dirname(__file__), "../test-framework"))
from core.test_run_utils import TestRun
from api.cas import installer
from api.cas import casadm
from test_utils.os_utils import Udev
from log.logger import create_log
from test_utils import git_utils
# TODO: Provide basic plugin subsystem
plugins_dir = os.path.join(os.path.dirname(__file__), "../plugins")
sys.path.append(plugins_dir)
try:
@ -33,6 +37,36 @@ def get_pytest_options(request):
pytest_options["remote"] = request.config.getoption("--remote")
pytest_options["branch"] = request.config.getoption("--repo-tag")
pytest_options["force_reinstall"] = request.config.getoption("--force-reinstall")
pytest_options["log_path"] = request.config.getoption("--log-path")
def pytest_runtest_teardown():
"""
This method is executed always in the end of each test, even if it fails or raises exception in
prepare stage.
"""
TestRun.LOGGER.end_all_groups()
with TestRun.LOGGER.step("Cleanup after test"):
try:
ssh_e = type(TestRun.executor) is SshExecutor
is_active = TestRun.executor.is_active()
if ssh_e and not is_active:
TestRun.executor.wait_for_connection()
Udev.enable()
unmount_cas_devices()
casadm.stop_all_caches()
except Exception:
TestRun.LOGGER.warning("Exception occured during platform cleanup.")
if 'test_wrapper' in sys.modules:
try:
test_wrapper.cleanup()
except Exception as e:
TestRun.LOGGER.warning(f"Exception occured during test wrapper cleanup.\n{str(e)}")
TestRun.LOGGER.end()
TestRun.LOGGER.get_additional_logs()
@pytest.fixture()
@ -49,40 +83,46 @@ def prepare_and_cleanup(request):
# or it should be commented out when user want to execute tests on local machine
#
# User can also have own test wrapper, which runs test prepare, cleanup, etc.
# Then in the config/configuration.py file there should be added path to it:
# test_wrapper_dir = 'wrapper_path'
# Then it should be placed in plugins package
try:
with open(request.config.getoption('--dut-config')) as cfg:
dut_config = yaml.safe_load(cfg)
except Exception:
dut_config = {}
test_name = request.node.name.split('[')[0]
TestRun.LOGGER = create_log(f'{get_log_path_param()}', test_name)
if 'test_wrapper' in sys.modules:
if 'ip' in dut_config:
with TestRun.LOGGER.step("Dut prepare"):
try:
try:
IP(dut_config['ip'])
except ValueError:
raise Exception("IP address from configuration file is in invalid format.")
dut_config = test_wrapper.prepare(request.param, dut_config)
with open(request.config.getoption('--dut-config')) as cfg:
dut_config = yaml.safe_load(cfg)
except Exception:
dut_config = {}
TestRun.prepare(dut_config)
if 'test_wrapper' in sys.modules:
if 'ip' in dut_config:
try:
IP(dut_config['ip'])
except ValueError:
raise Exception("IP address from configuration file is in invalid format.")
dut_config = test_wrapper.prepare(request.param, dut_config)
TestRun.plugins['opencas'] = {'already_updated': False}
TestRun.prepare(dut_config)
TestRun.LOGGER.info(f"**********Test {request.node.name} started!**********")
yield
if 'test_wrapper' in sys.modules:
test_wrapper.try_setup_serial_log(dut_config)
TestRun.LOGGER.info("Test cleanup")
Udev.enable()
unmount_cas_devices()
casadm.stop_all_caches()
if 'test_wrapper' in sys.modules:
test_wrapper.cleanup()
TestRun.plugins['opencas'] = {'already_updated': False}
except Exception as e:
TestRun.LOGGER.exception(f"{str(e)}\n{traceback.format_exc()}")
TestRun.LOGGER.info(f"DUT info: {TestRun.dut}")
base_prepare()
TestRun.LOGGER.write_to_command_log("Test body")
TestRun.LOGGER.start_group("Test body")
def pytest_addoption(parser):
parser.addoption("--dut-config", action="store", default="None")
parser.addoption("--log-path", action="store",
default=f"{os.path.join(os.path.dirname(__file__), '../results')}")
parser.addoption("--remote", action="store", default="origin")
parser.addoption("--repo-tag", action="store", default="master")
parser.addoption("--force-reinstall", action="store", default="False")
@ -101,6 +141,10 @@ def get_force_param():
return pytest_options["force_reinstall"]
def get_log_path_param():
return pytest_options["log_path"]
def unmount_cas_devices():
output = TestRun.executor.run("cat /proc/mounts | grep cas")
# If exit code is '1' but stdout is empty, there is no mounted cas devices
@ -130,22 +174,23 @@ def kill_all_io():
def base_prepare():
TestRun.LOGGER.info("Base test prepare")
TestRun.LOGGER.info(f"DUT info: {TestRun.dut}")
with TestRun.LOGGER.step("Cleanup before test"):
Udev.enable()
kill_all_io()
Udev.enable()
if installer.check_if_installed():
try:
unmount_cas_devices()
casadm.stop_all_caches()
except Exception:
pass # TODO: Reboot DUT if test is executed remotely
kill_all_io()
if installer.check_if_installed():
try:
unmount_cas_devices()
casadm.stop_all_caches()
except Exception:
pass # TODO: Reboot DUT if test is executed remotely
if get_force_param() is not "False" and not TestRun.plugins['opencas']['already_updated']:
installer.reinstall_opencas()
elif not installer.check_if_installed():
installer.install_opencas()
TestRun.plugins['opencas']['already_updated'] = True
if get_force_param() is not "False" and not TestRun.plugins['opencas']['already_updated']:
installer.reinstall_opencas()
elif not installer.check_if_installed():
installer.install_opencas()
TestRun.plugins['opencas']['already_updated'] = True
TestRun.LOGGER.add_build_info(f'Commit hash:')
TestRun.LOGGER.add_build_info(f"{git_utils.get_current_commit_hash()}")
TestRun.LOGGER.add_build_info(f'Commit message:')
TestRun.LOGGER.add_build_info(f'{git_utils.get_current_commit_message()}')