import contextvars import json import os import subprocess import uuid import allure import pytest from dotenv import load_dotenv from configs import config from oslo_log import log as logging CONF = config.CONF LOG = logging.getLogger(__name__) # load env from .env load_dotenv(override=True) # --------------------------------------------------------------------------- # Allure: route per-test addCleanup callbacks to the report's `afters[]` # section (i.e. the Teardown area), rather than letting them appear as # stray steps after the test body. # # For each test we create a TestResultContainer; BaseTestCase.addCleanup # (see tests/base.py) reads this ContextVar at registration time and wraps # the callback with allure_commons._allure.fixture so that, when the # callback runs during teardown, the start_fixture / stop_fixture hooks # fire and the resulting TestAfterResult lands on the container. # --------------------------------------------------------------------------- current_allure_cleanup_container = contextvars.ContextVar( "current_allure_cleanup_container", default=None ) def _find_allure_listener(config): """Locate the allure-pytest listener so we can manipulate its reporter.""" for plugin in config.pluginmanager.get_plugins(): if hasattr(plugin, "allure_logger") and hasattr(plugin, "_cache"): return plugin return None @pytest.hookimpl(hookwrapper=True) def pytest_runtest_protocol(item, nextitem): """Create a per-test allure container that collects addCleanup callbacks.""" listener = _find_allure_listener(item.config) if listener is None: yield return # Lazy imports so the rest of conftest stays usable even if allure # internals change shape. from allure_commons.model2 import TestResultContainer from allure_commons.utils import uuid4 as allure_uuid4 # allure-pytest's listener registers this hook with tryfirst=True, so by # the time we get here its pre-yield has already pushed the test_uuid # into the cache. Capture it now -- the cache survives across the yield # but allure_logger.stop_group needs container_uuid resolved before we # reach finally, and grabbing test_uuid eagerly keeps things race-free. test_uuid = listener._cache.get(item.nodeid) container_uuid = allure_uuid4() children = [test_uuid] if test_uuid is not None else [] container = TestResultContainer(uuid=container_uuid, children=children) listener.allure_logger.start_group(container_uuid, container) token = current_allure_cleanup_container.set((listener, container_uuid)) try: yield finally: listener.allure_logger.stop_group(container_uuid) current_allure_cleanup_container.reset(token) def pytest_configure(config): """Configure pytest. This function is called automatically by pytest to configure the test environment. """ # Sets the config file path based on the TEST_ENV env variable config_file = "./configs/" + os.environ["TEST_ENV"] + ".conf" if os.path.exists(os.path.abspath(config_file)): CONF.set_config_path(os.path.abspath(config_file)) else: raise FileNotFoundError("Config file: %s doesn't exist" % config_file) def pytest_sessionstart(session): """Clean up any existing tempest processes and lock files before starting tests.""" _cleanup_tempest_environment() def _cleanup_tempest_environment(): """Check for running tempest processes and clean up lock directory.""" try: # Check for running pytest/tempest processes (excluding current process) result = subprocess.run( ["ps", "aux"], capture_output=True, text=True, timeout=10 ) if result.returncode == 0: lines = result.stdout.split('\n') tempest_processes = [] for line in lines: if ('pytest' in line or 'tempest' in line) and 'conftest.py' not in line and 'grep' not in line: # Extract PID (second column) parts = line.split() if len(parts) > 1: try: pid = int(parts[1]) # Don't kill current process if pid != os.getpid(): tempest_processes.append(pid) except (ValueError, IndexError): continue # Kill found processes if tempest_processes: print(f"⚠️ Found {len(tempest_processes)} running tempest/pytest processes, cleaning up...") for pid in tempest_processes: try: os.kill(pid, 9) # SIGKILL except (OSError, ProcessLookupError): pass # Process already terminated # Remove tempest_lock directory if it exists if os.path.exists("tempest_lock"): import shutil shutil.rmtree("tempest_lock") print("🔓 Removed tempest_lock directory") except (subprocess.TimeoutExpired, subprocess.SubprocessError, OSError): # Silently continue if cleanup fails pass def pytest_runtest_setup(item): """Setup hook for pytest. Skip a test if specified env markers are not in the 'TEST_ENV' variable. """ test_env = os.environ.get("TEST_ENV") # Check for the env_only marker on the test item. env_marker = item.get_closest_marker("env_only") if env_marker: env_args = env_marker.args # Extract the 'reason' keyword argument if provided. reason = env_marker.kwargs.get("reason") # Check that all arguments in the env_only marker are included in the TEST_ENV variable. if not all(arg in test_env for arg in env_args): # If not all marker arguments are included in TEST_ENV, skip the test. # Use the provided reason if available, otherwise use the default message. if reason: pytest.skip(reason) else: pytest.skip(f"TEST_ENV '{test_env}' does not include all of {env_args}") @pytest.fixture(scope="session", autouse=True) def add_allure_environment_property(request): """Adds test environment information to the Allure report.""" alluredir = request.config.getoption("--alluredir") if alluredir: env_props = dict() env_props["Test_Environment"] = os.environ["TEST_ENV"] env_props["Pytest_Version"] = pytest.__version__ if not os.path.exists(alluredir): os.makedirs(alluredir) allure_env_path = os.path.join(alluredir, "environment.properties") with open(allure_env_path, "w") as f: for key, value in list(env_props.items()): f.write(f"{key}={value}\n") @pytest.hookimpl(hookwrapper=True) def pytest_runtest_makereport(item): """Custom pytest hook for each test item This hook is triggered during the pytest runtest protocol, specifically for the 'call', 'setup' and 'teardown' stages. It logs the test outcome along with additional details like component, file name, class name, function name, result, and duration. Also detects authentication failures (401/403 errors) and marks them as skipped instead of failed to differentiate auth issues from actual test failures. """ outcome = yield result = outcome.get_result() # Check for auth failures (401 Unauthorized or 403 Forbidden) and mark as skipped # This handles both test failures during "call" and "setup" phases if result.failed and result.when in ["call", "setup"]: exception_str = str(result.longrepr) if result.longrepr else "" # Check for authentication/credential related errors including InvalidCredentials auth_error_patterns = [ "401", "403", "Unauthorized", "Forbidden", "You don't have enough permissions", "Authentication failed", "Token validation failed", "credentials", "credential", # Common in setup credential failures "get_primary_creds", # Specific to credential provider errors "_get_free_hash", # Another credential setup error pattern "InvalidCredentials", # Direct InvalidCredentials exception ] if any(error in exception_str for error in auth_error_patterns): # Mark test as skipped instead of failed # Simply set the outcome and reason string if result.when == "setup": skip_reason = f"Auth/credential setup failure" LOG.warning( f"Test {item.nodeid} skipped due to auth/credential setup failure" ) else: skip_reason = f"Authentication failure" LOG.warning(f"Test {item.nodeid} skipped due to auth failure") # Set outcome to skipped with proper format result.outcome = "skipped" result.wasxfail = skip_reason # Logs result when 'call' finished or setup and teardown failed. if result.when == "call" or result.failed: # Extracting component, file_name, class_name, function_name from item.nodeid token = item.nodeid.split("::") extra_fields = { "component": token[0].split("/")[2], "file_name": token[0].split("/")[-1], "class_name": token[1], "function_name": token[-1], "result": ( "error" if result.when in ["setup", "teardown"] else result.outcome ), "duration": round(result.duration, 3), } LOG.info(f"{item.nodeid} {result.outcome}", extra=extra_fields) def pytest_collection_modifyitems(items): """ Collect pytest test items and modify the collection based on custom markers. This function handles two custom markers: 1. 'tc_repeat(count)': Repeats the test case the specified number of times. 2. 'last': Ensures the test case is run last. 3. 'second_to_last': Ensures the test case is run second to last. """ repeat_tcs_list = [] last_items = [] second_to_last_items = [] other_items = [] for tc_item in items: repeat_marker = tc_item.get_closest_marker("tc_repeat") last_marker = tc_item.get_closest_marker("last") second_to_last_marker = tc_item.get_closest_marker("second_to_last") # If TC has 'tc_repeat' marker, append repeat tc per count if repeat_marker: repeat_cnt = repeat_marker.kwargs.get("count") if not repeat_cnt: continue # allure-pytest does not carry the tc_repeat marker into the # result.json labels (it gets treated like a collection-only # marker, similar to parametrize). Inject an explicit allure # tag so the historyId post-processor below can identify # these results and give each iteration a unique historyId. tc_item.add_marker(allure.tag("tc_repeat")) for i in range(1, repeat_cnt): repeat_tcs_list.append(tc_item) # Separate items with 'last' marker if last_marker: last_items.append(tc_item) elif second_to_last_marker: second_to_last_items.append(tc_item) else: other_items.append(tc_item) # Reorder items: first other items, then second to last items, and finally last items items[:] = other_items + second_to_last_items + last_items # Extend the items list with repeated test cases if repeat_tcs_list: items.extend(repeat_tcs_list) def pytest_sessionfinish(session): """Finished session, modify history for repeat test cases""" alluredir = session.config.option.allure_report_dir if os.path.exists(alluredir): _set_history_id_for_repeat_tc(results_dir=alluredir) def _set_history_id_for_repeat_tc(results_dir): """Set History ID for repeat test cases in result json""" for filename in os.listdir(results_dir): if filename.endswith("-result.json"): filepath = os.path.join(results_dir, filename) _assign_unique_history_id_each_repeat_tc(filepath) def _assign_unique_history_id_each_repeat_tc(result_json_path): """Load allure result json file and update unique historyId""" with open(result_json_path, "r") as file: data = json.load(file) # If Test case result has "tc_repeat" marker, assign unique history id and save if "labels" in data and any( "tc_repeat" in label["value"] for label in data["labels"] ): data["historyId"] = uuid.uuid4().hex with open(result_json_path, "w") as file: json.dump(data, file, indent=4)