From 4c38f5b85e825f674813ee00877a6dd4c2a91f79 Mon Sep 17 00:00:00 2001 From: Marc Serrat Date: Fri, 27 Dec 2024 18:06:55 +0100 Subject: [PATCH] Audit trail plugin --- .github/workflows/pr-build-merge.yml | 6 +- pyproject.toml | 9 + sonar-project.properties | 6 +- swo/mpt/cli/plugins/audit_plugin/__init__.py | 0 swo/mpt/cli/plugins/audit_plugin/api.py | 43 ++++ swo/mpt/cli/plugins/audit_plugin/app.py | 138 +++++++++++ swo/mpt/cli/plugins/audit_plugin/utils.py | 75 ++++++ tests/audit_plugin/test_api.py | 136 +++++++++++ tests/audit_plugin/test_app.py | 226 +++++++++++++++++++ tests/audit_plugin/test_audit_utils.py | 128 +++++++++++ 10 files changed, 759 insertions(+), 8 deletions(-) create mode 100644 swo/mpt/cli/plugins/audit_plugin/__init__.py create mode 100644 swo/mpt/cli/plugins/audit_plugin/api.py create mode 100644 swo/mpt/cli/plugins/audit_plugin/app.py create mode 100644 swo/mpt/cli/plugins/audit_plugin/utils.py create mode 100644 tests/audit_plugin/test_api.py create mode 100644 tests/audit_plugin/test_app.py create mode 100644 tests/audit_plugin/test_audit_utils.py diff --git a/.github/workflows/pr-build-merge.yml b/.github/workflows/pr-build-merge.yml index 900b6ca..cc1c2ad 100644 --- a/.github/workflows/pr-build-merge.yml +++ b/.github/workflows/pr-build-merge.yml @@ -29,12 +29,8 @@ jobs: - name: 'Run validation & test' run: docker compose run --service-ports app_test - - name: 'Fix coverage paths for SonarCloud' - run: | - sed -i 's/\/swo\/swo\/mpt\/cli/\/github\/workspace\/swo\/mpt\/cli/g' coverage.xml - - name: 'Run SonarCloud Scan' - uses: SonarSource/sonarcloud-github-action@master + uses: SonarSource/sonarqube-scan-action@master env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }} diff --git a/pyproject.toml b/pyproject.toml index cc83153..7c30acf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,11 +71,17 @@ filterwarnings = [ [tool.coverage.run] branch = true +relative_files = true +source = ["swo"] [tool.coverage.report] exclude_also = [ "if __name__ == \"__main__\":", ] +include = [ + "swo/mpt/cli/**", + "swo/mpt/cli/plugins/audit_plugin/**", +] [tool.ruff] extend-exclude = [".vscode", ".devcontainer"] @@ -119,3 +125,6 @@ ignore_missing_imports = true [[tool.mypy.overrides]] module = "gunicorn.*" ignore_missing_imports = true + +[tool.poetry.plugins."swo.mpt.cli.plugins"] +"audit" = "swo.mpt.cli.plugins.audit_plugin.app:app" \ No newline at end of file diff --git a/sonar-project.properties b/sonar-project.properties index ef09de3..7445d92 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -3,9 +3,9 @@ sonar.organization=softwareone-mpt-github sonar.language=py -sonar.sources=./swo/mpt/cli -sonar.tests=./tests -sonar.inclusions=swo/mpt/cli/** +sonar.sources=swo +sonar.tests=tests +sonar.inclusions=swo/** sonar.exclusions=tests/** sonar.python.coverage.reportPaths=coverage.xml diff --git a/swo/mpt/cli/plugins/audit_plugin/__init__.py b/swo/mpt/cli/plugins/audit_plugin/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/swo/mpt/cli/plugins/audit_plugin/api.py b/swo/mpt/cli/plugins/audit_plugin/api.py new file mode 100644 index 0000000..4c24a3a --- /dev/null +++ b/swo/mpt/cli/plugins/audit_plugin/api.py @@ -0,0 +1,43 @@ +from typing import Any, Dict, List + +import typer +from swo.mpt.cli.core.console import console + + +def get_audit_trail(client: Any, record_id: str) -> Dict[str, Any]: + """Retrieve audit trail for a specific record.""" + try: + endpoint = f"/audit/records/{record_id}" + params = ( + "render()&select=object,actor,details,documents,request.api.geolocation" + ) + response = client.get(endpoint + "?" + params) + return response.json() + except Exception as e: + console.print(f"[red]Failed to retrieve audit trail for record {record_id}: {str(e)}[/red]") + raise typer.Exit(1) + + +def get_audit_records_by_object( + client: Any, + object_id: str, + limit: int = 10 +) -> List[Dict[str, Any]]: + """Retrieve all audit records for a specific object.""" + try: + endpoint = "/audit/records" + params = ( + "render()&select=object,actor,details,documents,request.api.geolocation" + f"&eq(object.id,{object_id})&order=-timestamp&limit={limit}" + ) + response = client.get(endpoint + "?" + params) + records = response.json().get('data', []) + if not records: + console.print(f"[red]No audit records found for object {object_id}[/red]") + raise typer.Exit(1) + return records + except Exception as e: + console.print( + f"[red]Failed to retrieve audit records for object {object_id}: {str(e)}[/red]" + ) + raise typer.Exit(1) diff --git a/swo/mpt/cli/plugins/audit_plugin/app.py b/swo/mpt/cli/plugins/audit_plugin/app.py new file mode 100644 index 0000000..e91f4cd --- /dev/null +++ b/swo/mpt/cli/plugins/audit_plugin/app.py @@ -0,0 +1,138 @@ +from typing import Any, Dict, Optional + +import typer +from rich.panel import Panel +from rich.table import Table +from swo.mpt.cli.core.accounts.app import get_active_account +from swo.mpt.cli.core.console import console +from swo.mpt.cli.core.mpt.client import client_from_account + +from .api import get_audit_records_by_object, get_audit_trail +from .utils import display_audit_records, flatten_dict, format_json_path + +app = typer.Typer(name="audit", help="Audit commands.") + + +def compare_audit_trails(source_trail: Dict[str, Any], target_trail: Dict[str, Any]) -> None: + """Compare two audit trails and display the differences.""" + # Ensure we're comparing the same object + source_object_id = source_trail.get('object', {}).get('id') + target_object_id = target_trail.get('object', {}).get('id') + + if source_object_id != target_object_id: + console.print("[red]Cannot compare different objects[/red]") + raise typer.Exit(1) + + # Create panel with basic information + panel = Panel( + f"Comparing audit trails...\n" + f"Object ID: {source_object_id}\n" + f"Source Timestamp: {source_trail.get('timestamp')}\n" + f"Target Timestamp: {target_trail.get('timestamp')}" + ) + console.print(panel) + + # Flatten both dictionaries for comparison + source_flat = flatten_dict(source_trail) + target_flat = flatten_dict(target_trail) + + # Create sets of all keys + all_keys = set(source_flat.keys()) | set(target_flat.keys()) + + # Create table for differences + table = Table(title="Audit Trail Differences") + table.add_column("Path", style="cyan") + table.add_column("Source Value", style="green") + table.add_column("Target Value", style="yellow") + + differences_found = False + + for key in sorted(all_keys): + source_value = source_flat.get(key) + target_value = target_flat.get(key) + + if source_value != target_value: + differences_found = True + formatted_path = format_json_path(key, source_trail, target_trail) + table.add_row( + formatted_path, + str(source_value) if source_value is not None else "[red][/red]", + str(target_value) if target_value is not None else "[red][/red]" + ) + + if differences_found: + console.print(table) + else: + console.print("\n[green]No differences found between the audit trails[/green]") + + +@app.command() +def diff_by_object_id( + object_id: str = typer.Argument( + ..., + help="Object ID to fetch and compare all audit records for" + ), + positions: Optional[str] = typer.Option( + None, + "--positions", + help="Positions to compare (e.g. '1,3')" + ), + limit: int = typer.Option(10, "--limit", help="Maximum number of audit records to retrieve") +): + """Compare audit trails for a specific object.""" + account = get_active_account() + client = client_from_account(account) + + records = get_audit_records_by_object(client, object_id, limit) + if len(records) < 2: + console.print("[red]Need at least 2 audit records to compare[/red]") + raise typer.Exit(1) + + display_audit_records(records) + + if positions: + try: + pos1, pos2 = map(int, positions.split(',')) + if not (1 <= pos1 <= len(records) and 1 <= pos2 <= len(records)): + raise ValueError + except ValueError: + msg = ( + "[red]Invalid positions. Please specify two numbers " + f"between 1 and {len(records)}[/red]" + ) + console.print(msg) + raise typer.Exit(1) + + source_trail = records[pos1 - 1] + target_trail = records[pos2 - 1] + else: + source_trail = records[0] + target_trail = records[1] + console.print( + "[yellow]No positions specified, comparing two most recent records (1,2)[/yellow]" + ) + + compare_audit_trails(source_trail, target_trail) + + +@app.command() +def diff_by_records_id( + source: str = typer.Argument(..., help="ID of the source audit record"), + target: str = typer.Argument(..., help="ID of the target audit record") +): + """Compare audit trails between two specific audit record IDs.""" + account = get_active_account() + client = client_from_account(account) + + source_trail = get_audit_trail(client, source) + target_trail = get_audit_trail(client, target) + + compare_audit_trails(source_trail, target_trail) + + +def main(): # pragma: no cover + app() + + +if __name__ == "__main__": # pragma: no cover + main() diff --git a/swo/mpt/cli/plugins/audit_plugin/utils.py b/swo/mpt/cli/plugins/audit_plugin/utils.py new file mode 100644 index 0000000..34d1ca4 --- /dev/null +++ b/swo/mpt/cli/plugins/audit_plugin/utils.py @@ -0,0 +1,75 @@ +from typing import Any, Dict + +from rich.table import Table +from swo.mpt.cli.core.console import console + + +def flatten_dict(d: Dict[str, Any], parent_key: str = '', sep: str = '.') -> Dict[str, Any]: + """Flatten a nested dictionary into a single level with dot notation keys.""" + items: list = [] + for k, v in d.items(): + new_key = f"{parent_key}{sep}{k}" if parent_key else k + if isinstance(v, dict): + items.extend(flatten_dict(v, new_key, sep=sep).items()) + elif isinstance(v, list): + for i, item in enumerate(v): + if isinstance(item, dict): + items.extend(flatten_dict(item, f"{new_key}[{i}]", sep=sep).items()) + else: + items.append((f"{new_key}[{i}]", item)) + else: + items.append((new_key, v)) + return dict(items) + + +def format_json_path(path: str, source_trail: Dict[str, Any], target_trail: Dict[str, Any]) -> str: + """Format JSON path with additional context from external IDs if available.""" + if '[' in path and ']' in path: + array_path, rest = path.split(']', 1) + base_path, index_str = array_path.split('[') + index = int(index_str) + + for trail in [source_trail, target_trail]: + try: + obj = trail + for part in base_path.split('.'): + obj = obj[part] + if isinstance(obj, list) and len(obj) > index: + if 'externalId' in obj[index]: + return f"{path} (externalId: {obj[index]['externalId']})" + except (KeyError, IndexError, TypeError): + continue + + return path + + +def display_audit_records(records: list) -> None: + """Display available audit records in a table format.""" + table = Table(title="Available Audit Records") + table.add_column("Position", style="cyan", no_wrap=True) + table.add_column("Timestamp", style="green", no_wrap=True) + table.add_column("Audit ID", style="bright_blue") + table.add_column("Actor", style="yellow") + table.add_column("Event", style="magenta") + table.add_column("Details", style="white") + + for idx, record in enumerate(records, 1): + timestamp = record.get('timestamp', 'N/A') + audit_id = record.get('id', 'N/A') + actor = record.get('actor', {}).get('name', 'N/A') + actor_account = record.get('actor', {}).get('account', {}).get('name', '') + if actor_account: + actor = f"{actor} ({actor_account})" + event = record.get('event', 'N/A') + details = record.get('details', 'N/A') + + table.add_row( + str(idx), + timestamp, + audit_id, + actor, + event.replace('platform.commerce.', ''), + details + ) + + console.print(table) diff --git a/tests/audit_plugin/test_api.py b/tests/audit_plugin/test_api.py new file mode 100644 index 0000000..fa967bb --- /dev/null +++ b/tests/audit_plugin/test_api.py @@ -0,0 +1,136 @@ +from unittest.mock import Mock + +import pytest +from swo.mpt.cli.plugins.audit_plugin.api import ( + get_audit_records_by_object, + get_audit_trail, +) +from typer import Exit + + +@pytest.fixture() +def mock_client(): + client = Mock() + client.get = Mock() + return client + + +@pytest.fixture() +def mock_response(): + response = Mock() + response.json = Mock() + return response + + +class TestGetAuditTrail: + def test_successful_retrieval(self, mock_client, mock_response): + # Setup + expected_data = { + "id": "audit123", + "object": {"id": "obj123"}, + "actor": {"name": "Test User"} + } + mock_response.json.return_value = expected_data + mock_client.get.return_value = mock_response + + # Execute + result = get_audit_trail(mock_client, "audit123") + + # Assert + assert result == expected_data + mock_client.get.assert_called_once() + called_endpoint = mock_client.get.call_args[0][0] + assert "/audit/records/audit123" in called_endpoint + assert "render()" in called_endpoint + assert "select=object,actor,details,documents,request.api.geolocation" in called_endpoint + + def test_api_error(self, mock_client): + # Setup + mock_client.get.side_effect = Exception("API Error") + + # Execute and Assert + with pytest.raises(Exit): + get_audit_trail(mock_client, "audit123") + mock_client.get.assert_called_once() + + def test_json_decode_error(self, mock_client, mock_response): + # Setup + mock_response.json.side_effect = ValueError("Invalid JSON") + mock_client.get.return_value = mock_response + + # Execute and Assert + with pytest.raises(Exit): + get_audit_trail(mock_client, "audit123") + + +class TestGetAuditRecordsByObject: + def test_successful_retrieval(self, mock_client, mock_response): + # Setup + expected_data = { + "data": [ + { + "id": "audit1", + "object": {"id": "obj123"}, + "timestamp": "2024-01-01T10:00:00Z" + }, + { + "id": "audit2", + "object": {"id": "obj123"}, + "timestamp": "2024-01-01T11:00:00Z" + } + ] + } + mock_response.json.return_value = expected_data + mock_client.get.return_value = mock_response + + # Execute + result = get_audit_records_by_object(mock_client, "obj123", limit=10) + + # Assert + assert result == expected_data["data"] + mock_client.get.assert_called_once() + called_endpoint = mock_client.get.call_args[0][0] + assert "/audit/records" in called_endpoint + assert "render()" in called_endpoint + assert "eq(object.id,obj123)" in called_endpoint + assert "limit=10" in called_endpoint + + def test_no_records_found(self, mock_client, mock_response): + # Setup + mock_response.json.return_value = {"data": []} + mock_client.get.return_value = mock_response + + # Execute and Assert + with pytest.raises(Exit): + get_audit_records_by_object(mock_client, "obj123") + mock_client.get.assert_called_once() + + def test_api_error(self, mock_client): + # Setup + mock_client.get.side_effect = Exception("API Error") + + # Execute and Assert + with pytest.raises(Exit): + get_audit_records_by_object(mock_client, "obj123") + mock_client.get.assert_called_once() + + def test_json_decode_error(self, mock_client, mock_response): + # Setup + mock_response.json.side_effect = ValueError("Invalid JSON") + mock_client.get.return_value = mock_response + + # Execute and Assert + with pytest.raises(Exit): + get_audit_records_by_object(mock_client, "obj123") + + def test_custom_limit(self, mock_client, mock_response): + # Setup + mock_response.json.return_value = {"data": [{"id": "audit1"}]} + mock_client.get.return_value = mock_response + + # Execute + get_audit_records_by_object(mock_client, "obj123", limit=5) + + # Assert + called_endpoint = mock_client.get.call_args[0][0] + assert "limit=5" in called_endpoint diff --git a/tests/audit_plugin/test_app.py b/tests/audit_plugin/test_app.py new file mode 100644 index 0000000..18ec8ce --- /dev/null +++ b/tests/audit_plugin/test_app.py @@ -0,0 +1,226 @@ +from unittest.mock import Mock, patch + +import pytest +from swo.mpt.cli.plugins.audit_plugin.app import app, compare_audit_trails +from typer.testing import CliRunner + +runner = CliRunner() + + +@pytest.fixture() +def mock_client(): + return Mock() + + +@pytest.fixture() +def mock_get_active_account(): + return Mock() + + +@pytest.fixture() +def mock_client_from_account(): + return Mock() + + +@pytest.fixture() +def sample_audit_records(): + return [ + { + "id": "audit1", + "timestamp": "2024-01-01T10:00:00Z", + "object": { + "id": "obj123", + "name": "Test Object", + "objectType": "TestType", + "value": "old_value" + } + }, + { + "id": "audit2", + "timestamp": "2024-01-02T10:00:00Z", + "object": { + "id": "obj123", + "name": "Test Object", + "objectType": "TestType", + "value": "new_value" + } + } + ] + + +class TestDiffByObjectId: + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_active_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.client_from_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_audit_records_by_object') + def test_successful_comparison( + self, + mock_get_records, + mock_client_from_account, + mock_get_active_account, + sample_audit_records + ): + # Setup + mock_get_records.return_value = sample_audit_records + + # Execute + result = runner.invoke(app, ['diff-by-object-id', 'obj123']) + + # Assert + assert result.exit_code == 0 + mock_get_records.assert_called_once_with( + mock_client_from_account.return_value, + 'obj123', + 10 + ) + assert "Comparing audit trails..." in result.stdout + assert "old_value" in result.stdout + assert "new_value" in result.stdout + + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_active_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.client_from_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_audit_records_by_object') + def test_with_custom_positions( + self, + mock_get_records, + mock_client_from_account, + mock_get_active_account, + sample_audit_records + ): + # Setup + mock_get_records.return_value = sample_audit_records + + # Execute + result = runner.invoke(app, [ + 'diff-by-object-id', + 'obj123', + '--positions', '2,1' + ]) + + # Assert + assert result.exit_code == 0 + assert "Comparing audit trails..." in result.stdout + assert "old_value" in result.stdout + assert "new_value" in result.stdout + + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_active_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.client_from_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_audit_records_by_object') + def test_insufficient_records( + self, + mock_get_records, + mock_client_from_account, + mock_get_active_account, + sample_audit_records + ): + # Setup + mock_get_records.return_value = [sample_audit_records[0]] + + # Execute + result = runner.invoke(app, ['diff-by-object-id', 'obj123']) + + # Assert + assert result.exit_code == 1 + assert "Need at least 2 audit records to compare" in result.stdout + + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_active_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.client_from_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_audit_records_by_object') + def test_invalid_positions( + self, + mock_get_records, + mock_client_from_account, + mock_get_active_account, + sample_audit_records + ): + # Setup + mock_get_records.return_value = sample_audit_records + + # Execute + result = runner.invoke(app, [ + 'diff-by-object-id', + 'obj123', + '--positions', '1,99' + ]) + + # Assert + assert result.exit_code == 1 + assert "Invalid positions" in result.stdout + + +class TestDiffByRecordsId: + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_active_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.client_from_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_audit_trail') + def test_successful_comparison( + self, + mock_get_trail, + mock_client_from_account, + mock_get_active_account, + sample_audit_records + ): + # Setup + mock_get_trail.side_effect = sample_audit_records + + # Execute + result = runner.invoke(app, [ + 'diff-by-records-id', + 'audit1', + 'audit2' + ]) + + # Assert + assert result.exit_code == 0 + assert "Comparing audit trails..." in result.stdout + assert "old_value" in result.stdout + assert "new_value" in result.stdout + + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_active_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.client_from_account') + @patch('swo.mpt.cli.plugins.audit_plugin.app.get_audit_trail') + def test_different_objects( + self, + mock_get_trail, + mock_client_from_account, + mock_get_active_account + ): + # Setup + mock_get_trail.side_effect = [ + { + "object": {"id": "obj1"}, + "timestamp": "2024-01-01T10:00:00Z" + }, + { + "object": {"id": "obj2"}, + "timestamp": "2024-01-01T11:00:00Z" + } + ] + + # Execute + result = runner.invoke(app, [ + 'diff-by-records-id', + 'audit1', + 'audit2' + ]) + + # Assert + assert result.exit_code == 1 + assert "Cannot compare different objects" in result.stdout + + +def test_compare_audit_trails_no_differences(): + source = { + "timestamp": "2024-01-01T10:00:00Z", + "object": { + "id": "obj123", + "name": "Test", + "value": "same" + } + } + target = source.copy() + + # Execute + with patch('swo.mpt.cli.plugins.audit_plugin.app.console.print') as mock_print: + compare_audit_trails(source, target) + + # Assert + mock_print.assert_any_call("\n[green]No differences found between the audit trails[/green]") diff --git a/tests/audit_plugin/test_audit_utils.py b/tests/audit_plugin/test_audit_utils.py new file mode 100644 index 0000000..57edb31 --- /dev/null +++ b/tests/audit_plugin/test_audit_utils.py @@ -0,0 +1,128 @@ +from swo.mpt.cli.plugins.audit_plugin.utils import ( + display_audit_records, + flatten_dict, + format_json_path, +) + + +class TestFlattenDict: + def test_simple_dict(self): + input_dict = {"a": 1, "b": 2} + result = flatten_dict(input_dict) + assert result == {"a": 1, "b": 2} + + def test_nested_dict(self): + input_dict = { + "a": { + "b": 1, + "c": { + "d": 2 + } + } + } + result = flatten_dict(input_dict) + assert result == { + "a.b": 1, + "a.c.d": 2 + } + + def test_dict_with_list(self): + input_dict = { + "a": [ + {"b": 1}, + {"c": 2} + ] + } + result = flatten_dict(input_dict) + assert result == { + "a[0].b": 1, + "a[1].c": 2 + } + + def test_dict_with_primitive_list(self): + input_dict = { + "a": [1, 2, 3] + } + result = flatten_dict(input_dict) + assert result == { + "a[0]": 1, + "a[1]": 2, + "a[2]": 3 + } + + +class TestFormatJsonPath: + def test_simple_path(self): + path = "object.name" + source = {"object": {"name": "test"}} + target = {"object": {"name": "test2"}} + result = format_json_path(path, source, target) + assert result == path + + def test_array_path_with_external_id(self): + path = "items[0].value" + source = { + "items": [ + {"value": "old", "externalId": "EXT123"} + ] + } + target = { + "items": [ + {"value": "new", "externalId": "EXT123"} + ] + } + result = format_json_path(path, source, target) + assert result == "items[0].value (externalId: EXT123)" + + def test_invalid_path(self): + path = "invalid.path[0]" + source = {"different": "structure"} + target = {"another": "structure"} + result = format_json_path(path, source, target) + assert result == path + + +class TestDisplayAuditRecords: + def test_display_records(self, capsys): + records = [ + { + "id": "audit1", + "timestamp": "2024-01-01T10:00:00Z", + "actor": { + "name": "Test User", + "account": {"name": "Test Account"} + }, + "event": "platform.commerce.create", + "details": "Created object" + } + ] + display_audit_records(records) + captured = capsys.readouterr() + output = captured.out + + # Check for key elements in the output + assert "Available Audit Records" in output + assert "2024-01-01T10:00:00Z" in output + assert "audit1" in output + assert "Test User" in output + assert "(Test" in output + assert "Account)" in output + assert "create" in output + assert "Created" in output + assert "object" in output + + def test_display_records_missing_fields(self, capsys): + records = [ + { + "id": "audit1", + "timestamp": "2024-01-01T10:00:00Z" + } + ] + display_audit_records(records) + captured = capsys.readouterr() + output = captured.out + + # Check individual elements instead of exact string matches + assert "audit1" in output + assert "2024-01-01T10:00:00Z" in output + assert "N/A" in output # For missing fields