diff --git a/README.md b/README.md index 6967866..39d4ec4 100644 --- a/README.md +++ b/README.md @@ -115,6 +115,36 @@ session.complete() # See examples/teleop_cancel.py for cancellation handling ``` +### CLI Usage + +Track USB copy operations from the command line. + +**For complete examples, see [examples/cli/usb_copy/](examples/cli/usb_copy/).** + +Quick example: + +```bash +# Start USB copy session +RUN_ID=$(airoa-lineage-usb-copy start \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash $(git rev-parse HEAD) \ + --repository-uri https://github.com/user/repo.git \ + --repository-tag v1.0.0 \ + --repository-branch main \ + --nominal-start-time "2025-11-26T00:00:00+00:00" \ + --nominal-end-time "2025-11-26T05:00:00+00:00") + +# Perform your USB copy operation +# ... + +# Complete session +airoa-lineage-usb-copy complete --run-id "$RUN_ID" +``` + +See [docs/cli-usage.md](docs/cli-usage.md) for full CLI reference. + ### Data Conversion Tracking Track data conversion pipelines (e.g., ROS Bag → LeRobot format): @@ -164,6 +194,14 @@ print("Conversion completed successfully") ## Documentation +- **[CLI Usage Guide](docs/cli-usage.md)** - Command-line interface reference + - Installation and quick start + - Configuration (file, environment variables, CLI arguments) + - Commands (start, complete, cancel) + - Output modes and exit codes + - Shell script integration examples + - Troubleshooting and common errors + - **[Architecture](docs/architecture.md)** - System design and core components - TeleopSession, ConversionSession, MarquezClient, Timestamp Utilities - Event flow and RUNNING events @@ -207,40 +245,7 @@ Shows how to track a batch ETL pipeline with input/output datasets. ## Development -### Setup - -```bash -# Clone the repository -git clone https://github.com/your-org/airoa-lineage.git -cd airoa-lineage - -# Install dependencies including dev tools -uv sync --group dev -``` - -### Run Tests - -```bash -make test -``` - -### Code Quality - -```bash -# Format code -make format - -# Run linter and type checker -make lint -``` - -### Build Package - -```bash -uv build -``` - -See the [Development Guide](docs/development.md) for comprehensive development documentation. +For developers who want to contribute to airoa-lineage, see the [Development Guide](docs/development.md). ## License diff --git a/docs/cli-usage.md b/docs/cli-usage.md new file mode 100644 index 0000000..c710f31 --- /dev/null +++ b/docs/cli-usage.md @@ -0,0 +1,700 @@ +# CLI Usage Guide + +This guide provides comprehensive documentation for the `airoa-lineage-usb-copy` command-line interface. + +## Table of Contents + +- [Installation](#installation) +- [Quick Start](#quick-start) +- [Configuration](#configuration) + - [Configuration File](#configuration-file) + - [Environment Variables](#environment-variables) + - [Priority Order](#priority-order) +- [Commands](#commands) + - [start](#start-command) + - [complete](#complete-command) + - [cancel](#cancel-command) +- [Global Options](#global-options) +- [Output Modes](#output-modes) +- [Exit Codes](#exit-codes) +- [Examples](#examples) + - [Basic Workflow](#basic-workflow) + - [Using Configuration File](#using-configuration-file) + - [Using Environment Variables](#using-environment-variables) + - [Shell Script Integration](#shell-script-integration) + - [Error Handling](#error-handling) +- [Troubleshooting](#troubleshooting) + +--- + +## Installation + +After installing the `airoa-lineage` package: + +```bash +# Using uv (recommended) +uv pip install -e . + +# Or using pip +pip install -e . +``` + +The `airoa-lineage-usb-copy` command will be available in your PATH. + +--- + +## Quick Start + +```bash +# 1. Start a USB copy session +RUN_ID=$(airoa-lineage-usb-copy start \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash $(git rev-parse HEAD) \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main \ + --nominal-start-time "2025-11-26T00:00:00+00:00" \ + --nominal-end-time "2025-11-26T05:00:00+00:00") + +# 2. Perform your USB copy operation +# (your data copy processing here) + +# 3. Complete the session +airoa-lineage-usb-copy complete --run-id "$RUN_ID" +``` + +--- + +## Configuration + +The CLI supports three configuration sources with the following priority: + +**CLI arguments > Environment variables > Configuration file > Defaults** + +### Configuration File + +Default path: `~/.config/airoa-lineage/config.json` + +You can override this with the `--config` option or by setting `XDG_CONFIG_HOME`. + +**Example config.json:** + +```json +{ + "marquez_url": "http://localhost:9000", + "namespace": "airoa_production", + "facet_prefix": "airoa", + "job_name": "usb-data-copy", + "common_facet": { + "robotId": "hsr001", + "location": "weblab", + "repositoryHash": "df110d5a8e3b9c2f1a7d6e4f5c3a2b1", + "repositoryUri": "https://github.com/AIRoA/airoa-lineage.git", + "repositoryTag": "v1.0.0", + "repositoryBranch": "main" + } +} +``` + +**Configuration Schema:** + +| Field | Type | Required | Default | Description | +|-------|------|----------|---------|-------------| +| `marquez_url` | string | No | `http://localhost:9000` | Marquez server URL | +| `namespace` | string | **Yes** | - | OpenLineage namespace | +| `job_name` | string | No | `usb-data-copy` | Job identifier | +| `facet_prefix` | string | No | `""` (empty) | Prefix for custom facets | +| `common_facet` | object | **Yes** | - | CommonRunFacet fields | + +**CommonRunFacet Fields:** + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `robotId` | string | **Yes** | Robot identifier (e.g., `hsr001`) | +| `location` | string | **Yes** | Location identifier (e.g., `weblab`) | +| `repositoryHash` | string | **Yes** | Git commit hash | +| `repositoryUri` | string | **Yes** | Repository URI | +| `repositoryTag` | string | **Yes** | Git tag (e.g., `v1.0.0`) | +| `repositoryBranch` | string | **Yes** | Git branch (e.g., `main`) | + +### Environment Variables + +All configuration values can be set via environment variables: + +```bash +# Top-level configuration +export AIROA_MARQUEZ_URL=http://localhost:9000 +export AIROA_NAMESPACE=airoa_production +export AIROA_FACET_PREFIX=airoa +export AIROA_JOB_NAME=usb-data-copy + +# CommonRunFacet fields +export AIROA_ROBOT_ID=hsr001 +export AIROA_LOCATION=weblab +export AIROA_REPOSITORY_HASH=$(git rev-parse HEAD) +export AIROA_REPOSITORY_URI=https://github.com/AIRoA/airoa-lineage.git +export AIROA_REPOSITORY_TAG=$(git describe --tags) +export AIROA_REPOSITORY_BRANCH=$(git rev-parse --abbrev-ref HEAD) +``` + +**Note:** `MARQUEZ_URL` is also supported for backward compatibility, but `AIROA_MARQUEZ_URL` takes precedence. + +### Priority Order + +When the same value is specified in multiple places: + +1. **CLI arguments** (highest priority) +2. **Environment variables** +3. **Configuration file** +4. **Default values** (lowest priority) + +**Example:** + +```bash +# config.json has: "namespace": "dev" +# Environment has: AIROA_NAMESPACE=staging + +# This uses CLI argument (production) +airoa-lineage-usb-copy start --namespace production + +# This uses environment variable (staging) +airoa-lineage-usb-copy start + +# If no env var, uses config file (dev) +unset AIROA_NAMESPACE +airoa-lineage-usb-copy start +``` + +--- + +## Commands + +### start Command + +Start a USB copy session and emit a START event to Marquez. + +**Synopsis:** + +```bash +airoa-lineage-usb-copy start [OPTIONS] +``` + +**Options:** + +| Option | Type | Required | Description | +|--------|------|----------|-------------| +| `--namespace` | string | Yes* | OpenLineage namespace | +| `--robot-id` | string | Yes* | Robot identifier | +| `--location` | string | Yes* | Location identifier | +| `--repository-hash` | string | Yes* | Git commit hash | +| `--repository-uri` | string | Yes* | Repository URI | +| `--repository-tag` | string | Yes* | Git tag | +| `--repository-branch` | string | Yes* | Git branch | +| `--nominal-start-time` | string | No | Nominal start time (ISO 8601) | +| `--nominal-end-time` | string | No | Nominal end time (ISO 8601) | +| `--job-name` | string | No | Job name (default: `usb-data-copy`) | +| `--marquez-url` | string | No | Marquez server URL | +| `--facet-prefix` | string | No | Facet prefix | +| `--dry-run` | flag | No | Show what would be done without executing | + +\* Required unless provided via config file or environment variables + +**Output:** + +- **Default:** Prints the `run_id` (UUID) to stdout +- **JSON mode:** `{"run_id": "...", "status": "started"}` +- **Verbose mode:** Additional messages to stderr, run_id to stdout +- **Dry-run mode:** Shows configuration without emitting events + +**Examples:** + +```bash +# Basic usage +RUN_ID=$(airoa-lineage-usb-copy start \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash $(git rev-parse HEAD) \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main) + +# With nominal times (for batch processing) +RUN_ID=$(airoa-lineage-usb-copy start \ + --nominal-start-time "2025-11-26T00:00:00+00:00" \ + --nominal-end-time "2025-11-26T05:00:00+00:00") + +# Dry-run mode (verify configuration) +airoa-lineage-usb-copy start --dry-run + +# JSON output (for programmatic parsing) +airoa-lineage-usb-copy start --json +``` + +--- + +### complete Command + +Complete a USB copy session and emit a COMPLETE event to Marquez. + +**Synopsis:** + +```bash +airoa-lineage-usb-copy complete --run-id RUN_ID [OPTIONS] +``` + +**Options:** + +| Option | Type | Required | Description | +|--------|------|----------|-------------| +| `--run-id` | string | **Yes** | Session run ID (from start command) | +| `--namespace` | string | Yes* | OpenLineage namespace | +| `--job-name` | string | No | Job name | +| `--marquez-url` | string | No | Marquez server URL | +| `--facet-prefix` | string | No | Facet prefix | +| `--robot-id` | string | Yes* | Robot identifier | +| `--location` | string | Yes* | Location identifier | +| `--repository-hash` | string | Yes* | Git commit hash | +| `--repository-uri` | string | Yes* | Repository URI | +| `--repository-tag` | string | Yes* | Git tag | +| `--repository-branch` | string | Yes* | Git branch | + +\* Required unless provided via config file or environment variables + +**Note:** CommonRunFacet values provided during `complete` do not need to match the `start` values. This allows for flexible scenarios (e.g., multi-robot collaboration). However, ensure consistency when tracking single-robot operations. + +**Output:** + +- **Default:** `Session completed successfully` +- **JSON mode:** `{"run_id": "...", "status": "completed"}` +- **Quiet mode:** No output + +**Examples:** + +```bash +# Basic usage (CommonRunFacet from config file or environment variables) +airoa-lineage-usb-copy complete --run-id "$RUN_ID" + +# With namespace override +airoa-lineage-usb-copy complete --run-id "$RUN_ID" --namespace production + +# Override specific CommonRunFacet fields +airoa-lineage-usb-copy complete \ + --run-id "$RUN_ID" \ + --robot-id hsr002 \ + --location lab2 \ + --repository-hash $(git rev-parse HEAD) + +# Specify all arguments (no config file needed) +airoa-lineage-usb-copy complete \ + --run-id "$RUN_ID" \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash abc123 \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main + +# JSON output +airoa-lineage-usb-copy complete --run-id "$RUN_ID" --json + +# Quiet mode (no output) +airoa-lineage-usb-copy complete --run-id "$RUN_ID" --quiet +``` + +--- + +### cancel Command + +Cancel a USB copy session and emit an ABORT event to Marquez. + +**Synopsis:** + +```bash +airoa-lineage-usb-copy cancel --run-id RUN_ID [OPTIONS] +``` + +**Options:** + +| Option | Type | Required | Description | +|--------|------|----------|-------------| +| `--run-id` | string | **Yes** | Session run ID (from start command) | +| `--namespace` | string | Yes* | OpenLineage namespace | +| `--job-name` | string | No | Job name | +| `--marquez-url` | string | No | Marquez server URL | +| `--facet-prefix` | string | No | Facet prefix | +| `--robot-id` | string | Yes* | Robot identifier | +| `--location` | string | Yes* | Location identifier | +| `--repository-hash` | string | Yes* | Git commit hash | +| `--repository-uri` | string | Yes* | Repository URI | +| `--repository-tag` | string | Yes* | Git tag | +| `--repository-branch` | string | Yes* | Git branch | + +\* Required unless provided via config file or environment variables + +**Note:** CommonRunFacet values provided during `cancel` do not need to match the `start` values. This allows for flexible scenarios (e.g., multi-robot collaboration). However, ensure consistency when tracking single-robot operations. + +**Output:** + +- **Default:** `Session cancelled successfully` +- **JSON mode:** `{"run_id": "...", "status": "cancelled"}` +- **Quiet mode:** No output + +**Examples:** + +```bash +# Basic usage (CommonRunFacet from config file or environment variables) +airoa-lineage-usb-copy cancel --run-id "$RUN_ID" + +# Override specific CommonRunFacet fields +airoa-lineage-usb-copy cancel \ + --run-id "$RUN_ID" \ + --robot-id hsr002 \ + --location lab2 + +# Specify all arguments (no config file needed) +airoa-lineage-usb-copy cancel \ + --run-id "$RUN_ID" \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash abc123 \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main + +# JSON output +airoa-lineage-usb-copy cancel --run-id "$RUN_ID" --json + +# Quiet mode (no output) +airoa-lineage-usb-copy cancel --run-id "$RUN_ID" --quiet +``` + +--- + +## Global Options + +These options apply to all commands: + +| Option | Short | Description | +|--------|-------|-------------| +| `--config PATH` | - | Path to config file (default: `~/.config/airoa-lineage/config.json`) | +| `--verbose` | `-v` | Enable verbose output (messages to stderr) | +| `--quiet` | `-q` | Suppress output except errors | +| `--json` | - | Output in JSON format | +| `--version` | - | Show version and exit | +| `--help` | `-h` | Show help message and exit | + +**Examples:** + +```bash +# Show version +airoa-lineage-usb-copy --version + +# Show help +airoa-lineage-usb-copy --help +airoa-lineage-usb-copy start --help + +# Use custom config file +airoa-lineage-usb-copy --config /path/to/config.json start + +# Verbose mode +airoa-lineage-usb-copy -v start + +# JSON output +airoa-lineage-usb-copy --json start +``` + +--- + +## Output Modes + +### Default Mode + +- **start:** Prints only `run_id` to stdout (for piping) +- **complete/cancel:** Prints success message to stdout + +```bash +RUN_ID=$(airoa-lineage-usb-copy start) +# Output: 550e8400-e29b-41d4-a716-446655440000 +``` + +### Verbose Mode (`-v`, `--verbose`) + +- All messages go to **stderr** +- `run_id` goes to **stdout** (for piping) +- Includes configuration details and event status + +```bash +airoa-lineage-usb-copy -v start +# stderr: Loading config from ~/.config/airoa-lineage/config.json +# stderr: Connecting to Marquez at http://localhost:9000 +# stderr: Session started successfully +# stdout: 550e8400-e29b-41d4-a716-446655440000 +``` + +### Quiet Mode (`-q`, `--quiet`) + +- **start:** Prints only `run_id` to stdout +- **complete/cancel:** No output (silent) +- Errors still go to stderr + +```bash +airoa-lineage-usb-copy -q complete --run-id "$RUN_ID" +# (no output) +``` + +### JSON Mode (`--json`) + +- All output in JSON format +- Suitable for programmatic parsing + +```bash +airoa-lineage-usb-copy --json start +# {"run_id": "550e8400-e29b-41d4-a716-446655440000", "status": "started"} + +airoa-lineage-usb-copy --json complete --run-id "$RUN_ID" +# {"run_id": "550e8400-e29b-41d4-a716-446655440000", "status": "completed"} +``` + +--- + +## Exit Codes + +The CLI uses standard exit codes to indicate success or failure: + +| Code | Meaning | Description | +|------|---------|-------------| +| `0` | Success | Command completed successfully | +| `1` | General error | Session error, runtime error, or unknown error | +| `2` | Connection error | Failed to connect to Marquez server | +| `3` | Configuration error | Missing or invalid configuration | + +**Example:** + +```bash +airoa-lineage-usb-copy start +EXIT_CODE=$? + +if [ $EXIT_CODE -eq 0 ]; then + echo "Success" +elif [ $EXIT_CODE -eq 3 ]; then + echo "Configuration error - check your config file" +elif [ $EXIT_CODE -eq 2 ]; then + echo "Connection error - is Marquez running?" +else + echo "Unknown error" +fi +``` + +--- + +## Examples + +For complete working examples, see [examples/cli/usb_copy/](../examples/cli/usb_copy/). + +### Available Examples + +The following executable shell scripts demonstrate different usage patterns: + +1. **[basic_workflow.sh](../examples/cli/usb_copy/basic_workflow.sh)** - Simple start → complete workflow + - All configuration via CLI arguments + - Ideal for quick testing and learning + +2. **[with_config_file.sh](../examples/cli/usb_copy/with_config_file.sh)** - Configuration file usage + - Loads settings from `~/.config/airoa-lineage/config.json` + - Minimal CLI arguments required + - Includes dry-run mode for testing + +3. **[with_env_vars.sh](../examples/cli/usb_copy/with_env_vars.sh)** - Environment variables + - Configuration via `AIROA_*` environment variables + - Dynamically retrieves repository info from git + +4. **[error_handling.sh](../examples/cli/usb_copy/error_handling.sh)** - Error handling with trap + - Automatic session cancellation on error + - Cleanup function with proper exit codes + +5. **[retry_logic.sh](../examples/cli/usb_copy/retry_logic.sh)** - Connection retry logic + - Automatic retry on connection errors + - Maximum 3 attempts with exponential backoff + +See [examples/cli/usb_copy/README.md](../examples/cli/usb_copy/README.md) for detailed documentation on each example, including prerequisites, usage instructions, and integration patterns. + +--- + +## Troubleshooting + +### Common Errors + +#### 1. "Missing required CommonRunFacet fields" + +**Error message:** +``` +Configuration error: Missing required CommonRunFacet fields: robotId, location, repositoryHash, repositoryUri, repositoryTag, repositoryBranch. +Please provide them in config file, environment variables, or CLI arguments. +``` + +**Solution:** + +Provide all required fields via one of these methods: + +```bash +# Option 1: CLI arguments +airoa-lineage-usb-copy start \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash $(git rev-parse HEAD) \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main + +# Option 2: Environment variables +export AIROA_ROBOT_ID=hsr001 +export AIROA_LOCATION=weblab +export AIROA_REPOSITORY_HASH=$(git rev-parse HEAD) +export AIROA_REPOSITORY_URI=https://github.com/AIRoA/airoa-lineage.git +export AIROA_REPOSITORY_TAG=v1.0.0 +export AIROA_REPOSITORY_BRANCH=main + +# Option 3: Config file +cat > ~/.config/airoa-lineage/config.json < ~/.config/airoa-lineage/config.json +``` + +#### 3. "Invalid run_id format" + +**Error message:** +``` +Configuration error: Invalid run_id format: not-a-uuid. Must be a valid UUID. +``` + +**Solution:** + +The `run_id` must be a valid UUID (36 characters with hyphens). Get the correct run_id from the `start` command: + +```bash +# Correct usage +RUN_ID=$(airoa-lineage-usb-copy start) +airoa-lineage-usb-copy complete --run-id "$RUN_ID" +``` + +#### 4. "Cannot resume session. Session already completed." + +**Error message:** +``` +Session error: Cannot resume session 550e8400-e29b-41d4-a716-446655440000. Session already completed. +``` + +**Cause:** You're trying to complete or cancel a session that was already completed. + +**Solution:** Each session can only be completed or cancelled once. Start a new session if needed. + +#### 5. Connection refused / Marquez server not running + +**Error message:** +``` +Connection error: Failed to connect to Marquez at http://localhost:9000 +``` + +**Solution:** + +Start the Marquez server: + +```bash +# Using Docker +docker run -d -p 3000:3000 -p 9000:9000 marquezproject/marquez:latest + +# Verify it's running +curl http://localhost:9000/api/v1/namespaces +``` + +### Debugging Tips + +#### 1. Use verbose mode + +```bash +airoa-lineage-usb-copy -v start +``` + +This shows: +- Config file path being used +- Loaded configuration values +- Marquez connection details +- Event emission status + +#### 2. Use dry-run mode + +```bash +airoa-lineage-usb-copy start --dry-run +``` + +This shows what would be sent to Marquez without actually sending it. + +#### 3. Check configuration loading + +```bash +# Test with explicit config file +airoa-lineage-usb-copy --config ~/.config/airoa-lineage/config.json start --dry-run + +# Verify environment variables +env | grep AIROA_ +``` + +#### 4. Verify Marquez connection + +```bash +# Test Marquez API +curl http://localhost:9000/api/v1/namespaces + +# Check Marquez logs +docker logs +``` + +#### 5. Check JSON syntax + +```bash +# Validate config file JSON +cat ~/.config/airoa-lineage/config.json | python -m json.tool +``` + +### Getting Help + +- CLI help: `airoa-lineage-usb-copy --help` +- Command help: `airoa-lineage-usb-copy start --help` +- Version: `airoa-lineage-usb-copy --version` +- GitHub Issues: https://github.com/AIRoA/airoa-lineage/issues +- OpenLineage docs: https://openlineage.io/docs +- Marquez docs: https://marquezproject.github.io/marquez/ diff --git a/examples/cli/README.md b/examples/cli/README.md new file mode 100644 index 0000000..5ab87fe --- /dev/null +++ b/examples/cli/README.md @@ -0,0 +1,82 @@ +# CLI Examples + +This directory contains command-line interface examples for airoa-lineage. + +## Available CLI Tools + +### USB Copy Session + +Track USB data copy operations from the command line. + +**Location**: [usb_copy/](usb_copy/) + +**Examples**: +- Basic workflow (start → complete) +- Configuration file usage +- Environment variable configuration +- Error handling with automatic cancellation +- Connection retry logic + +### Wasabi Upload Session *(Coming Soon)* + +Track Wasabi cloud upload operations from the command line. + +**Status**: Planned for future release + +## Common Patterns + +All CLI tools in this project follow these common patterns: + +### Configuration Methods + +Configuration can be provided in three ways (in order of priority): + +1. **CLI Arguments** - Highest priority, overrides all other sources +2. **Environment Variables** - Second priority, prefixed with `AIROA_` +3. **Configuration File** - Lowest priority, located at `~/.config/airoa-lineage/config.json` + +### Session Lifecycle + +All session-based CLI tools follow this lifecycle: + +1. **START** - Initiate a new session, returns a unique `RUN_ID` +2. **Operation** - Perform your data operation (copy, upload, etc.) +3. **COMPLETE** or **CANCEL** - Mark the session as finished (success or failure) + +Example: + +```bash +# Start a session +RUN_ID=$(airoa-lineage-usb-copy start [options]) + +# Perform your operation +# ... your data operation logic ... + +# Complete the session +airoa-lineage-usb-copy complete --run-id "$RUN_ID" + +# Or cancel if something went wrong +airoa-lineage-usb-copy cancel --run-id "$RUN_ID" +``` + +## Getting Started + +1. Install the package: + ```bash + pip install airoa-lineage + ``` + +2. Choose a CLI tool and navigate to its directory + +3. Review the available examples + +4. Start with the basic workflow example to understand the fundamentals + +5. Explore advanced examples for production use cases + +## Documentation + +For detailed API reference and configuration options, see: + +- [CLI Usage Documentation](../../docs/cli-usage.md) +- [Architecture Documentation](../../docs/architecture.md) diff --git a/examples/cli/usb_copy/README.md b/examples/cli/usb_copy/README.md new file mode 100644 index 0000000..d3fab4b --- /dev/null +++ b/examples/cli/usb_copy/README.md @@ -0,0 +1,219 @@ +# USB Copy CLI Examples + +Command-line interface examples for tracking USB data copy operations with OpenLineage. + +## Prerequisites + +### Installation + +Install the airoa-lineage package: + +```bash +pip install airoa-lineage +``` + +Verify installation: + +```bash +airoa-lineage-usb-copy --version +``` + +### Marquez Server (Optional) + +A Marquez server is required for actual event tracking. However, you can test the CLI using `--dry-run` mode without a server. + +**To start a local Marquez server with Docker**: + +```bash +docker run -p 3000:3000 -p 9000:9000 marquezproject/marquez:latest +``` + +Access the Marquez UI at: http://localhost:3000 + +## Available Examples + +### 1. [basic_workflow.sh](basic_workflow.sh) + +**Purpose**: Demonstrates the simplest start → complete flow + +**Key Features**: +- All configuration via CLI arguments +- Clear step-by-step workflow +- Ideal for quick testing and learning + +**Usage**: +```bash +chmod +x basic_workflow.sh +./basic_workflow.sh +``` + +### 2. [with_config_file.sh](with_config_file.sh) + +**Purpose**: Use a configuration file to minimize CLI arguments + +**Key Features**: +- Loads settings from `~/.config/airoa-lineage/config.json` +- Minimal CLI arguments required +- Includes dry-run mode for testing configuration + +**Usage**: +```bash +# Set up configuration file first (see config.example.json) +mkdir -p ~/.config/airoa-lineage +cp config.example.json ~/.config/airoa-lineage/config.json +# Edit config.json with your settings + +chmod +x with_config_file.sh +./with_config_file.sh +``` + +### 3. [with_env_vars.sh](with_env_vars.sh) + +**Purpose**: Use environment variables for configuration + +**Key Features**: +- Configuration via `AIROA_*` environment variables +- Dynamically retrieves repository info from git commands +- Includes fallback handling for missing git tags + +**Usage**: +```bash +chmod +x with_env_vars.sh +./with_env_vars.sh +``` + +### 4. [error_handling.sh](error_handling.sh) + +**Purpose**: Demonstrate proper error handling with automatic cancellation + +**Key Features**: +- Uses `trap` for cleanup on error +- Automatically cancels sessions on failure +- Proper exit code handling + +**Usage**: +```bash +chmod +x error_handling.sh +./error_handling.sh +``` + +### 5. [retry_logic.sh](retry_logic.sh) + +**Purpose**: Handle connection errors with automatic retry + +**Key Features**: +- Automatic retry on connection errors (EXIT_CODE=2) +- Maximum 3 retry attempts with exponential backoff +- Useful for unreliable network conditions + +**Usage**: +```bash +chmod +x retry_logic.sh +./retry_logic.sh +``` + +## Configuration + +### Configuration File + +See [config.example.json](config.example.json) for a complete configuration file example. + +**Location**: `~/.config/airoa-lineage/config.json` + +**Required Fields**: +- `namespace` - OpenLineage namespace +- `marquez_url` - Marquez server URL +- `job_name` - Job name for tracking +- `common_facet` - Robot and repository metadata + - `robotId` - Robot identifier + - `location` - Robot location + - `repositoryHash` - Git commit hash + - `repositoryUri` - Git repository URL + - `repositoryTag` - Git tag + - `repositoryBranch` - Git branch + +### Environment Variables + +All configuration file fields can be set via environment variables with the `AIROA_` prefix: + +- `AIROA_NAMESPACE` +- `AIROA_MARQUEZ_URL` +- `AIROA_JOB_NAME` +- `AIROA_FACET_PREFIX` +- `AIROA_ROBOT_ID` +- `AIROA_LOCATION` +- `AIROA_REPOSITORY_HASH` +- `AIROA_REPOSITORY_URI` +- `AIROA_REPOSITORY_TAG` +- `AIROA_REPOSITORY_BRANCH` + +## Running Examples + +### Testing with Dry-Run Mode + +All examples can be tested without a Marquez server using `--dry-run`: + +```bash +airoa-lineage-usb-copy start \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash $(git rev-parse HEAD) \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main \ + --nominal-start-time "2025-11-26T00:00:00+00:00" \ + --nominal-end-time "2025-11-26T05:00:00+00:00" \ + --dry-run +``` + +This will display the configuration without sending events to Marquez. + +### Integration with Your Scripts + +To integrate USB copy tracking into your existing scripts: + +1. **Start a session** before your USB copy operation: + ```bash + RUN_ID=$(airoa-lineage-usb-copy start [options]) + ``` + +2. **Perform your USB copy operation**: + ```bash + # Your actual USB copy logic (rsync, cp, etc.) + rsync -av /source/path /destination/path + ``` + +3. **Complete or cancel** the session: + ```bash + # On success: + airoa-lineage-usb-copy complete --run-id "$RUN_ID" + + # On failure: + airoa-lineage-usb-copy cancel --run-id "$RUN_ID" + ``` + +## Troubleshooting + +### Common Issues + +**Error: Connection refused** +- Ensure Marquez server is running on the configured URL +- Check `marquez_url` in your configuration +- Try using `--dry-run` mode to test without server connection + +**Error: Invalid configuration** +- Verify all required fields are set in config file or environment variables +- Use `--verbose` flag to see detailed configuration loading +- Check configuration file syntax with a JSON validator + +**Error: Invalid RUN_ID format** +- Ensure RUN_ID is a valid UUID +- Verify the `start` command completed successfully +- Check that you're passing RUN_ID correctly to `complete`/`cancel` + +## Next Steps + +- Review the [CLI Usage Documentation](../../../docs/cli-usage.md) for complete API reference +- Learn about [OpenLineage concepts](https://openlineage.io/docs/) +- Explore the [Marquez UI](http://localhost:3000) to visualize your data lineage diff --git a/examples/cli/usb_copy/basic_workflow.sh b/examples/cli/usb_copy/basic_workflow.sh new file mode 100755 index 0000000..1904729 --- /dev/null +++ b/examples/cli/usb_copy/basic_workflow.sh @@ -0,0 +1,47 @@ +#!/bin/bash + +################################################################################ +# Basic USB Copy Session Workflow +# +# Purpose: +# Demonstrates the simplest start → complete flow for USB copy operations. +# All configuration is provided via CLI arguments. +# +# Usage: +# ./basic_workflow.sh +# +# Prerequisites: +# - airoa-lineage package installed (pip install airoa-lineage) +# - Marquez server running (optional for dry-run mode) +################################################################################ + +set -e # Exit immediately if a command exits with a non-zero status + +# 1. Start a USB copy session +# The start command returns a RUN_ID which uniquely identifies this session +echo "Starting USB copy session..." + +RUN_ID=$(airoa-lineage-usb-copy start \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash $(git rev-parse HEAD) \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main \ + --nominal-start-time "2025-11-26T00:00:00+00:00" \ + --nominal-end-time "2025-11-26T05:00:00+00:00") + +echo "Started USB copy session with RUN_ID: $RUN_ID" + +# 2. Perform your USB copy operation +# Replace this placeholder with your actual USB copy logic +echo "Performing USB copy operation... (this is a placeholder)" +# Example: rsync -av /source/path /destination/path + +# 3. Complete the session +# Mark the session as successfully completed +echo "Completing USB copy session..." +airoa-lineage-usb-copy complete --run-id "$RUN_ID" + +echo "USB copy session completed successfully!" diff --git a/examples/cli/usb_copy/config.example.json b/examples/cli/usb_copy/config.example.json new file mode 100644 index 0000000..ad01f10 --- /dev/null +++ b/examples/cli/usb_copy/config.example.json @@ -0,0 +1,14 @@ +{ + "namespace": "airoa_production", + "marquez_url": "http://localhost:9000", + "job_name": "usb-data-copy", + "facet_prefix": "airoa", + "common_facet": { + "robotId": "hsr001", + "location": "weblab", + "repositoryHash": "df110d5b8c5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a", + "repositoryUri": "https://github.com/AIRoA/airoa-lineage.git", + "repositoryTag": "v1.0.0", + "repositoryBranch": "main" + } +} diff --git a/examples/cli/usb_copy/error_handling.sh b/examples/cli/usb_copy/error_handling.sh new file mode 100755 index 0000000..1e049fb --- /dev/null +++ b/examples/cli/usb_copy/error_handling.sh @@ -0,0 +1,91 @@ +#!/bin/bash + +################################################################################ +# USB Copy Session with Error Handling +# +# Purpose: +# Demonstrates proper error handling using trap to automatically cancel +# sessions when errors occur. +# +# Usage: +# ./error_handling.sh +# +# Prerequisites: +# - airoa-lineage package installed +# - Configuration file or environment variables set up +# +# Key Features: +# - Automatic session cancellation on error +# - Cleanup function with trap EXIT +# - Proper error code handling +################################################################################ + +# Don't use 'set -e' here - we want to handle errors ourselves +set -u # Exit on undefined variable + +# Global variable to store RUN_ID +RUN_ID="" + +# Cleanup function - called on script exit (success or failure) +cleanup() { + local EXIT_CODE=$? + + if [ -n "$RUN_ID" ]; then + if [ $EXIT_CODE -ne 0 ]; then + echo "" + echo "Error detected (exit code: $EXIT_CODE)" + echo "Cancelling USB copy session..." + + # Cancel the session on error + if airoa-lineage-usb-copy cancel --run-id "$RUN_ID"; then + echo "Session cancelled successfully" + else + echo "Warning: Failed to cancel session" + fi + fi + fi +} + +# Register cleanup function to run on exit +trap cleanup EXIT + +# Start session +echo "Starting USB copy session..." +RUN_ID=$(airoa-lineage-usb-copy start \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash $(git rev-parse HEAD) \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main \ + --nominal-start-time "2025-11-26T00:00:00+00:00" \ + --nominal-end-time "2025-11-26T05:00:00+00:00") + +if [ -z "$RUN_ID" ]; then + echo "Error: Failed to start session (no RUN_ID returned)" + exit 1 +fi + +echo "Started USB copy session with RUN_ID: $RUN_ID" + +# Simulate USB copy operation (replace with actual logic) +echo "Performing USB copy operation..." + +# Example: Uncomment the following line to simulate an error +# exit 1 + +# If we get here, operation was successful +echo "USB copy operation completed" + +# Complete the session +echo "Completing session..." +if ! airoa-lineage-usb-copy complete --run-id "$RUN_ID"; then + echo "Error: Failed to complete session" + exit 1 +fi + +# Clear RUN_ID to prevent cleanup from cancelling completed session +RUN_ID="" + +echo "Session completed successfully!" diff --git a/examples/cli/usb_copy/retry_logic.sh b/examples/cli/usb_copy/retry_logic.sh new file mode 100755 index 0000000..c5410df --- /dev/null +++ b/examples/cli/usb_copy/retry_logic.sh @@ -0,0 +1,118 @@ +#!/bin/bash + +################################################################################ +# USB Copy Session with Retry Logic +# +# Purpose: +# Demonstrates automatic retry logic for handling connection errors. +# Useful when Marquez server is temporarily unavailable. +# +# Usage: +# ./retry_logic.sh +# +# Prerequisites: +# - airoa-lineage package installed +# - Configuration file or environment variables set up +# +# Key Features: +# - Automatic retry on connection errors (EXIT_CODE=2) +# - Maximum 3 retry attempts +# - Exponential backoff between retries +################################################################################ + +set -e + +MAX_RETRIES=3 +RETRY_DELAY=2 # seconds + +# Function to start session with retry logic +start_session_with_retry() { + local attempt=1 + + while [ $attempt -le $MAX_RETRIES ]; do + echo "Attempt $attempt/$MAX_RETRIES: Starting USB copy session..." + + # Disable 'set -e' temporarily to capture exit code + set +e + RUN_ID=$(airoa-lineage-usb-copy start \ + --namespace production \ + --robot-id hsr001 \ + --location weblab \ + --repository-hash $(git rev-parse HEAD) \ + --repository-uri https://github.com/AIRoA/airoa-lineage.git \ + --repository-tag v1.0.0 \ + --repository-branch main \ + --nominal-start-time "2025-11-26T00:00:00+00:00" \ + --nominal-end-time "2025-11-26T05:00:00+00:00" 2>&1) + local exit_code=$? + set -e + + # Check exit code + if [ $exit_code -eq 0 ]; then + echo "Successfully started session with RUN_ID: $RUN_ID" + return 0 + elif [ $exit_code -eq 2 ]; then + # EXIT_CODE=2 indicates connection error + echo "Connection error detected" + + if [ $attempt -lt $MAX_RETRIES ]; then + local wait_time=$((RETRY_DELAY * attempt)) + echo "Retrying in ${wait_time} seconds..." + sleep $wait_time + attempt=$((attempt + 1)) + else + echo "Error: Maximum retry attempts reached" + return 1 + fi + else + # Other errors - don't retry + echo "Error: Failed to start session (exit code: $exit_code)" + echo "$RUN_ID" + return 1 + fi + done + + return 1 +} + +# Start session with retry +if ! start_session_with_retry; then + echo "Failed to start USB copy session after $MAX_RETRIES attempts" + exit 1 +fi + +# Perform USB copy operation +echo "Performing USB copy operation..." +# Your USB copy logic here + +# Complete the session with retry logic +echo "Completing session..." +attempt=1 +while [ $attempt -le $MAX_RETRIES ]; do + echo "Attempt $attempt/$MAX_RETRIES: Completing session..." + + set +e + airoa-lineage-usb-copy complete --run-id "$RUN_ID" + exit_code=$? + set -e + + if [ $exit_code -eq 0 ]; then + echo "Session completed successfully!" + exit 0 + elif [ $exit_code -eq 2 ]; then + if [ $attempt -lt $MAX_RETRIES ]; then + wait_time=$((RETRY_DELAY * attempt)) + echo "Connection error. Retrying in ${wait_time} seconds..." + sleep $wait_time + attempt=$((attempt + 1)) + else + echo "Error: Maximum retry attempts reached" + exit 1 + fi + else + echo "Error: Failed to complete session (exit code: $exit_code)" + exit 1 + fi +done + +exit 1 diff --git a/examples/cli/usb_copy/with_config_file.sh b/examples/cli/usb_copy/with_config_file.sh new file mode 100755 index 0000000..a6e4a6b --- /dev/null +++ b/examples/cli/usb_copy/with_config_file.sh @@ -0,0 +1,65 @@ +#!/bin/bash + +################################################################################ +# USB Copy Session with Configuration File +# +# Purpose: +# Demonstrates using a configuration file to minimize CLI arguments. +# Settings are loaded from ~/.config/airoa-lineage/config.json +# +# Usage: +# ./with_config_file.sh +# +# Prerequisites: +# - airoa-lineage package installed +# - Configuration file at ~/.config/airoa-lineage/config.json +# +# Configuration File Setup: +# mkdir -p ~/.config/airoa-lineage +# cat > ~/.config/airoa-lineage/config.json </dev/null; then + export AIROA_REPOSITORY_TAG=$(git describe --exact-match --tags) +else + echo "Warning: No exact tag found for current commit, using 'untagged'" + export AIROA_REPOSITORY_TAG="untagged" +fi + +echo "Configuration from environment variables:" +echo " Namespace: $AIROA_NAMESPACE" +echo " Robot ID: $AIROA_ROBOT_ID" +echo " Location: $AIROA_LOCATION" +echo " Repository Hash: $AIROA_REPOSITORY_HASH" +echo " Repository URI: $AIROA_REPOSITORY_URI" +echo " Repository Tag: $AIROA_REPOSITORY_TAG" +echo " Repository Branch: $AIROA_REPOSITORY_BRANCH" +echo "" + +# Start session - configuration is read from environment variables +RUN_ID=$(airoa-lineage-usb-copy start \ + --nominal-start-time "2025-11-26T00:00:00+00:00" \ + --nominal-end-time "2025-11-26T05:00:00+00:00") + +echo "Started USB copy session with RUN_ID: $RUN_ID" + +# Perform USB copy operation +echo "Performing USB copy operation..." +# Your USB copy logic here + +# Complete the session +echo "Completing session..." +airoa-lineage-usb-copy complete --run-id "$RUN_ID" + +echo "Session completed successfully!" diff --git a/pyproject.toml b/pyproject.toml index d06154c..00a355a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,6 +11,9 @@ dependencies = [ "requests>=2.31.0", ] +[project.scripts] +airoa-lineage-usb-copy = "airoa_lineage.cli.usb_copy:main" + [dependency-groups] dev = [ "mypy>=1.10.0,<1.12.0", diff --git a/src/airoa_lineage/cli/__init__.py b/src/airoa_lineage/cli/__init__.py new file mode 100644 index 0000000..82cf374 --- /dev/null +++ b/src/airoa_lineage/cli/__init__.py @@ -0,0 +1 @@ +"""CLI module for airoa-lineage.""" diff --git a/src/airoa_lineage/cli/config.py b/src/airoa_lineage/cli/config.py new file mode 100644 index 0000000..655132f --- /dev/null +++ b/src/airoa_lineage/cli/config.py @@ -0,0 +1,170 @@ +"""Configuration management for CLI.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any, Dict, Optional + +from airoa_lineage.facets import CommonRunFacet + + +def get_default_config_path() -> Path: + """ + Get default configuration file path (XDG Base Directory compliant). + + Returns: + Path to default config file (~/.config/airoa-lineage/config.json) + + Examples: + >>> path = get_default_config_path() + >>> print(path) + /Users/username/.config/airoa-lineage/config.json + """ + xdg_config_home = os.getenv("XDG_CONFIG_HOME") + if xdg_config_home: + config_dir = Path(xdg_config_home) / "airoa-lineage" + else: + config_dir = Path.home() / ".config" / "airoa-lineage" + + return config_dir / "config.json" + + +def load_config(config_path: Optional[Path] = None) -> Dict[str, Any]: + """ + Load configuration from JSON file and environment variables. + + Priority: CLI args > environment variables > config file > defaults + + This function loads configuration from: + 1. JSON config file (if exists) + 2. Environment variables (override config file) + + CLI arguments should be merged by the caller after calling this function. + + Args: + config_path: Path to config file (default: ~/.config/airoa-lineage/config.json) + + Returns: + Dictionary containing merged configuration + + Raises: + json.JSONDecodeError: If config file contains invalid JSON + FileNotFoundError: If specified config_path does not exist + + Examples: + >>> # Load from default path + >>> config = load_config() + >>> + >>> # Load from custom path + >>> config = load_config(Path("/path/to/config.json")) + >>> + >>> # Access config values + >>> print(config["marquez_url"]) + http://localhost:9000 + """ + config: Dict[str, Any] = {} + + # 1. Load from config file if exists + if config_path is None: + config_path = get_default_config_path() + + if config_path.exists(): + with open(config_path) as f: + config = json.load(f) + + # 2. Override with environment variables + if marquez_url := os.getenv("AIROA_MARQUEZ_URL") or os.getenv("MARQUEZ_URL"): + config["marquez_url"] = marquez_url + + if namespace := os.getenv("AIROA_NAMESPACE"): + config["namespace"] = namespace + + if facet_prefix := os.getenv("AIROA_FACET_PREFIX"): + config["facet_prefix"] = facet_prefix + + if job_name := os.getenv("AIROA_JOB_NAME"): + config["job_name"] = job_name + + # CommonRunFacet fields from environment variables + common_facet = config.get("common_facet", {}) + + if robot_id := os.getenv("AIROA_ROBOT_ID"): + common_facet["robotId"] = robot_id + + if location := os.getenv("AIROA_LOCATION"): + common_facet["location"] = location + + if repository_hash := os.getenv("AIROA_REPOSITORY_HASH"): + common_facet["repositoryHash"] = repository_hash + + if repository_uri := os.getenv("AIROA_REPOSITORY_URI"): + common_facet["repositoryUri"] = repository_uri + + if repository_tag := os.getenv("AIROA_REPOSITORY_TAG"): + common_facet["repositoryTag"] = repository_tag + + if repository_branch := os.getenv("AIROA_REPOSITORY_BRANCH"): + common_facet["repositoryBranch"] = repository_branch + + if common_facet: + config["common_facet"] = common_facet + + return config + + +def build_common_facet(config: Dict[str, Any]) -> CommonRunFacet: + """ + Build CommonRunFacet from configuration. + + Args: + config: Configuration dictionary (from load_config()) + + Returns: + CommonRunFacet instance + + Raises: + ValueError: If required fields are missing + + Examples: + >>> config = { + ... "common_facet": { + ... "robotId": "hsr001", + ... "location": "weblab", + ... "repositoryHash": "df110d5", + ... "repositoryUri": "https://github.com/user/repo.git", + ... "repositoryTag": "v1.0.0", + ... "repositoryBranch": "main" + ... } + ... } + >>> facet = build_common_facet(config) + >>> print(facet.robotId) + hsr001 + """ + facet_data = config.get("common_facet", {}) + + required_fields = [ + "robotId", + "location", + "repositoryHash", + "repositoryUri", + "repositoryTag", + "repositoryBranch", + ] + + missing = [f for f in required_fields if f not in facet_data] + if missing: + raise ValueError( + f"Missing required CommonRunFacet fields: {', '.join(missing)}. " + f"Please provide them in config file, environment variables, or CLI arguments." + ) + + return CommonRunFacet( + robotId=facet_data["robotId"], + location=facet_data["location"], + repositoryHash=facet_data["repositoryHash"], + repositoryUri=facet_data["repositoryUri"], + repositoryTag=facet_data["repositoryTag"], + repositoryBranch=facet_data["repositoryBranch"], + ) diff --git a/src/airoa_lineage/cli/usb_copy.py b/src/airoa_lineage/cli/usb_copy.py new file mode 100644 index 0000000..33e6c31 --- /dev/null +++ b/src/airoa_lineage/cli/usb_copy.py @@ -0,0 +1,501 @@ +"""CLI for USBCopySession. + +This module provides a command-line interface for tracking USB data copy +operations using OpenLineage. + +Usage: + airoa-lineage-usb-copy start [options] + airoa-lineage-usb-copy complete --run-id RUN_ID [options] + airoa-lineage-usb-copy cancel --run-id RUN_ID [options] +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any, Dict + +from airoa_lineage.cli.config import build_common_facet, load_config +from airoa_lineage.usb_copy import USBCopySession + + +def _add_common_args(parser: argparse.ArgumentParser) -> None: + """ + Add common arguments shared across all commands. + + These arguments can be provided via CLI, environment variables, + or configuration file (CLI takes highest precedence). + + Args: + parser: ArgumentParser to add arguments to + """ + parser.add_argument( + "--namespace", + help="OpenLineage namespace", + ) + parser.add_argument( + "--job-name", + help="Job name", + ) + parser.add_argument( + "--marquez-url", + help="Marquez server URL", + ) + parser.add_argument( + "--facet-prefix", + help="Facet prefix (default: empty string)", + ) + + +def _add_common_facet_args(parser: argparse.ArgumentParser) -> None: + """ + Add CommonRunFacet arguments. + + These arguments represent robot and repository metadata required + for OpenLineage event tracking. + + Args: + parser: ArgumentParser to add arguments to + """ + parser.add_argument( + "--robot-id", + help="Robot identifier", + ) + parser.add_argument( + "--location", + help="Location identifier", + ) + parser.add_argument( + "--repository-hash", + help="Git commit hash", + ) + parser.add_argument( + "--repository-uri", + help="Repository URI", + ) + parser.add_argument( + "--repository-tag", + help="Git tag", + ) + parser.add_argument( + "--repository-branch", + help="Git branch", + ) + + +def create_parser() -> argparse.ArgumentParser: + """ + Create argument parser for airoa-lineage-usb-copy CLI. + + Returns: + Configured ArgumentParser instance + """ + parser = argparse.ArgumentParser( + prog="airoa-lineage-usb-copy", + description="Track USB data copy operations with OpenLineage", + ) + + parser.add_argument( + "--version", + action="version", + version="airoa-lineage-usb-copy 0.1.0", + ) + + parser.add_argument( + "--config", + type=str, + help="Path to config file (default: ~/.config/airoa-lineage/config.json)", + ) + + parser.add_argument( + "-v", + "--verbose", + action="store_true", + help="Enable verbose output", + ) + + parser.add_argument( + "-q", + "--quiet", + action="store_true", + help="Suppress output (except errors)", + ) + + parser.add_argument( + "--json", + action="store_true", + help="Output in JSON format", + ) + + # Create subparsers for commands + subparsers = parser.add_subparsers(dest="command", required=True) + + # START command + start_parser = subparsers.add_parser( + "start", + help="Start USB copy session and emit START event", + ) + _add_common_args(start_parser) + _add_common_facet_args(start_parser) + start_parser.add_argument( + "--nominal-start-time", + help="Nominal start time (ISO 8601 format)", + ) + start_parser.add_argument( + "--nominal-end-time", + help="Nominal end time (ISO 8601 format)", + ) + start_parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be done without executing", + ) + + # COMPLETE command + complete_parser = subparsers.add_parser( + "complete", + help="Complete USB copy session and emit COMPLETE event", + ) + complete_parser.add_argument( + "--run-id", + required=True, + help="Session run ID (from start command)", + ) + _add_common_args(complete_parser) + _add_common_facet_args(complete_parser) + + # CANCEL command + cancel_parser = subparsers.add_parser( + "cancel", + help="Cancel USB copy session and emit ABORT event", + ) + cancel_parser.add_argument( + "--run-id", + required=True, + help="Session run ID (from start command)", + ) + _add_common_args(cancel_parser) + _add_common_facet_args(cancel_parser) + + return parser + + +def merge_config_with_args( + config: Dict[str, Any], + args: argparse.Namespace, +) -> Dict[str, Any]: + """ + Merge configuration with CLI arguments. + + CLI arguments take precedence over config file and environment variables. + + Args: + config: Configuration from load_config() + args: Parsed CLI arguments + + Returns: + Merged configuration dictionary + """ + # Top-level configuration + if args.namespace: + config["namespace"] = args.namespace + + if hasattr(args, "marquez_url") and args.marquez_url: + config["marquez_url"] = args.marquez_url + + if hasattr(args, "job_name") and args.job_name: + config["job_name"] = args.job_name + + if hasattr(args, "facet_prefix") and args.facet_prefix: + config["facet_prefix"] = args.facet_prefix + + # CommonRunFacet fields + if hasattr(args, "robot_id") and args.robot_id: + if "common_facet" not in config: + config["common_facet"] = {} + config["common_facet"]["robotId"] = args.robot_id + + if hasattr(args, "location") and args.location: + if "common_facet" not in config: + config["common_facet"] = {} + config["common_facet"]["location"] = args.location + + if hasattr(args, "repository_hash") and args.repository_hash: + if "common_facet" not in config: + config["common_facet"] = {} + config["common_facet"]["repositoryHash"] = args.repository_hash + + if hasattr(args, "repository_uri") and args.repository_uri: + if "common_facet" not in config: + config["common_facet"] = {} + config["common_facet"]["repositoryUri"] = args.repository_uri + + if hasattr(args, "repository_tag") and args.repository_tag: + if "common_facet" not in config: + config["common_facet"] = {} + config["common_facet"]["repositoryTag"] = args.repository_tag + + if hasattr(args, "repository_branch") and args.repository_branch: + if "common_facet" not in config: + config["common_facet"] = {} + config["common_facet"]["repositoryBranch"] = args.repository_branch + + return config + + +def cmd_start(args: argparse.Namespace, config: Dict[str, Any]) -> int: + """ + Execute start command. + + Args: + args: Parsed CLI arguments + config: Merged configuration + + Returns: + Exit code (0=success, 1=error) + """ + try: + # Build CommonRunFacet + common_facet = build_common_facet(config) + + # Validate required fields + if "namespace" not in config: + print( + "Error: namespace is required. " + "Provide via --namespace, config file, or AIROA_NAMESPACE env var.", + file=sys.stderr, + ) + return 1 + + # Dry run mode + if args.dry_run: + print("[DRY RUN] Would start session with:") + print(f" Namespace: {config['namespace']}") + print(f" Job name: {config.get('job_name', 'usb-data-copy')}") + print(f" Robot ID: {common_facet.robotId}") + print(f" Location: {common_facet.location}") + print(f" Repository: {common_facet.repositoryUri}") + print(f" Commit: {common_facet.repositoryHash}") + print(f" Branch: {common_facet.repositoryBranch}") + print(f" Tag: {common_facet.repositoryTag}") + if hasattr(args, "nominal_start_time") and args.nominal_start_time: + print(f" Nominal start: {args.nominal_start_time}") + if hasattr(args, "nominal_end_time") and args.nominal_end_time: + print(f" Nominal end: {args.nominal_end_time}") + print("[DRY RUN] Would output run_id (not generated in dry-run mode)") + return 0 + + # Create session + session = USBCopySession( + namespace=config["namespace"], + common_facet=common_facet, + job_name=config.get("job_name", "usb-data-copy"), + marquez_url=config.get("marquez_url"), + facet_prefix=config.get("facet_prefix", ""), + ) + + # Start session + run_id = session.start( + nominal_start_time=getattr(args, "nominal_start_time", None), + nominal_end_time=getattr(args, "nominal_end_time", None), + ) + + # Output run_id + if args.json: + output = {"run_id": run_id, "status": "started"} + print(json.dumps(output)) + elif args.verbose: + print("Session started successfully", file=sys.stderr) + print(run_id) + else: + # Default: just print run_id + print(run_id) + + return 0 + + except ValueError as e: + print(f"Configuration error: {e}", file=sys.stderr) + return 3 + except RuntimeError as e: + print(f"Session error: {e}", file=sys.stderr) + return 1 + except Exception as e: + print(f"Unexpected error: {e}", file=sys.stderr) + if args.verbose: + import traceback + + traceback.print_exc() + return 1 + + +def cmd_complete(args: argparse.Namespace, config: Dict[str, Any]) -> int: + """ + Execute complete command. + + Args: + args: Parsed CLI arguments + config: Merged configuration + + Returns: + Exit code (0=success, 1=error) + """ + try: + # Build CommonRunFacet + common_facet = build_common_facet(config) + + # Validate required fields + if "namespace" not in config: + print( + "Error: namespace is required. " + "Provide via --namespace, config file, or AIROA_NAMESPACE env var.", + file=sys.stderr, + ) + return 1 + + # Create session + session = USBCopySession( + namespace=config["namespace"], + common_facet=common_facet, + job_name=config.get("job_name", "usb-data-copy"), + marquez_url=config.get("marquez_url"), + facet_prefix=config.get("facet_prefix", ""), + ) + + # Resume existing session + session.resume(args.run_id) + + # Complete session + session.complete() + + # Output success message + if not args.quiet: + if args.json: + output = {"run_id": args.run_id, "status": "completed"} + print(json.dumps(output)) + else: + print(f"Session {args.run_id} completed successfully") + + return 0 + + except ValueError as e: + print(f"Configuration error: {e}", file=sys.stderr) + return 3 + except RuntimeError as e: + print(f"Session error: {e}", file=sys.stderr) + return 1 + except Exception as e: + print(f"Unexpected error: {e}", file=sys.stderr) + if args.verbose: + import traceback + + traceback.print_exc() + return 1 + + +def cmd_cancel(args: argparse.Namespace, config: Dict[str, Any]) -> int: + """ + Execute cancel command. + + Args: + args: Parsed CLI arguments + config: Merged configuration + + Returns: + Exit code (0=success, 1=error) + """ + try: + # Build CommonRunFacet + common_facet = build_common_facet(config) + + # Validate required fields + if "namespace" not in config: + print( + "Error: namespace is required. " + "Provide via --namespace, config file, or AIROA_NAMESPACE env var.", + file=sys.stderr, + ) + return 1 + + # Create session + session = USBCopySession( + namespace=config["namespace"], + common_facet=common_facet, + job_name=config.get("job_name", "usb-data-copy"), + marquez_url=config.get("marquez_url"), + facet_prefix=config.get("facet_prefix", ""), + ) + + # Resume existing session + session.resume(args.run_id) + + # Cancel session + session.cancel() + + # Output success message + if not args.quiet: + if args.json: + output = {"run_id": args.run_id, "status": "cancelled"} + print(json.dumps(output)) + else: + print(f"Session {args.run_id} cancelled successfully") + + return 0 + + except ValueError as e: + print(f"Configuration error: {e}", file=sys.stderr) + return 3 + except RuntimeError as e: + print(f"Session error: {e}", file=sys.stderr) + return 1 + except Exception as e: + print(f"Unexpected error: {e}", file=sys.stderr) + if args.verbose: + import traceback + + traceback.print_exc() + return 1 + + +def main() -> int: + """ + Main entry point for airoa-lineage-usb-copy CLI. + + Returns: + Exit code (0=success, 1=general error, 2=connection error, 3=config error) + """ + parser = create_parser() + args = parser.parse_args() + + # Load configuration + try: + config_path = Path(args.config) if args.config else None + config = load_config(config_path) + except FileNotFoundError as e: + print(f"Configuration file not found: {e}", file=sys.stderr) + return 3 + except json.JSONDecodeError as e: + print(f"Invalid JSON in configuration file: {e}", file=sys.stderr) + return 3 + except Exception as e: + print(f"Error loading configuration: {e}", file=sys.stderr) + return 3 + + # Merge with CLI arguments + config = merge_config_with_args(config, args) + + # Execute command + if args.command == "start": + return cmd_start(args, config) + elif args.command == "complete": + return cmd_complete(args, config) + elif args.command == "cancel": + return cmd_cancel(args, config) + else: + print(f"Unknown command: {args.command}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/airoa_lineage/core/base_session.py b/src/airoa_lineage/core/base_session.py index dc0485f..380d0dc 100644 --- a/src/airoa_lineage/core/base_session.py +++ b/src/airoa_lineage/core/base_session.py @@ -511,3 +511,65 @@ def running(self) -> None: ) self._emit_event(RunState.RUNNING) + + def resume(self, run_id: str) -> None: + """ + Resume an existing session with the given run_id. + + This method allows completing or cancelling a session started + in a different process or instance. It marks the session as + already started without emitting a START event. + + This is particularly useful for CLI commands where start, complete, + and cancel are executed in separate processes. + + Args: + run_id: The run ID of the session to resume (must be valid UUID format) + + Raises: + ValueError: If run_id is not a valid UUID format + RuntimeError: If session was already started, completed, or cancelled + + Examples: + >>> # Start session in one process (e.g., CLI start command) + >>> session1 = USBCopySession(namespace="prod", common_facet=facet) + >>> run_id = session1.start() + >>> + >>> # Complete session in another process (e.g., CLI complete command) + >>> session2 = USBCopySession(namespace="prod", common_facet=facet) + >>> session2.resume(run_id) # Resume existing session + >>> session2.complete() # Complete the session + >>> + >>> # Cancel session (alternative ending) + >>> session3 = USBCopySession(namespace="prod", common_facet=facet) + >>> session3.resume(run_id) + >>> session3.cancel() + """ + # UUID format validation + try: + uuid.UUID(run_id) + except ValueError as e: + raise ValueError( + f"Invalid run_id format: {run_id}. Must be a valid UUID." + ) from e + + # State validation (check completed/cancelled before started) + if self._completed: + raise RuntimeError( + f"Cannot resume session {run_id}. Session already completed." + ) + + if self._cancelled: + raise RuntimeError( + f"Cannot resume session {run_id}. Session already cancelled." + ) + + if self._started: + raise RuntimeError( + f"Cannot resume session {run_id}. " + f"Session already started with run_id {self.run_id}." + ) + + # Set run_id and mark as started + self.run_id = run_id + self._started = True diff --git a/tests/unit/cli/__init__.py b/tests/unit/cli/__init__.py new file mode 100644 index 0000000..336e40b --- /dev/null +++ b/tests/unit/cli/__init__.py @@ -0,0 +1 @@ +"""Unit tests for CLI modules.""" diff --git a/tests/unit/cli/test_config.py b/tests/unit/cli/test_config.py new file mode 100644 index 0000000..93a12dc --- /dev/null +++ b/tests/unit/cli/test_config.py @@ -0,0 +1,259 @@ +"""Unit tests for CLI configuration management.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +from unittest.mock import patch + +import pytest + +from airoa_lineage.cli.config import ( + build_common_facet, + get_default_config_path, + load_config, +) +from airoa_lineage.facets import CommonRunFacet + + +class TestGetDefaultConfigPath: + """Test get_default_config_path() function.""" + + def test_default_path_without_xdg(self): + """Test default path when XDG_CONFIG_HOME is not set.""" + with patch.dict(os.environ, {}, clear=True): + path = get_default_config_path() + expected = Path.home() / ".config" / "airoa-lineage" / "config.json" + assert path == expected + + def test_default_path_with_xdg(self): + """Test default path when XDG_CONFIG_HOME is set.""" + with patch.dict(os.environ, {"XDG_CONFIG_HOME": "/custom/config"}): + path = get_default_config_path() + expected = Path("/custom/config") / "airoa-lineage" / "config.json" + assert path == expected + + +class TestLoadConfig: + """Test load_config() function.""" + + def test_load_from_nonexistent_file(self, tmp_path): + """Test loading from nonexistent file returns empty dict.""" + config_path = tmp_path / "nonexistent.json" + config = load_config(config_path) + assert config == {} + + def test_load_from_json_file(self, tmp_path): + """Test loading from JSON file.""" + config_path = tmp_path / "config.json" + config_data = { + "marquez_url": "http://example.com:9000", + "namespace": "test_namespace", + "common_facet": { + "robotId": "robot001", + "location": "lab", + "repositoryHash": "abc123", + "repositoryUri": "https://github.com/user/repo.git", + "repositoryTag": "v1.0.0", + "repositoryBranch": "main", + }, + } + + with open(config_path, "w") as f: + json.dump(config_data, f) + + config = load_config(config_path) + assert config["marquez_url"] == "http://example.com:9000" + assert config["namespace"] == "test_namespace" + assert config["common_facet"]["robotId"] == "robot001" + + def test_load_from_invalid_json(self, tmp_path): + """Test loading from invalid JSON raises error.""" + config_path = tmp_path / "invalid.json" + config_path.write_text("{ invalid json }") + + with pytest.raises(json.JSONDecodeError): + load_config(config_path) + + def test_env_vars_override_config_file(self, tmp_path): + """Test environment variables override config file values.""" + config_path = tmp_path / "config.json" + config_data = { + "marquez_url": "http://file.com:9000", + "namespace": "file_namespace", + } + + with open(config_path, "w") as f: + json.dump(config_data, f) + + with patch.dict( + os.environ, + { + "AIROA_MARQUEZ_URL": "http://env.com:9000", + "AIROA_NAMESPACE": "env_namespace", + }, + ): + config = load_config(config_path) + assert config["marquez_url"] == "http://env.com:9000" + assert config["namespace"] == "env_namespace" + + def test_airoa_marquez_url_takes_precedence(self, tmp_path): + """Test AIROA_MARQUEZ_URL takes precedence over MARQUEZ_URL.""" + config_path = tmp_path / "config.json" + + with patch.dict( + os.environ, + { + "MARQUEZ_URL": "http://marquez.com:9000", + "AIROA_MARQUEZ_URL": "http://airoa.com:9000", + }, + ): + config = load_config(config_path) + assert config["marquez_url"] == "http://airoa.com:9000" + + def test_marquez_url_fallback(self, tmp_path): + """Test MARQUEZ_URL is used when AIROA_MARQUEZ_URL is not set.""" + config_path = tmp_path / "config.json" + + with patch.dict(os.environ, {"MARQUEZ_URL": "http://marquez.com:9000"}): + config = load_config(config_path) + assert config["marquez_url"] == "http://marquez.com:9000" + + def test_common_facet_env_vars(self, tmp_path): + """Test CommonRunFacet fields from environment variables.""" + config_path = tmp_path / "config.json" + config_data = {"common_facet": {"robotId": "file_robot"}} + + with open(config_path, "w") as f: + json.dump(config_data, f) + + with patch.dict( + os.environ, + { + "AIROA_ROBOT_ID": "env_robot", + "AIROA_LOCATION": "env_location", + "AIROA_REPOSITORY_HASH": "env_hash", + "AIROA_REPOSITORY_URI": "https://env.com/repo.git", + "AIROA_REPOSITORY_TAG": "env_tag", + "AIROA_REPOSITORY_BRANCH": "env_branch", + }, + ): + config = load_config(config_path) + assert config["common_facet"]["robotId"] == "env_robot" + assert config["common_facet"]["location"] == "env_location" + assert config["common_facet"]["repositoryHash"] == "env_hash" + assert config["common_facet"]["repositoryUri"] == "https://env.com/repo.git" + assert config["common_facet"]["repositoryTag"] == "env_tag" + assert config["common_facet"]["repositoryBranch"] == "env_branch" + + def test_partial_common_facet_merge(self, tmp_path): + """Test partial CommonRunFacet fields merge correctly.""" + config_path = tmp_path / "config.json" + config_data = { + "common_facet": { + "robotId": "file_robot", + "location": "file_location", + "repositoryHash": "file_hash", + } + } + + with open(config_path, "w") as f: + json.dump(config_data, f) + + with patch.dict( + os.environ, + { + "AIROA_ROBOT_ID": "env_robot", # Override + "AIROA_REPOSITORY_URI": "https://env.com/repo.git", # Add new + }, + ): + config = load_config(config_path) + # Environment variable overrides file + assert config["common_facet"]["robotId"] == "env_robot" + # File values are kept + assert config["common_facet"]["location"] == "file_location" + assert config["common_facet"]["repositoryHash"] == "file_hash" + # Environment variable adds new field + assert config["common_facet"]["repositoryUri"] == "https://env.com/repo.git" + + +class TestBuildCommonFacet: + """Test build_common_facet() function.""" + + def test_build_facet_with_all_fields(self): + """Test building facet with all required fields.""" + config = { + "common_facet": { + "robotId": "robot001", + "location": "lab", + "repositoryHash": "abc123", + "repositoryUri": "https://github.com/user/repo.git", + "repositoryTag": "v1.0.0", + "repositoryBranch": "main", + } + } + + facet = build_common_facet(config) + + assert isinstance(facet, CommonRunFacet) + assert facet.robotId == "robot001" + assert facet.location == "lab" + assert facet.repositoryHash == "abc123" + assert facet.repositoryUri == "https://github.com/user/repo.git" + assert facet.repositoryTag == "v1.0.0" + assert facet.repositoryBranch == "main" + + def test_build_facet_with_missing_fields(self): + """Test building facet with missing required fields raises error.""" + config = { + "common_facet": { + "robotId": "robot001", + "location": "lab", + # Missing other required fields + } + } + + with pytest.raises(ValueError) as excinfo: + build_common_facet(config) + + error_msg = str(excinfo.value) + assert "Missing required CommonRunFacet fields" in error_msg + assert "repositoryHash" in error_msg + assert "repositoryUri" in error_msg + assert "repositoryTag" in error_msg + assert "repositoryBranch" in error_msg + + def test_build_facet_without_common_facet_key(self): + """Test building facet without common_facet key raises error.""" + config = {} + + with pytest.raises(ValueError) as excinfo: + build_common_facet(config) + + error_msg = str(excinfo.value) + assert "Missing required CommonRunFacet fields" in error_msg + # All fields should be missing + assert "robotId" in error_msg + assert "location" in error_msg + assert "repositoryHash" in error_msg + assert "repositoryUri" in error_msg + assert "repositoryTag" in error_msg + assert "repositoryBranch" in error_msg + + def test_error_message_includes_all_missing_fields(self): + """Test error message lists all missing fields.""" + config = {"common_facet": {"robotId": "robot001"}} + + with pytest.raises(ValueError) as excinfo: + build_common_facet(config) + + error_msg = str(excinfo.value) + # Should list all 5 missing fields + assert "location" in error_msg + assert "repositoryHash" in error_msg + assert "repositoryUri" in error_msg + assert "repositoryTag" in error_msg + assert "repositoryBranch" in error_msg + # Should have helpful message + assert "Please provide them in config file, environment variables" in error_msg diff --git a/tests/unit/cli/test_usb_copy.py b/tests/unit/cli/test_usb_copy.py new file mode 100644 index 0000000..cd42aaf --- /dev/null +++ b/tests/unit/cli/test_usb_copy.py @@ -0,0 +1,268 @@ +"""Tests for CLI usb_copy module.""" + +import argparse + + +from airoa_lineage.cli.usb_copy import ( + _add_common_args, + _add_common_facet_args, + create_parser, + merge_config_with_args, +) + + +class TestHelperFunctions: + """Test helper functions for argument parsing.""" + + def test_add_common_args(self): + """Test that _add_common_args adds expected arguments.""" + parser = argparse.ArgumentParser() + _add_common_args(parser) + + # Parse with arguments + args = parser.parse_args( + [ + "--namespace", + "test", + "--job-name", + "test-job", + "--marquez-url", + "http://localhost:9000", + "--facet-prefix", + "test-prefix", + ] + ) + + assert args.namespace == "test" + assert args.job_name == "test-job" + assert args.marquez_url == "http://localhost:9000" + assert args.facet_prefix == "test-prefix" + + def test_add_common_facet_args(self): + """Test that _add_common_facet_args adds expected arguments.""" + parser = argparse.ArgumentParser() + _add_common_facet_args(parser) + + # Parse with arguments + args = parser.parse_args( + [ + "--robot-id", + "hsr001", + "--location", + "weblab", + "--repository-hash", + "abc123", + "--repository-uri", + "https://github.com/test/repo.git", + "--repository-tag", + "v1.0.0", + "--repository-branch", + "main", + ] + ) + + assert args.robot_id == "hsr001" + assert args.location == "weblab" + assert args.repository_hash == "abc123" + assert args.repository_uri == "https://github.com/test/repo.git" + assert args.repository_tag == "v1.0.0" + assert args.repository_branch == "main" + + +class TestCreateParser: + """Test argument parser creation.""" + + def test_complete_parser_has_common_facet_args(self): + """Test that complete command has CommonRunFacet arguments.""" + parser = create_parser() + + # Parse complete command with CommonRunFacet args + args = parser.parse_args( + [ + "complete", + "--run-id", + "550e8400-e29b-41d4-a716-446655440000", + "--robot-id", + "hsr001", + "--location", + "weblab", + "--repository-hash", + "abc123", + "--repository-uri", + "https://github.com/test/repo.git", + "--repository-tag", + "v1.0.0", + "--repository-branch", + "main", + ] + ) + + assert args.command == "complete" + assert args.run_id == "550e8400-e29b-41d4-a716-446655440000" + assert args.robot_id == "hsr001" + assert args.location == "weblab" + assert args.repository_hash == "abc123" + assert args.repository_uri == "https://github.com/test/repo.git" + assert args.repository_tag == "v1.0.0" + assert args.repository_branch == "main" + + def test_complete_parser_has_facet_prefix(self): + """Test that complete command has --facet-prefix argument.""" + parser = create_parser() + + args = parser.parse_args( + [ + "complete", + "--run-id", + "550e8400-e29b-41d4-a716-446655440000", + "--facet-prefix", + "airoa", + ] + ) + + assert args.facet_prefix == "airoa" + + def test_cancel_parser_has_common_facet_args(self): + """Test that cancel command has CommonRunFacet arguments.""" + parser = create_parser() + + # Parse cancel command with CommonRunFacet args + args = parser.parse_args( + [ + "cancel", + "--run-id", + "550e8400-e29b-41d4-a716-446655440000", + "--robot-id", + "hsr001", + "--location", + "weblab", + "--repository-hash", + "abc123", + "--repository-uri", + "https://github.com/test/repo.git", + "--repository-tag", + "v1.0.0", + "--repository-branch", + "main", + ] + ) + + assert args.command == "cancel" + assert args.run_id == "550e8400-e29b-41d4-a716-446655440000" + assert args.robot_id == "hsr001" + assert args.location == "weblab" + + def test_cancel_parser_has_facet_prefix(self): + """Test that cancel command has --facet-prefix argument.""" + parser = create_parser() + + args = parser.parse_args( + [ + "cancel", + "--run-id", + "550e8400-e29b-41d4-a716-446655440000", + "--facet-prefix", + "airoa", + ] + ) + + assert args.facet_prefix == "airoa" + + +class TestMergeConfigWithArgs: + """Test configuration merging with CLI arguments.""" + + def test_complete_cli_args_override_config(self): + """Test that complete CLI args override config values.""" + config = { + "namespace": "config-namespace", + "common_facet": { + "robotId": "config-robot", + "location": "config-location", + }, + } + + # Simulate complete command args + args = argparse.Namespace( + command="complete", + run_id="550e8400-e29b-41d4-a716-446655440000", + namespace="cli-namespace", + robot_id="cli-robot", + location="cli-location", + repository_hash="cli-hash", + repository_uri="https://github.com/test/repo.git", + repository_tag="v1.0.0", + repository_branch="main", + job_name=None, + marquez_url=None, + facet_prefix=None, + ) + + result = merge_config_with_args(config, args) + + assert result["namespace"] == "cli-namespace" + assert result["common_facet"]["robotId"] == "cli-robot" + assert result["common_facet"]["location"] == "cli-location" + assert result["common_facet"]["repositoryHash"] == "cli-hash" + assert ( + result["common_facet"]["repositoryUri"] + == "https://github.com/test/repo.git" + ) + assert result["common_facet"]["repositoryTag"] == "v1.0.0" + assert result["common_facet"]["repositoryBranch"] == "main" + + def test_facet_prefix_merge_in_complete(self): + """Test that --facet-prefix is correctly merged in complete.""" + config = {"facet_prefix": "config-prefix"} + + args = argparse.Namespace( + command="complete", + run_id="550e8400-e29b-41d4-a716-446655440000", + namespace=None, + job_name=None, + marquez_url=None, + facet_prefix="cli-prefix", + robot_id=None, + location=None, + repository_hash=None, + repository_uri=None, + repository_tag=None, + repository_branch=None, + ) + + result = merge_config_with_args(config, args) + + assert result["facet_prefix"] == "cli-prefix" + + def test_cancel_cli_args_override_config(self): + """Test that cancel CLI args override config values.""" + config = { + "namespace": "config-namespace", + "common_facet": { + "robotId": "config-robot", + "location": "config-location", + }, + } + + # Simulate cancel command args + args = argparse.Namespace( + command="cancel", + run_id="550e8400-e29b-41d4-a716-446655440000", + namespace="cli-namespace", + robot_id="cli-robot", + location="cli-location", + repository_hash="cli-hash", + repository_uri="https://github.com/test/repo.git", + repository_tag="v1.0.0", + repository_branch="main", + job_name=None, + marquez_url=None, + facet_prefix=None, + ) + + result = merge_config_with_args(config, args) + + assert result["namespace"] == "cli-namespace" + assert result["common_facet"]["robotId"] == "cli-robot" + assert result["common_facet"]["location"] == "cli-location" + assert result["common_facet"]["repositoryHash"] == "cli-hash" diff --git a/tests/unit/core/test_base_session.py b/tests/unit/core/test_base_session.py index e5e70dd..52c6a7f 100644 --- a/tests/unit/core/test_base_session.py +++ b/tests/unit/core/test_base_session.py @@ -618,3 +618,155 @@ def test_cancel_event_includes_inputs_outputs( assert cancel_event.inputs[0].name == "input_data" assert len(cancel_event.outputs) == 1 assert cancel_event.outputs[0].name == "output_data" + + +class TestBaseSessionResume: + """Test BaseSession resume() method.""" + + @patch("airoa_lineage.core.base_session.MarquezClient") + def test_resume_with_valid_uuid(self, mock_client_class, common_facet): + """Test resume() with valid UUID format.""" + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + session = MinimalSession("test_namespace", common_facet) + test_run_id = str(uuid.uuid4()) + + session.resume(test_run_id) + + # Verify run_id was set + assert session.run_id == test_run_id + # Verify session marked as started + assert session._started is True + # Verify no events were emitted + mock_client.emit.assert_not_called() + + @patch("airoa_lineage.core.base_session.MarquezClient") + def test_resume_with_invalid_uuid_raises_error( + self, mock_client_class, common_facet + ): + """Test resume() with invalid UUID format raises ValueError.""" + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + session = MinimalSession("test_namespace", common_facet) + + with pytest.raises(ValueError) as excinfo: + session.resume("not-a-uuid") + + assert "Invalid run_id format" in str(excinfo.value) + assert "Must be a valid UUID" in str(excinfo.value) + # Verify session not marked as started + assert session._started is False + + @patch("airoa_lineage.core.base_session.MarquezClient") + def test_resume_after_start_raises_error(self, mock_client_class, common_facet): + """Test resume() after start() raises RuntimeError.""" + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + session = MinimalSession("test_namespace", common_facet) + session.start() + + test_run_id = str(uuid.uuid4()) + with pytest.raises(RuntimeError) as excinfo: + session.resume(test_run_id) + + assert "Cannot resume session" in str(excinfo.value) + assert "already started" in str(excinfo.value) + + @patch("airoa_lineage.core.base_session.MarquezClient") + def test_resume_after_complete_raises_error(self, mock_client_class, common_facet): + """Test resume() after complete() raises RuntimeError.""" + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + session = MinimalSession("test_namespace", common_facet) + session.start() + session.complete() + + test_run_id = str(uuid.uuid4()) + with pytest.raises(RuntimeError) as excinfo: + session.resume(test_run_id) + + assert "Cannot resume session" in str(excinfo.value) + assert "already completed" in str(excinfo.value) + + @patch("airoa_lineage.core.base_session.MarquezClient") + def test_resume_after_cancel_raises_error(self, mock_client_class, common_facet): + """Test resume() after cancel() raises RuntimeError.""" + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + session = MinimalSession("test_namespace", common_facet) + session.start() + session.cancel() + + test_run_id = str(uuid.uuid4()) + with pytest.raises(RuntimeError) as excinfo: + session.resume(test_run_id) + + assert "Cannot resume session" in str(excinfo.value) + assert "already cancelled" in str(excinfo.value) + + @patch("airoa_lineage.core.base_session.MarquezClient") + def test_resume_then_complete(self, mock_client_class, common_facet): + """Test resume() followed by complete() works correctly.""" + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + session = MinimalSession("test_namespace", common_facet) + test_run_id = str(uuid.uuid4()) + + # Resume existing session + session.resume(test_run_id) + # Complete the session + session.complete() + + # Verify state + assert session._started is True + assert session._completed is True + assert session.run_id == test_run_id + # Verify only complete event was emitted (not start) + assert mock_client.emit.call_count == 1 + + @patch("airoa_lineage.core.base_session.MarquezClient") + def test_resume_then_cancel(self, mock_client_class, common_facet): + """Test resume() followed by cancel() works correctly.""" + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + session = MinimalSession("test_namespace", common_facet) + test_run_id = str(uuid.uuid4()) + + # Resume existing session + session.resume(test_run_id) + # Cancel the session + session.cancel() + + # Verify state + assert session._started is True + assert session._cancelled is True + assert session.run_id == test_run_id + # Verify only cancel event was emitted (not start) + assert mock_client.emit.call_count == 1 + + @patch("airoa_lineage.core.base_session.MarquezClient") + def test_resume_then_running(self, mock_client_class, common_facet): + """Test resume() followed by running() works for supported sessions.""" + mock_client = MagicMock() + mock_client_class.return_value = mock_client + + session = RunningSession("test_namespace", common_facet) + test_run_id = str(uuid.uuid4()) + + # Resume existing session + session.resume(test_run_id) + # Send running event + session.running() + + # Verify state + assert session._started is True + assert session.run_id == test_run_id + # Verify only running event was emitted (not start) + assert mock_client.emit.call_count == 1