Audit trail plugin

.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 }}

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"

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

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)

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]<missing>[/red]",
+                str(target_value) if target_value is not None else "[red]<missing>[/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()

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)