diff --git a/src/reportportal/ap.py b/src/reportportal/ap.py index 4cdb44d..5c6a7a4 100644 --- a/src/reportportal/ap.py +++ b/src/reportportal/ap.py @@ -229,6 +229,12 @@ def _add_query_arguments(subparsers: argparse.ArgumentParser, defaults: dict) -> help="Output only test case names for STEP items (one per line, no formatting, excludes SUITE items). Only applies when --launch-id is specified.", default=False ) + parser.add_argument( + "--show-logs", + action="store_true", + help="Display error logs for failed test items (only applies when --launch-id is specified)", + default=False + ) parser.add_argument( "--limit", type=int, diff --git a/src/reportportal/rp_api_client.py b/src/reportportal/rp_api_client.py index 4252846..52928b5 100644 --- a/src/reportportal/rp_api_client.py +++ b/src/reportportal/rp_api_client.py @@ -383,6 +383,26 @@ def get_test_items(self, logger.debug(f"Retrieved {len(items)} test items") return items + def get_logs(self, + item_id: str, + level: Optional[str] = 'ERROR', + page_size: int = DEFAULT_PAGE_SIZE) -> List[Dict]: + """Get log entries for a test item, optionally filtered by level.""" + logger.debug(f"Fetching logs for item {item_id}") + + params = {'filter.eq.item': item_id, 'page.size': page_size} + if level: + params['filter.in.level'] = level + + query_string = '&'.join(f'{k}={v}' for k, v in params.items() if v is not None) + endpoint = f'/api/v1/{self.project}/log?{query_string}' + + data = self._get(endpoint) + logs = data.get('content', []) + + logger.debug(f"Retrieved {len(logs)} log entries for item {item_id}") + return logs + def get_test_item_by_id(self, item_id: str) -> Dict: """ Get a specific test item by ID. diff --git a/src/reportportal/rp_query.py b/src/reportportal/rp_query.py index 311af26..aeea8a2 100644 --- a/src/reportportal/rp_query.py +++ b/src/reportportal/rp_query.py @@ -437,6 +437,7 @@ def _extract_filter_options(options: Namespace) -> Dict[str, Any]: 'attribute_regex_filters': getattr(options, 'attribute_regex', None) if hasattr(options, 'attribute_regex') and options.attribute_regex else None, 'show_attributes': getattr(options, 'show_attributes', False), 'names_only': getattr(options, 'names_only', False), + 'show_logs': getattr(options, 'show_logs', False), 'limit': getattr(options, 'limit', None), } @@ -487,6 +488,99 @@ def _resolve_test_target(client: 'ReportPortalAPIClient', launch_id: str, target return parent_id +# ============================================================================= +# Log Display +# ============================================================================= + +MAX_LOG_MESSAGE_LENGTH = 1000 + + +def _truncate_message(message) -> str: + message = str(message) if message else '' + if len(message) > MAX_LOG_MESSAGE_LENGTH: + return message[:MAX_LOG_MESSAGE_LENGTH] + '...' + return message + + +def _fetch_error_logs(item_id: str, client: ReportPortalAPIClient) -> List[Dict]: + """Fetch ERROR-level logs for an item. Returns empty list on failure.""" + try: + return client.get_logs(str(item_id)) + except Exception as e: + logger.debug(f"Failed to fetch logs for item {item_id}: {e}") + return [] + + +def _print_logs(logs: List[Dict], indent: str = " ") -> None: + """Print truncated log messages with the given indentation.""" + for log_entry in logs: + message = _truncate_message(log_entry.get('message', '')) + for line in message.splitlines(): + print(f"{indent}{line}") + + +def _get_retry_items(item: Dict, client: ReportPortalAPIClient) -> List[Dict]: + """Fetch retry items (earlier attempts) for a test item.""" + item_id = item.get('id') + try: + detail = client.get_test_item_by_id(str(item_id)) + retries = detail.get('retries', []) + if not retries: + logger.debug(f"No retries found for item {item_id}") + return [] + + logger.debug(f"Found {len(retries)} retries for item {item_id}") + retry_items = [] + for retry in retries: + retry_id = retry.get('id') if isinstance(retry, dict) else retry + try: + retry_detail = client.get_test_item_by_id(str(retry_id)) + retry_items.append(retry_detail) + except Exception as e: + logger.debug(f"Failed to fetch retry item {retry_id}: {e}") + return retry_items + except Exception as e: + logger.debug(f"Failed to fetch item detail for {item_id}: {e}") + return [] + + +def _output_failed_logs(items: List[Dict], client: ReportPortalAPIClient) -> None: + """Fetch and display error logs for failed test items, including retries.""" + failed_steps = [ + item for item in items + if item.get('type') == utils.ITEM_TYPE_STEP and item.get('status') == utils.STATUS_FAILED + ] + + if not failed_steps: + print("\nNo failed tests found, nothing to show for --show-logs.") + return + + print("\n--- Failure Details ---\n") + + for item in failed_steps: + item_name = item.get('name', 'N/A') + retry_items = _get_retry_items(item, client) + + if retry_items: + chain = retry_items + [item] + print(f"{item_name} ({len(chain)} attempts):") + for i, attempt in enumerate(chain, 1): + status = attempt.get('status', 'N/A') + print(f" Attempt {i} ({status}):") + if status == utils.STATUS_FAILED: + logs = _fetch_error_logs(attempt.get('id'), client) + _print_logs(logs, indent=" ") + print() + else: + logs = _fetch_error_logs(item.get('id'), client) + print(f"{item_name}:") + if not logs: + print(" (no error logs found)") + else: + _print_logs(logs) + print() + + # ============================================================================= # Main Entry Point # ============================================================================= @@ -564,6 +658,18 @@ def run_query(options: Namespace) -> int: show_attributes=filter_opts['show_attributes'], names_only=filter_opts['names_only'] ) + + if filter_opts['show_logs'] and filter_opts['names_only']: + logger.warning("--show-logs is ignored when --names-only is used") + elif filter_opts['show_logs']: + filtered_items = apply_all_filters( + items, + name_regex=filter_opts['name_regex'], + attribute_filters=filter_opts['attribute_filters'], + attribute_regex_filters=filter_opts['attribute_regex_filters'], + ) + if filtered_items: + _output_failed_logs(filtered_items, client) else: # Query launches if filter_msg: diff --git a/tests/unit/test_rp_query_logs.py b/tests/unit/test_rp_query_logs.py new file mode 100644 index 0000000..1c28d1f --- /dev/null +++ b/tests/unit/test_rp_query_logs.py @@ -0,0 +1,299 @@ +"""Tests for query --show-logs functionality.""" + +import pytest +from argparse import Namespace +from unittest.mock import MagicMock, patch + +from reportportal.rp_query import _output_failed_logs, run_query + + +@pytest.fixture +def mock_client(): + """Create a mock ReportPortal API client.""" + return MagicMock() + + +@pytest.fixture +def items_with_failed(): + """Test items including a failed STEP.""" + return [ + { + 'id': 'item1', + 'name': 'test_login', + 'type': 'STEP', + 'status': 'PASSED', + }, + { + 'id': 'item2', + 'name': 'test_forbidden', + 'type': 'STEP', + 'status': 'FAILED', + }, + { + 'id': 'suite1', + 'name': 'Auth Suite', + 'type': 'SUITE', + 'status': 'FAILED', + }, + ] + + +@pytest.mark.unit +class TestOutputFailedLogs: + """Test _output_failed_logs display function.""" + + def test_fetches_logs_only_for_failed_steps(self, items_with_failed, mock_client, capsys): + mock_client.get_test_item_by_id.return_value = {'id': 'item2', 'name': 'test_forbidden'} + mock_client.get_logs.return_value = [ + {'message': 'assert 200 == 403', 'level': 'ERROR'} + ] + + _output_failed_logs(items_with_failed, mock_client) + + mock_client.get_logs.assert_called_once_with('item2') + + def test_prints_error_message(self, items_with_failed, mock_client, capsys): + mock_client.get_test_item_by_id.return_value = {'id': 'item2', 'name': 'test_forbidden'} + mock_client.get_logs.return_value = [ + {'message': 'E assert 200 == 403\nE + where 200 = result.status_code', 'level': 'ERROR'} + ] + + _output_failed_logs(items_with_failed, mock_client) + + output = capsys.readouterr().out + assert 'test_forbidden:' in output + assert 'assert 200 == 403' in output + assert 'where 200 = result.status_code' in output + + def test_shows_no_logs_message_when_empty(self, items_with_failed, mock_client, capsys): + mock_client.get_test_item_by_id.return_value = {'id': 'item2', 'name': 'test_forbidden'} + mock_client.get_logs.return_value = [] + + _output_failed_logs(items_with_failed, mock_client) + + output = capsys.readouterr().out + assert 'test_forbidden:' in output + assert 'no error logs found' in output + + def test_no_failures_message_when_no_failed_items(self, mock_client, capsys): + items = [ + {'id': 'item1', 'name': 'test_login', 'type': 'STEP', 'status': 'PASSED'}, + ] + + _output_failed_logs(items, mock_client) + + mock_client.get_logs.assert_not_called() + output = capsys.readouterr().out + assert 'No failed tests found' in output + + def test_truncates_long_messages(self, items_with_failed, mock_client, capsys): + long_message = 'x' * 2000 + mock_client.get_test_item_by_id.return_value = {'id': 'item2', 'name': 'test_forbidden'} + mock_client.get_logs.return_value = [ + {'message': long_message, 'level': 'ERROR'} + ] + + _output_failed_logs(items_with_failed, mock_client) + + output = capsys.readouterr().out + assert '...' in output + + def test_handles_log_api_error_gracefully(self, items_with_failed, mock_client, capsys): + mock_client.get_test_item_by_id.return_value = {'id': 'item2', 'name': 'test_forbidden'} + mock_client.get_logs.side_effect = Exception("API error") + + _output_failed_logs(items_with_failed, mock_client) + + output = capsys.readouterr().out + assert 'Failure Details' in output + assert 'no error logs found' in output + + def test_multiple_failed_items(self, mock_client, capsys): + items = [ + {'id': 'f1', 'name': 'test_a', 'type': 'STEP', 'status': 'FAILED'}, + {'id': 'f2', 'name': 'test_b', 'type': 'STEP', 'status': 'FAILED'}, + ] + mock_client.get_test_item_by_id.side_effect = [ + {'id': 'f1', 'name': 'test_a'}, + {'id': 'f2', 'name': 'test_b'}, + ] + mock_client.get_logs.side_effect = [ + [{'message': 'error A', 'level': 'ERROR'}], + [{'message': 'error B', 'level': 'ERROR'}], + ] + + _output_failed_logs(items, mock_client) + + output = capsys.readouterr().out + assert 'test_a:' in output + assert 'error A' in output + assert 'test_b:' in output + assert 'error B' in output + + +@pytest.mark.unit +class TestOutputFailedLogsWithRetries: + """Test _output_failed_logs with retry items from RP item detail.""" + + def test_retries_shown_as_attempts(self, mock_client, capsys): + items = [ + {'id': 'item1', 'name': 'test_flaky', 'type': 'STEP', 'status': 'FAILED'}, + ] + mock_client.get_test_item_by_id.side_effect = [ + # First call: detail for item1 (has retries) + {'id': 'item1', 'retries': [{'id': 'r1'}, {'id': 'r2'}, {'id': 'r3'}]}, + # Then fetching each retry item detail + {'id': 'r1', 'status': 'FAILED'}, + {'id': 'r2', 'status': 'FAILED'}, + {'id': 'r3', 'status': 'FAILED'}, + ] + mock_client.get_logs.side_effect = [ + [{'message': 'first try error', 'level': 'ERROR'}], + [{'message': 'second try error', 'level': 'ERROR'}], + [{'message': 'third try error', 'level': 'ERROR'}], + [{'message': 'final error', 'level': 'ERROR'}], + ] + + _output_failed_logs(items, mock_client) + + output = capsys.readouterr().out + assert 'test_flaky (4 attempts):' in output + assert 'Attempt 1' in output + assert 'first try error' in output + assert 'Attempt 2' in output + assert 'second try error' in output + assert 'Attempt 3' in output + assert 'third try error' in output + assert 'Attempt 4' in output + assert 'final error' in output + + def test_single_log_no_attempt_label(self, mock_client, capsys): + items = [ + {'id': 'item1', 'name': 'test_simple', 'type': 'STEP', 'status': 'FAILED'}, + ] + # No retries in item detail + mock_client.get_test_item_by_id.return_value = {'id': 'item1'} + mock_client.get_logs.return_value = [ + {'message': 'simple error', 'level': 'ERROR'}, + ] + + _output_failed_logs(items, mock_client) + + output = capsys.readouterr().out + assert 'test_simple:' in output + assert 'simple error' in output + assert 'attempts' not in output + assert 'Attempt' not in output + + def test_does_not_show_passed_items(self, mock_client, capsys): + items = [ + {'id': 'ok1', 'name': 'test_stable', 'type': 'STEP', 'status': 'PASSED'}, + ] + + _output_failed_logs(items, mock_client) + + output = capsys.readouterr().out + assert 'test_stable' not in output + + def test_retry_with_mixed_statuses(self, mock_client, capsys): + """Test retries where some passed and the final attempt failed.""" + items = [ + {'id': 'item1', 'name': 'test_mixed', 'type': 'STEP', 'status': 'FAILED'}, + ] + mock_client.get_test_item_by_id.side_effect = [ + {'id': 'item1', 'retries': [{'id': 'r1'}]}, + {'id': 'r1', 'status': 'PASSED'}, + ] + mock_client.get_logs.return_value = [ + {'message': 'final failure', 'level': 'ERROR'}, + ] + + _output_failed_logs(items, mock_client) + + output = capsys.readouterr().out + assert 'test_mixed (2 attempts):' in output + assert 'Attempt 1 (PASSED):' in output + assert 'Attempt 2 (FAILED):' in output + assert 'final failure' in output + + def test_item_detail_fetch_fails_gracefully(self, mock_client, capsys): + """If fetching item detail fails, fall back to simple log display.""" + items = [ + {'id': 'item1', 'name': 'test_detail_err', 'type': 'STEP', 'status': 'FAILED'}, + ] + mock_client.get_test_item_by_id.side_effect = Exception("API down") + mock_client.get_logs.return_value = [ + {'message': 'error msg', 'level': 'ERROR'}, + ] + + _output_failed_logs(items, mock_client) + + output = capsys.readouterr().out + assert 'test_detail_err:' in output + assert 'error msg' in output + assert 'Attempt' not in output + + +@pytest.mark.unit +class TestRunQueryShowLogsIntegration: + """Test that run_query applies local filters and names_only to show_logs.""" + + @pytest.fixture + def base_options(self): + return Namespace( + rp_url='http://rp.test', + rp_project='proj', + rp_token='token', + launch_id='123', + launch_name=None, + log_level='INFO', + status=None, + name=None, + name_regex=None, + attribute=None, + attribute_regex=None, + show_attributes=False, + names_only=False, + show_logs=True, + limit=None, + test_target=None, + ) + + @pytest.fixture + def all_items(self): + return [ + {'id': '1', 'name': 'test_alpha', 'type': 'STEP', 'status': 'FAILED'}, + {'id': '2', 'name': 'test_beta', 'type': 'STEP', 'status': 'FAILED'}, + ] + + @patch('reportportal.rp_query.ReportPortalAPIClient') + def test_show_logs_respects_name_regex(self, MockClient, base_options, all_items, capsys): + base_options.name_regex = 'test_alpha' + + client = MockClient.return_value + client.get_launch_by_id.return_value = { + 'id': '123', 'name': 'launch', 'number': 1, + 'startTime': 1000, 'statistics': {'executions': {}, 'defects': {}}, + } + client.get_test_items.return_value = all_items + client.get_test_item_by_id.return_value = {'id': '1', 'name': 'test_alpha'} + client.get_logs.return_value = [{'message': 'alpha error', 'level': 'ERROR'}] + + run_query(base_options) + + output = capsys.readouterr().out + assert 'alpha error' in output + assert 'test_beta' not in output.split('Failure Details')[-1] + + @patch('reportportal.rp_query.ReportPortalAPIClient') + def test_show_logs_suppressed_with_names_only(self, MockClient, base_options, all_items, capsys): + base_options.names_only = True + + client = MockClient.return_value + client.get_test_items.return_value = all_items + + run_query(base_options) + + output = capsys.readouterr().out + assert 'Failure Details' not in output + client.get_logs.assert_not_called()