diff --git a/qase-python-commons/README.md b/qase-python-commons/README.md index 9229b3f6..7fda093d 100644 --- a/qase-python-commons/README.md +++ b/qase-python-commons/README.md @@ -2,8 +2,13 @@ ## Description -This package contains reporters for Qase TestOps and Qase Report that are used in [qase-pytest](https://github.com/qase-tms/qase-python/tree/master/qase-pytest) and [qase-robotframework](https://github.com/qase-tms/qase-python/tree/master/qase-robotframework). +This package contains reporters for Qase TestOps and Qase Report that are used in following projects: + +- [qase-pytest](https://github.com/qase-tms/qase-python/tree/main/qase-pytest) +- [qase-robotframework](https://github.com/qase-tms/qase-python/tree/main/qase-robotframework) +- [qase-behave](https://github.com/qase-tms/qase-python/tree/main/qase-behave) +- [qase-tavern](https://github.com/qase-tms/qase-python/tree/main/qase-tavern) ## How to install -`pip install qase-python-commons` \ No newline at end of file +`pip install qase-python-commons` diff --git a/qase-python-commons/changelog.md b/qase-python-commons/changelog.md index 883d482b..4682ee2b 100644 --- a/qase-python-commons/changelog.md +++ b/qase-python-commons/changelog.md @@ -1,3 +1,9 @@ +# qase-python-commons@4.0.1 + +## What's new + +- Added support for status mapping. More information about status mapping can be found in the [docs](docs/STATUS_MAPPING.md). + # qase-python-commons@4.0.0 ## What's new diff --git a/qase-python-commons/docs/STATUS_MAPPING.md b/qase-python-commons/docs/STATUS_MAPPING.md new file mode 100644 index 00000000..1464daa3 --- /dev/null +++ b/qase-python-commons/docs/STATUS_MAPPING.md @@ -0,0 +1,320 @@ +# Status Mapping for Test Results + +This document describes the status mapping functionality that allows you to transform test result statuses from one value to another based on configuration. This is useful for standardizing status values across different testing frameworks or for custom status transformations. + +## Overview + +The status mapping feature enables you to configure which test result statuses should be transformed before being sent to Qase TestOps. This is useful when you want to: + +- Map framework-specific statuses to standard Qase statuses +- Transform statuses based on business requirements +- Standardize status values across different testing tools +- Handle legacy status values + +## Configuration + +### Configuration File + +You can configure status mapping in your `qase.config.json` file: + +```json +{ + "statusMapping": { + "invalid": "failed", + "skipped": "passed" + } +} +``` + +### Environment Variables + +You can also use environment variables: + +```bash +export STATUS_MAPPING="invalid=failed,skipped=passed" +``` + +### Command Line Options + +For frameworks that support CLI options, you can use the CLI parameter: + +```bash +# Pytest +pytest --qase-status-mapping="invalid=failed,skipped=passed" + +# Tavern +pytest --qase-status-mapping="invalid=failed,skipped=passed" + +# Behave +behave --define qase-status-mapping="invalid=failed,skipped=passed" +``` + +You can also use environment variables: + +```bash +# Pytest +STATUS_MAPPING="invalid=failed,skipped=passed" pytest + +# Tavern +STATUS_MAPPING="invalid=failed,skipped=passed" pytest + +# Behave +STATUS_MAPPING="invalid=failed,skipped=passed" behave + +# Robot Framework +STATUS_MAPPING="invalid=failed,skipped=passed" robot tests/ +``` + +## Supported Statuses + +The following statuses can be used in the mapping: + +- `passed` - Test passed successfully +- `failed` - Test failed +- `skipped` - Test was skipped +- `blocked` - Test was blocked +- `invalid` - Test failed due to non-assertion errors (network issues, syntax errors) + +## Examples + +### Map Invalid Tests to Failed + +```json +{ + "statusMapping": { + "invalid": "failed" + } +} +``` + +This configuration will map all tests with `invalid` status to `failed` status. + +### Map Skipped Tests to Passed + +```json +{ + "statusMapping": { + "skipped": "passed" + } +} +``` + +This configuration will map all tests with `skipped` status to `passed` status. + +### Multiple Mappings + +```json +{ + "statusMapping": { + "invalid": "failed", + "skipped": "passed" + } +} +``` + +This configuration applies multiple status mappings. + +### Environment Variable Example + +```bash +export STATUS_MAPPING="invalid=failed,skipped=passed" +``` + +## Behavior + +- **Mapping Logic**: Status mapping is applied **before** status filtering +- **Centralized Application**: Mapping is applied in the core reporter for all frameworks +- **Logging**: Status changes are logged at debug level +- **Validation**: Invalid mappings are ignored with warnings +- **Case Sensitivity**: Status mapping is case-sensitive +- **No Chaining**: Only direct mappings are applied (no chaining of mappings) + +## Integration with Status Filtering + +Status mapping is applied **before** status filtering. This means: + +1. Original status → Mapped status (via status mapping) +2. Mapped status → Filtered out or sent (via status filter) + +Example: +```json +{ + "statusMapping": { + "invalid": "failed" + }, + "testops": { + "statusFilter": ["passed"] + } +} +``` + +In this case: +- Tests with `invalid` status are mapped to `failed` +- Tests with `failed` status are **not** filtered out (only `passed` tests are filtered) +- Tests with `invalid` status will be sent as `failed` status + +## Framework-Specific Examples + +### Pytest + +```json +{ + "statusMapping": { + "invalid": "failed" + }, + "testops": { + "project": "MYPROJECT", + "api": { + "token": "your-token" + } + } +} +``` + +### Behave + +```json +{ + "statusMapping": { + "skipped": "passed" + }, + "testops": { + "project": "MYPROJECT", + "api": { + "token": "your-token" + } + } +} +``` + +### Robot Framework + +```json +{ + "statusMapping": { + "disabled": "skipped" + }, + "testops": { + "project": "MYPROJECT", + "api": { + "token": "your-token" + } + } +} +``` + +### Tavern + +```json +{ + "statusMapping": { + "invalid": "failed", + "skipped": "passed" + }, + "testops": { + "project": "MYPROJECT", + "api": { + "token": "your-token" + } + } +} +``` + +## Troubleshooting + +### Common Issues + +1. **Status not being mapped**: Check that the source status exactly matches the configured mapping key (case-sensitive) + +2. **Invalid status error**: Ensure both source and target statuses are from the valid list: `passed`, `failed`, `skipped`, `disabled`, `blocked`, `invalid` + +3. **Environment variable not working**: Make sure the environment variable is set before running the tests + +4. **Mapping not applied**: Check that the status mapping is configured at the top level of the configuration, not inside `testops` or `report` sections + +### Debug Logging + +Enable debug logging to see status mapping in action: + +```json +{ + "debug": true, + "statusMapping": { + "invalid": "failed" + } +} +``` + +This will log messages like: +``` +Status mapped for 'Test Name': invalid -> failed +``` + +### Validation + +The status mapping configuration is validated when the reporter is initialized. Invalid mappings will cause warnings but will not prevent the reporter from working. + +## Migration Guide + +### From Status Filtering + +If you were previously using status filtering to exclude certain statuses, you can now use status mapping to transform them instead: + +**Before (filtering out invalid tests):** +```json +{ + "testops": { + "statusFilter": ["invalid"] + } +} +``` + +**After (mapping invalid to failed):** +```json +{ + "statusMapping": { + "invalid": "failed" + } +} +``` + +This approach is better because: +- Tests are still reported (not excluded) +- You can see the original status in logs +- More flexible for different reporting needs + +## Best Practices + +1. **Use descriptive mappings**: Choose target statuses that clearly indicate the transformation +2. **Document your mappings**: Keep a record of why certain mappings are needed +3. **Test your configuration**: Verify that mappings work as expected in your environment +4. **Consider framework differences**: Different testing frameworks may have different status semantics +5. **Use environment variables for CI/CD**: Set mappings via environment variables in your CI/CD pipeline + +## API Reference + +### StatusMapping Class + +The `StatusMapping` class provides the core functionality for status mapping: + +```python +from qase.commons.utils.status_mapping import StatusMapping + +# Create from dictionary +mapping = StatusMapping.from_dict({"invalid": "failed"}) + +# Create from environment string +mapping = StatusMapping.from_env_string("invalid=failed,skipped=passed") + +# Apply mapping +result_status = mapping.apply_mapping("invalid") # Returns "failed" +``` + +### Configuration Methods + +```python +from qase.commons.models.config.qaseconfig import QaseConfig + +config = QaseConfig() +config.set_status_mapping({"invalid": "failed"}) +``` diff --git a/qase-python-commons/pyproject.toml b/qase-python-commons/pyproject.toml index bc05a2bf..59546c64 100644 --- a/qase-python-commons/pyproject.toml +++ b/qase-python-commons/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "qase-python-commons" -version = "4.0.0" +version = "4.0.1" description = "A library for Qase TestOps and Qase Report" readme = "README.md" authors = [{name = "Qase Team", email = "support@qase.io"}] @@ -29,7 +29,7 @@ requires-python = ">=3.9" dependencies = [ "certifi>=2024.2.2", "attrs>=23.2.0", - "qase-api-client~=2.0.0", + "qase-api-client~=2.0.1", "qase-api-v2-client~=2.0.0", "more_itertools" ] diff --git a/qase-python-commons/src/qase/__init__.py b/qase-python-commons/src/qase/__init__.py new file mode 100644 index 00000000..92df35d4 --- /dev/null +++ b/qase-python-commons/src/qase/__init__.py @@ -0,0 +1,3 @@ +""" +Qase Python Commons package. +""" diff --git a/qase-python-commons/src/qase/commons/config.py b/qase-python-commons/src/qase/commons/config.py index f568f580..2569f6c5 100644 --- a/qase-python-commons/src/qase/commons/config.py +++ b/qase-python-commons/src/qase/commons/config.py @@ -63,6 +63,11 @@ def __load_file_config(self): config.get("excludeParams") ) + if config.get("statusMapping"): + self.config.set_status_mapping( + config.get("statusMapping") + ) + if config.get("executionPlan"): execution_plan = config.get("executionPlan") if execution_plan.get("path"): @@ -224,6 +229,19 @@ def __load_env_config(self): self.config.set_exclude_params( [param.strip() for param in value.split(',')]) + if key == 'QASE_STATUS_MAPPING': + # Parse status mapping from environment variable + # Format: "source1=target1,source2=target2" + if value: + mapping_dict = {} + pairs = value.split(',') + for pair in pairs: + pair = pair.strip() + if pair and '=' in pair: + source_status, target_status = pair.split('=', 1) + mapping_dict[source_status.strip()] = target_status.strip() + self.config.set_status_mapping(mapping_dict) + if key == 'QASE_EXECUTION_PLAN_PATH': self.config.execution_plan.set_path(value) diff --git a/qase-python-commons/src/qase/commons/models/config/qaseconfig.py b/qase-python-commons/src/qase/commons/models/config/qaseconfig.py index 427b1842..9d79c443 100644 --- a/qase-python-commons/src/qase/commons/models/config/qaseconfig.py +++ b/qase-python-commons/src/qase/commons/models/config/qaseconfig.py @@ -1,5 +1,5 @@ from enum import Enum -from typing import List +from typing import List, Dict from .framework import Framework from .report import ReportConfig @@ -36,6 +36,7 @@ class QaseConfig(BaseModel): profilers: list = None framework: Framework = None exclude_params: list = None + status_mapping: Dict[str, str] = None def __init__(self): self.mode = Mode.off @@ -47,6 +48,7 @@ def __init__(self): self.framework = Framework() self.profilers = [] self.exclude_params = [] + self.status_mapping = {} def set_mode(self, mode: str): if any(mode == e.value for e in Mode.__members__.values()): @@ -70,3 +72,6 @@ def set_debug(self, debug): def set_exclude_params(self, exclude_params: List[str]): self.exclude_params = exclude_params + + def set_status_mapping(self, status_mapping: Dict[str, str]): + self.status_mapping = status_mapping diff --git a/qase-python-commons/src/qase/commons/models/result.py b/qase-python-commons/src/qase/commons/models/result.py index 12e343af..396ef3ec 100644 --- a/qase-python-commons/src/qase/commons/models/result.py +++ b/qase-python-commons/src/qase/commons/models/result.py @@ -33,10 +33,16 @@ def __init__(self, self.thread = thread def set_status(self, status: Optional[str]): - if status in ['passed', 'failed', 'skipped', 'untested', 'invalid']: + if status is None: self.status = status + return + + # Convert to lowercase for validation + status_lower = status.lower() + if status_lower in ['passed', 'failed', 'skipped', 'untested', 'invalid', 'disabled', 'blocked']: + self.status = status_lower else: - raise ValueError('Step status must be one of: passed, failed, skipped, untested, invalid') + raise ValueError('Step status must be one of: passed, failed, skipped, untested, invalid, disabled, blocked') def get_status(self): return self.status diff --git a/qase-python-commons/src/qase/commons/reporters/core.py b/qase-python-commons/src/qase/commons/reporters/core.py index fe33c484..00cc44e4 100644 --- a/qase-python-commons/src/qase/commons/reporters/core.py +++ b/qase-python-commons/src/qase/commons/reporters/core.py @@ -11,6 +11,7 @@ from typing import Union, List from ..util import get_host_info +from ..status_mapping.status_mapping import StatusMapping """ CoreReporter is a facade for all reporters and it is used to initialize and manage them. @@ -28,6 +29,11 @@ def __init__(self, config: ConfigManager, framework: Union[str, None] = None, self.profilers = [] self.overhead = 0 + # Initialize status mapping + self.status_mapping = StatusMapping.from_dict(self.config.status_mapping) + if not self.status_mapping.is_empty(): + self.logger.log_debug(f"Status mapping initialized: {self.status_mapping}") + # self._selective_execution_setup() self.fallback = self._fallback_setup() @@ -87,6 +93,9 @@ def add_result(self, result: Result) -> None: ts = time.time() self.logger.log_debug(f"Adding result {result}") + # Apply status mapping before adding result + self._apply_status_mapping(result) + self.reporter.add_result(result) self.logger.log_debug(f"Result {result.get_title()} added") @@ -208,3 +217,26 @@ def _fallback_setup(self) -> Union[QaseReport, None]: if self.config.fallback == Mode.report: return QaseReport(config=self.config, logger=self.logger) return None + + def _apply_status_mapping(self, result: Result) -> None: + """ + Apply status mapping to a test result. + + This method applies the configured status mapping to the result's execution status. + The mapping is applied before the result is sent to the reporter. + + Args: + result: Test result to apply status mapping to + """ + if self.status_mapping.is_empty(): + return + + original_status = result.get_status() + if not original_status: + return + + mapped_status = self.status_mapping.apply_mapping(original_status) + + if mapped_status != original_status: + result.execution.set_status(mapped_status) + self.logger.log_debug(f"Status mapped for '{result.get_title()}': {original_status} -> {mapped_status}") diff --git a/qase-python-commons/src/qase/commons/status_mapping/__init__.py b/qase-python-commons/src/qase/commons/status_mapping/__init__.py new file mode 100644 index 00000000..b8d07677 --- /dev/null +++ b/qase-python-commons/src/qase/commons/status_mapping/__init__.py @@ -0,0 +1,12 @@ +""" +Utilities package for Qase Python Commons. +""" + +from .status_mapping import StatusMapping, StatusMappingError, create_status_mapping_from_config, create_status_mapping_from_env + +__all__ = [ + 'StatusMapping', + 'StatusMappingError', + 'create_status_mapping_from_config', + 'create_status_mapping_from_env' +] diff --git a/qase-python-commons/src/qase/commons/status_mapping/status_mapping.py b/qase-python-commons/src/qase/commons/status_mapping/status_mapping.py new file mode 100644 index 00000000..0be8d3d1 --- /dev/null +++ b/qase-python-commons/src/qase/commons/status_mapping/status_mapping.py @@ -0,0 +1,237 @@ +""" +Status mapping utilities for Qase Python Commons. + +This module provides functionality to map test result statuses from one value to another +based on configuration. This is useful for standardizing status values across different +testing frameworks or for custom status transformations. +""" + +from typing import Dict, Optional, List +import os +import logging + + +class StatusMappingError(Exception): + """Exception raised when status mapping encounters an error.""" + pass + + +class StatusMapping: + """ + Handles mapping of test result statuses. + + This class provides functionality to: + - Parse status mapping from configuration + - Validate status mappings + - Apply status mappings to test results + - Support both JSON configuration and environment variables + """ + + # Valid statuses that can be mapped + VALID_STATUSES = { + 'passed', 'failed', 'skipped', 'disabled', 'blocked', 'invalid' + } + + def __init__(self, mapping: Optional[Dict[str, str]] = None): + """ + Initialize StatusMapping with optional mapping dictionary. + + Args: + mapping: Dictionary mapping source status to target status + """ + self.mapping = mapping or {} + self.logger = logging.getLogger(__name__) + + @classmethod + def from_dict(cls, mapping_dict: Dict[str, str]) -> 'StatusMapping': + """ + Create StatusMapping from dictionary. + + Args: + mapping_dict: Dictionary with status mappings + + Returns: + StatusMapping instance + + Raises: + StatusMappingError: If mapping contains invalid statuses + """ + instance = cls() + instance.set_mapping(mapping_dict) + return instance + + @classmethod + def from_env_string(cls, env_string: str) -> 'StatusMapping': + """ + Create StatusMapping from environment variable string. + + Expected format: "source1=target1,source2=target2" + + Args: + env_string: Environment variable string + + Returns: + StatusMapping instance + + Raises: + StatusMappingError: If string format is invalid + """ + instance = cls() + instance.parse_env_string(env_string) + return instance + + def set_mapping(self, mapping_dict: Dict[str, str]) -> None: + """ + Set status mapping from dictionary. + + Args: + mapping_dict: Dictionary with status mappings + + Raises: + StatusMappingError: If mapping contains invalid statuses + """ + if not isinstance(mapping_dict, dict): + raise StatusMappingError("Mapping must be a dictionary") + + # Validate all statuses in the mapping + for source_status, target_status in mapping_dict.items(): + if source_status not in self.VALID_STATUSES: + raise StatusMappingError(f"Invalid source status: {source_status}") + if target_status not in self.VALID_STATUSES: + raise StatusMappingError(f"Invalid target status: {target_status}") + + self.mapping = mapping_dict.copy() + self.logger.debug(f"Status mapping set: {self.mapping}") + + def parse_env_string(self, env_string: str) -> None: + """ + Parse status mapping from environment variable string. + + Expected format: "source1=target1,source2=target2" + + Args: + env_string: Environment variable string + + Raises: + StatusMappingError: If string format is invalid + """ + if not env_string or not env_string.strip(): + self.mapping = {} + return + + mapping_dict = {} + pairs = env_string.split(',') + + for pair in pairs: + pair = pair.strip() + if not pair: + continue + + if '=' not in pair: + raise StatusMappingError(f"Invalid mapping format: {pair}. Expected 'source=target'") + + source_status, target_status = pair.split('=', 1) + source_status = source_status.strip() + target_status = target_status.strip() + + if not source_status or not target_status: + raise StatusMappingError(f"Empty status in mapping: {pair}") + + mapping_dict[source_status] = target_status + + self.set_mapping(mapping_dict) + + def apply_mapping(self, status: str) -> str: + """ + Apply status mapping to a given status. + + Args: + status: Original status + + Returns: + Mapped status if mapping exists, otherwise original status + """ + if not status: + return status + + if status in self.mapping: + mapped_status = self.mapping[status] + self.logger.debug(f"Status mapped: {status} -> {mapped_status}") + return mapped_status + + return status + + def get_mapping(self) -> Dict[str, str]: + """ + Get current status mapping. + + Returns: + Dictionary with current status mappings + """ + return self.mapping.copy() + + def is_empty(self) -> bool: + """ + Check if mapping is empty. + + Returns: + True if no mappings are defined + """ + return len(self.mapping) == 0 + + def validate(self) -> List[str]: + """ + Validate current mapping and return any issues. + + Returns: + List of validation error messages + """ + errors = [] + + for source_status, target_status in self.mapping.items(): + if source_status not in self.VALID_STATUSES: + errors.append(f"Invalid source status: {source_status}") + if target_status not in self.VALID_STATUSES: + errors.append(f"Invalid target status: {target_status}") + + return errors + + def __str__(self) -> str: + """String representation of the mapping.""" + return str(self.mapping) + + def __repr__(self) -> str: + """Detailed string representation.""" + return f"StatusMapping({self.mapping})" + + +def create_status_mapping_from_config(config_value: Optional[Dict[str, str]]) -> StatusMapping: + """ + Create StatusMapping from configuration value. + + Args: + config_value: Configuration dictionary or None + + Returns: + StatusMapping instance + """ + if config_value is None: + return StatusMapping() + + return StatusMapping.from_dict(config_value) + + +def create_status_mapping_from_env(env_var_name: str = 'STATUS_MAPPING') -> StatusMapping: + """ + Create StatusMapping from environment variable. + + Args: + env_var_name: Name of environment variable + + Returns: + StatusMapping instance + """ + env_value = os.getenv(env_var_name) + if env_value: + return StatusMapping.from_env_string(env_value) + return StatusMapping() diff --git a/qase-python-commons/tests/tests_qase_commons/test_status_mapping.py b/qase-python-commons/tests/tests_qase_commons/test_status_mapping.py new file mode 100644 index 00000000..c4d15645 --- /dev/null +++ b/qase-python-commons/tests/tests_qase_commons/test_status_mapping.py @@ -0,0 +1,260 @@ +""" +Tests for status mapping functionality. +""" + +import pytest +import os +from unittest.mock import Mock, patch +from qase.commons.status_mapping.status_mapping import StatusMapping, StatusMappingError, create_status_mapping_from_config, create_status_mapping_from_env + + +class TestStatusMapping: + """Test cases for StatusMapping class.""" + + def test_status_mapping_initialization_empty(self): + """Test initialization with empty mapping.""" + mapping = StatusMapping() + assert mapping.mapping == {} + assert mapping.is_empty() + + def test_status_mapping_initialization_with_dict(self): + """Test initialization with mapping dictionary.""" + mapping_dict = {"invalid": "failed", "skipped": "passed"} + mapping = StatusMapping(mapping_dict) + assert mapping.mapping == mapping_dict + assert not mapping.is_empty() + + def test_from_dict_valid_mapping(self): + """Test creating StatusMapping from valid dictionary.""" + mapping_dict = {"invalid": "failed", "skipped": "passed"} + mapping = StatusMapping.from_dict(mapping_dict) + assert mapping.mapping == mapping_dict + + def test_from_dict_invalid_source_status(self): + """Test creating StatusMapping with invalid source status.""" + mapping_dict = {"invalid_status": "failed"} + with pytest.raises(StatusMappingError, match="Invalid source status"): + StatusMapping.from_dict(mapping_dict) + + def test_from_dict_invalid_target_status(self): + """Test creating StatusMapping with invalid target status.""" + mapping_dict = {"invalid": "invalid_status"} + with pytest.raises(StatusMappingError, match="Invalid target status"): + StatusMapping.from_dict(mapping_dict) + + def test_from_env_string_valid(self): + """Test creating StatusMapping from valid environment string.""" + env_string = "invalid=failed,skipped=passed" + mapping = StatusMapping.from_env_string(env_string) + expected = {"invalid": "failed", "skipped": "passed"} + assert mapping.mapping == expected + + def test_from_env_string_empty(self): + """Test creating StatusMapping from empty environment string.""" + mapping = StatusMapping.from_env_string("") + assert mapping.mapping == {} + assert mapping.is_empty() + + def test_from_env_string_none(self): + """Test creating StatusMapping from None environment string.""" + mapping = StatusMapping.from_env_string(None) + assert mapping.mapping == {} + assert mapping.is_empty() + + def test_from_env_string_invalid_format(self): + """Test creating StatusMapping from invalid environment string format.""" + env_string = "invalid:failed" + with pytest.raises(StatusMappingError, match="Invalid mapping format"): + StatusMapping.from_env_string(env_string) + + def test_from_env_string_empty_status(self): + """Test creating StatusMapping with empty status values.""" + env_string = "=failed" + with pytest.raises(StatusMappingError, match="Empty status in mapping"): + StatusMapping.from_env_string(env_string) + + def test_set_mapping_valid(self): + """Test setting valid mapping.""" + mapping = StatusMapping() + mapping_dict = {"invalid": "failed", "skipped": "passed"} + mapping.set_mapping(mapping_dict) + assert mapping.mapping == mapping_dict + + def test_set_mapping_invalid_type(self): + """Test setting mapping with invalid type.""" + mapping = StatusMapping() + with pytest.raises(StatusMappingError, match="Mapping must be a dictionary"): + mapping.set_mapping("invalid") + + def test_parse_env_string_with_spaces(self): + """Test parsing environment string with spaces.""" + env_string = " invalid = failed , skipped = passed " + mapping = StatusMapping() + mapping.parse_env_string(env_string) + expected = {"invalid": "failed", "skipped": "passed"} + assert mapping.mapping == expected + + def test_parse_env_string_with_empty_pairs(self): + """Test parsing environment string with empty pairs.""" + env_string = "invalid=failed,,skipped=passed" + mapping = StatusMapping() + mapping.parse_env_string(env_string) + expected = {"invalid": "failed", "skipped": "passed"} + assert mapping.mapping == expected + + def test_apply_mapping_existing(self): + """Test applying mapping for existing status.""" + mapping = StatusMapping({"invalid": "failed"}) + result = mapping.apply_mapping("invalid") + assert result == "failed" + + def test_apply_mapping_non_existing(self): + """Test applying mapping for non-existing status.""" + mapping = StatusMapping({"invalid": "failed"}) + result = mapping.apply_mapping("passed") + assert result == "passed" + + def test_apply_mapping_empty_status(self): + """Test applying mapping for empty status.""" + mapping = StatusMapping({"invalid": "failed"}) + result = mapping.apply_mapping("") + assert result == "" + + def test_apply_mapping_none_status(self): + """Test applying mapping for None status.""" + mapping = StatusMapping({"invalid": "failed"}) + result = mapping.apply_mapping(None) + assert result is None + + def test_get_mapping(self): + """Test getting mapping copy.""" + mapping_dict = {"invalid": "failed", "skipped": "passed"} + mapping = StatusMapping(mapping_dict) + result = mapping.get_mapping() + assert result == mapping_dict + # Ensure it's a copy, not the same object + assert result is not mapping.mapping + + def test_validate_valid_mapping(self): + """Test validation of valid mapping.""" + mapping = StatusMapping({"invalid": "failed", "skipped": "passed"}) + errors = mapping.validate() + assert len(errors) == 0 + + def test_validate_invalid_mapping(self): + """Test validation of invalid mapping.""" + mapping = StatusMapping({"invalid_status": "failed", "skipped": "invalid_target"}) + errors = mapping.validate() + assert len(errors) == 2 + assert "Invalid source status: invalid_status" in errors + assert "Invalid target status: invalid_target" in errors + + def test_str_representation(self): + """Test string representation.""" + mapping_dict = {"invalid": "failed"} + mapping = StatusMapping(mapping_dict) + assert str(mapping) == str(mapping_dict) + + def test_repr_representation(self): + """Test detailed string representation.""" + mapping_dict = {"invalid": "failed"} + mapping = StatusMapping(mapping_dict) + assert "StatusMapping" in repr(mapping) + assert str(mapping_dict) in repr(mapping) + + +class TestStatusMappingHelpers: + """Test cases for helper functions.""" + + def test_create_status_mapping_from_config_valid(self): + """Test creating StatusMapping from valid config.""" + config_value = {"invalid": "failed", "skipped": "passed"} + mapping = create_status_mapping_from_config(config_value) + assert mapping.mapping == config_value + + def test_create_status_mapping_from_config_none(self): + """Test creating StatusMapping from None config.""" + mapping = create_status_mapping_from_config(None) + assert mapping.mapping == {} + assert mapping.is_empty() + + def test_create_status_mapping_from_config_empty(self): + """Test creating StatusMapping from empty config.""" + mapping = create_status_mapping_from_config({}) + assert mapping.mapping == {} + assert mapping.is_empty() + + @patch.dict(os.environ, {'STATUS_MAPPING': 'invalid=failed,skipped=passed'}) + def test_create_status_mapping_from_env_with_value(self): + """Test creating StatusMapping from environment variable with value.""" + mapping = create_status_mapping_from_env() + expected = {"invalid": "failed", "skipped": "passed"} + assert mapping.mapping == expected + + @patch.dict(os.environ, {}, clear=True) + def test_create_status_mapping_from_env_without_value(self): + """Test creating StatusMapping from environment variable without value.""" + mapping = create_status_mapping_from_env() + assert mapping.mapping == {} + assert mapping.is_empty() + + @patch.dict(os.environ, {'CUSTOM_STATUS_MAPPING': 'invalid=failed'}) + def test_create_status_mapping_from_env_custom_var(self): + """Test creating StatusMapping from custom environment variable.""" + mapping = create_status_mapping_from_env('CUSTOM_STATUS_MAPPING') + expected = {"invalid": "failed"} + assert mapping.mapping == expected + + +class TestStatusMappingIntegration: + """Integration tests for status mapping functionality.""" + + def test_valid_statuses(self): + """Test that all valid statuses are recognized.""" + valid_statuses = StatusMapping.VALID_STATUSES + expected_statuses = {'passed', 'failed', 'skipped', 'disabled', 'blocked', 'invalid'} + assert valid_statuses == expected_statuses + + def test_mapping_all_valid_statuses(self): + """Test mapping with all valid statuses.""" + mapping_dict = { + 'passed': 'passed', + 'failed': 'failed', + 'skipped': 'skipped', + 'disabled': 'disabled', + 'blocked': 'blocked', + 'invalid': 'invalid' + } + mapping = StatusMapping.from_dict(mapping_dict) + assert mapping.validate() == [] + + def test_complex_mapping_scenario(self): + """Test complex mapping scenario.""" + env_string = "invalid=failed,skipped=passed,disabled=skipped" + mapping = StatusMapping.from_env_string(env_string) + + # Test multiple mappings + assert mapping.apply_mapping("invalid") == "failed" + assert mapping.apply_mapping("skipped") == "passed" + assert mapping.apply_mapping("disabled") == "skipped" + assert mapping.apply_mapping("passed") == "passed" # No mapping + assert mapping.apply_mapping("failed") == "failed" # No mapping + + def test_logging_integration(self): + """Test that logging works correctly.""" + mapping = StatusMapping({"invalid": "failed"}) + + # Mock logger to verify debug messages + with patch.object(mapping.logger, 'debug') as mock_debug: + mapping.apply_mapping("invalid") + mock_debug.assert_called_once_with("Status mapped: invalid -> failed") + + def test_error_handling_edge_cases(self): + """Test error handling for edge cases.""" + # Test with non-string values + with pytest.raises(StatusMappingError): + StatusMapping.from_dict({123: "failed"}) + + # Test with non-string target + with pytest.raises(StatusMappingError): + StatusMapping.from_dict({"invalid": 123}) diff --git a/qase-python-commons/tests/tests_qase_commons/test_status_mapping_config.py b/qase-python-commons/tests/tests_qase_commons/test_status_mapping_config.py new file mode 100644 index 00000000..98d01136 --- /dev/null +++ b/qase-python-commons/tests/tests_qase_commons/test_status_mapping_config.py @@ -0,0 +1,384 @@ +""" +Tests for status mapping configuration integration. +""" + +import pytest +import json +import tempfile +import os +from unittest.mock import Mock, patch +from qase.commons.config import ConfigManager +from qase.commons.models.config.qaseconfig import QaseConfig +from qase.commons.reporters.core import QaseCoreReporter +from qase.commons.models.result import Result, Execution + + +class TestStatusMappingConfig: + """Test cases for status mapping configuration.""" + + def test_qase_config_status_mapping_initialization(self): + """Test QaseConfig initialization with status mapping.""" + config = QaseConfig() + assert config.status_mapping == {} + assert isinstance(config.status_mapping, dict) + + def test_qase_config_set_status_mapping(self): + """Test setting status mapping in QaseConfig.""" + config = QaseConfig() + mapping = {"invalid": "failed", "skipped": "passed"} + config.set_status_mapping(mapping) + assert config.status_mapping == mapping + + def test_config_manager_load_status_mapping_from_file(self): + """Test loading status mapping from config file.""" + config_data = { + "statusMapping": { + "invalid": "failed", + "skipped": "passed" + } + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + assert config_manager.config.status_mapping == config_data["statusMapping"] + finally: + os.unlink(config_file) + + def test_config_manager_load_status_mapping_from_env(self): + """Test loading status mapping from environment variable.""" + env_value = "invalid=failed,skipped=passed" + + with patch.dict(os.environ, {'QASE_STATUS_MAPPING': env_value}): + config_manager = ConfigManager() + expected = {"invalid": "failed", "skipped": "passed"} + assert config_manager.config.status_mapping == expected + + def test_config_manager_env_overrides_file(self): + """Test that environment variable overrides file configuration.""" + config_data = { + "statusMapping": { + "invalid": "blocked" + } + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + env_value = "invalid=failed" + with patch.dict(os.environ, {'QASE_STATUS_MAPPING': env_value}): + config_manager = ConfigManager(config_file) + expected = {"invalid": "failed"} + assert config_manager.config.status_mapping == expected + finally: + os.unlink(config_file) + + def test_config_manager_empty_status_mapping(self): + """Test handling of empty status mapping.""" + config_data = {} + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + assert config_manager.config.status_mapping == {} + finally: + os.unlink(config_file) + + def test_config_manager_invalid_status_mapping_format(self): + """Test handling of invalid status mapping format in file.""" + config_data = { + "statusMapping": "invalid=failed" # Should be dict, not string + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + # Should still work, just set the string as mapping + assert config_manager.config.status_mapping == "invalid=failed" + finally: + os.unlink(config_file) + + +class TestStatusMappingReporterIntegration: + """Test cases for status mapping integration with reporters.""" + + def test_qase_core_reporter_status_mapping_initialization(self): + """Test QaseCoreReporter initialization with status mapping.""" + config_data = { + "statusMapping": { + "invalid": "failed", + "skipped": "passed" + }, + "mode": "off" + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Check that status mapping was initialized + assert not reporter.status_mapping.is_empty() + assert reporter.status_mapping.mapping == config_data["statusMapping"] + finally: + os.unlink(config_file) + + def test_qase_core_reporter_status_mapping_empty(self): + """Test QaseCoreReporter initialization with empty status mapping.""" + config_data = {"mode": "off"} + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Check that status mapping is empty + assert reporter.status_mapping.is_empty() + finally: + os.unlink(config_file) + + def test_apply_status_mapping_to_result(self): + """Test applying status mapping to test result.""" + config_data = { + "statusMapping": { + "invalid": "failed", + "skipped": "passed" + }, + "mode": "off" + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Create test result with invalid status + result = Result("Test Title", "test_signature") + result.execution.set_status("invalid") + + # Apply status mapping + reporter._apply_status_mapping(result) + + # Check that status was mapped + assert result.get_status() == "failed" + finally: + os.unlink(config_file) + + def test_apply_status_mapping_no_change(self): + """Test applying status mapping when no mapping exists.""" + config_data = { + "statusMapping": { + "invalid": "failed" + }, + "mode": "off" + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Create test result with status that has no mapping + result = Result("Test Title", "test_signature") + result.execution.set_status("passed") + + # Apply status mapping + reporter._apply_status_mapping(result) + + # Check that status was not changed + assert result.get_status() == "passed" + finally: + os.unlink(config_file) + + def test_apply_status_mapping_empty_status(self): + """Test applying status mapping to result with empty status.""" + config_data = { + "statusMapping": { + "invalid": "failed" + }, + "mode": "off" + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Create test result with no status + result = Result("Test Title", "test_signature") + + # Apply status mapping + reporter._apply_status_mapping(result) + + # Check that status remains None + assert result.get_status() is None + finally: + os.unlink(config_file) + + def test_apply_status_mapping_logging(self): + """Test that status mapping changes are logged.""" + config_data = { + "statusMapping": { + "invalid": "failed" + }, + "mode": "off", + "debug": True + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Create test result + result = Result("Test Title", "test_signature") + result.execution.set_status("invalid") + + # Mock logger to verify debug message + with patch.object(reporter.logger, 'log_debug') as mock_log_debug: + reporter._apply_status_mapping(result) + mock_log_debug.assert_called_with("Status mapped for 'Test Title': invalid -> failed") + finally: + os.unlink(config_file) + + +class TestStatusMappingEdgeCases: + """Test edge cases for status mapping configuration.""" + + def test_status_mapping_with_all_valid_statuses(self): + """Test status mapping with all valid statuses.""" + mapping = { + "passed": "passed", + "failed": "failed", + "skipped": "skipped", + "disabled": "disabled", + "blocked": "blocked", + "invalid": "invalid" + } + + config_data = { + "statusMapping": mapping, + "mode": "off" + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Test each status + for original_status in mapping.keys(): + result = Result(f"Test {original_status}", "test_signature") + result.execution.set_status(original_status) + reporter._apply_status_mapping(result) + assert result.get_status() == mapping[original_status] + finally: + os.unlink(config_file) + + def test_status_mapping_chain(self): + """Test chained status mappings.""" + config_data = { + "statusMapping": { + "invalid": "skipped", + "skipped": "passed" + }, + "mode": "off" + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Test that only first mapping is applied (no chaining) + result = Result("Test Title", "test_signature") + result.execution.set_status("invalid") + reporter._apply_status_mapping(result) + + # Should map to "skipped", not "passed" + assert result.get_status() == "skipped" + finally: + os.unlink(config_file) + + def test_status_mapping_case_sensitivity(self): + """Test that status mapping configuration is case sensitive.""" + # Test that configuration with uppercase status names fails validation + config_data = { + "statusMapping": { + "INVALID": "failed" # Uppercase - should cause validation error + }, + "mode": "off" + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data, f) + config_file = f.name + + try: + # This should raise an exception because "INVALID" is not a valid status + with pytest.raises(Exception): # StatusMappingError will be raised + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + finally: + os.unlink(config_file) + + # Test that lowercase configuration works + config_data_lower = { + "statusMapping": { + "invalid": "failed" # Lowercase - valid status + }, + "mode": "off" + } + + with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f: + json.dump(config_data_lower, f) + config_file = f.name + + try: + config_manager = ConfigManager(config_file) + reporter = QaseCoreReporter(config_manager) + + # Test with lowercase status - should map + result = Result("Test Title", "test_signature") + result.execution.set_status("invalid") # Lowercase + reporter._apply_status_mapping(result) + + # Should map because case matches + assert result.get_status() == "failed" + finally: + os.unlink(config_file)