332 lines
12 KiB
Python
332 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Debian Forge CLI Integration Module
|
|
|
|
This module provides integration between debian-forge and debian-forge-cli,
|
|
ensuring 1:1 compatibility with the upstream osbuild/image-builder-cli.
|
|
"""
|
|
|
|
import json
|
|
import subprocess
|
|
import os
|
|
from typing import Dict, List, Optional, Any
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
@dataclass
|
|
class ImageBuildRequest:
|
|
"""Image build request for CLI integration"""
|
|
blueprint_path: str
|
|
output_format: str
|
|
output_path: str
|
|
architecture: str = "amd64"
|
|
distro: str = "debian-12"
|
|
extra_repos: Optional[List[str]] = None
|
|
ostree_ref: Optional[str] = None
|
|
ostree_parent: Optional[str] = None
|
|
ostree_url: Optional[str] = None
|
|
|
|
@dataclass
|
|
class ImageBuildResult:
|
|
"""Result of image build operation"""
|
|
success: bool
|
|
output_path: str
|
|
build_time: float
|
|
image_size: int
|
|
error_message: Optional[str] = None
|
|
metadata: Optional[Dict[str, Any]] = None
|
|
|
|
class DebianForgeCLI:
|
|
"""Integration with debian-forge-cli (fork of osbuild/image-builder-cli)"""
|
|
|
|
def __init__(self, cli_path: str = "../debian-forge-cli", data_dir: str = "./cli-data"):
|
|
self.cli_path = Path(cli_path)
|
|
self.data_dir = Path(data_dir)
|
|
self.cli_binary = self.cli_path / "cmd" / "image-builder" / "image-builder"
|
|
|
|
# Ensure data directory exists
|
|
self.data_dir.mkdir(exist_ok=True)
|
|
|
|
# Verify CLI binary exists
|
|
if not self.cli_binary.exists():
|
|
raise FileNotFoundError(f"CLI binary not found at {self.cli_binary}")
|
|
|
|
def list_images(self, filter_args: Optional[List[str]] = None,
|
|
format_type: str = "json") -> List[Dict[str, Any]]:
|
|
"""List available images using the CLI"""
|
|
cmd = [str(self.cli_binary), "list", "--data-dir", str(self.data_dir)]
|
|
|
|
if filter_args:
|
|
for filter_arg in filter_args:
|
|
cmd.extend(["--filter", filter_arg])
|
|
|
|
if format_type:
|
|
cmd.extend(["--format", format_type])
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
|
|
if format_type == "json":
|
|
return json.loads(result.stdout)
|
|
else:
|
|
return [{"output": result.stdout}]
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"CLI list command failed: {e}")
|
|
print(f"Error output: {e.stderr}")
|
|
return []
|
|
except json.JSONDecodeError as e:
|
|
print(f"Failed to parse CLI output: {e}")
|
|
return []
|
|
|
|
def build_image(self, request: ImageBuildRequest) -> ImageBuildResult:
|
|
"""Build an image using the CLI"""
|
|
cmd = [
|
|
str(self.cli_binary), "build",
|
|
"--blueprint", request.blueprint_path,
|
|
"--output", request.output_path,
|
|
"--format", request.output_format,
|
|
"--data-dir", str(self.data_dir)
|
|
]
|
|
|
|
# Add architecture if specified
|
|
if request.architecture:
|
|
cmd.extend(["--arch", request.architecture])
|
|
|
|
# Add distro if specified
|
|
if request.distro:
|
|
cmd.extend(["--distro", request.distro])
|
|
|
|
# Add extra repositories
|
|
if request.extra_repos:
|
|
for repo in request.extra_repos:
|
|
cmd.extend(["--extra-repo", repo])
|
|
|
|
# Add OSTree options
|
|
if request.ostree_ref:
|
|
cmd.extend(["--ostree-ref", request.ostree_ref])
|
|
if request.ostree_parent:
|
|
cmd.extend(["--ostree-parent", request.ostree_parent])
|
|
if request.ostree_url:
|
|
cmd.extend(["--ostree-url", request.ostree_url])
|
|
|
|
try:
|
|
print(f"Executing CLI build command: {' '.join(cmd)}")
|
|
|
|
start_time = time.time()
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
build_time = time.time() - start_time
|
|
|
|
# Check if output file was created
|
|
output_file = Path(request.output_path)
|
|
if output_file.exists():
|
|
image_size = output_file.stat().st_size
|
|
return ImageBuildResult(
|
|
success=True,
|
|
output_path=str(output_file),
|
|
build_time=build_time,
|
|
image_size=image_size,
|
|
metadata={"cli_output": result.stdout}
|
|
)
|
|
else:
|
|
return ImageBuildResult(
|
|
success=False,
|
|
output_path=request.output_path,
|
|
build_time=build_time,
|
|
image_size=0,
|
|
error_message="Output file not created"
|
|
)
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return ImageBuildResult(
|
|
success=False,
|
|
output_path=request.output_path,
|
|
build_time=0,
|
|
image_size=0,
|
|
error_message=f"CLI build failed: {e.stderr}"
|
|
)
|
|
|
|
def describe_image(self, image_path: str) -> Dict[str, Any]:
|
|
"""Describe an image using the CLI"""
|
|
cmd = [
|
|
str(self.cli_binary), "describeimg",
|
|
"--data-dir", str(self.data_dir),
|
|
image_path
|
|
]
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
|
|
# Try to parse as JSON, fallback to text
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
return {"description": result.stdout, "raw_output": True}
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
return {"error": f"Failed to describe image: {e.stderr}"}
|
|
|
|
def list_distros(self) -> List[str]:
|
|
"""List available distributions"""
|
|
cmd = [
|
|
str(self.cli_binary), "distro",
|
|
"--data-dir", str(self.data_dir)
|
|
]
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
return [line.strip() for line in result.stdout.splitlines() if line.strip()]
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Failed to list distros: {e.stderr}")
|
|
return []
|
|
|
|
def list_repositories(self) -> List[Dict[str, Any]]:
|
|
"""List available repositories"""
|
|
cmd = [
|
|
str(self.cli_binary), "repos",
|
|
"--data-dir", str(self.data_dir)
|
|
]
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
|
|
# Try to parse as JSON
|
|
try:
|
|
return json.loads(result.stdout)
|
|
except json.JSONDecodeError:
|
|
# Parse text output
|
|
repos = []
|
|
current_repo = {}
|
|
|
|
for line in result.stdout.splitlines():
|
|
line = line.strip()
|
|
if line.startswith("Repository:"):
|
|
if current_repo:
|
|
repos.append(current_repo)
|
|
current_repo = {"name": line.split(":", 1)[1].strip()}
|
|
elif ":" in line and current_repo:
|
|
key, value = line.split(":", 1)
|
|
current_repo[key.strip().lower()] = value.strip()
|
|
|
|
if current_repo:
|
|
repos.append(current_repo)
|
|
|
|
return repos
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
print(f"Failed to list repositories: {e.stderr}")
|
|
return []
|
|
|
|
def get_cli_version(self) -> str:
|
|
"""Get CLI version information"""
|
|
cmd = [str(self.cli_binary), "version"]
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
return result.stdout.strip()
|
|
except subprocess.CalledProcessError as e:
|
|
return f"Version unknown: {e.stderr}"
|
|
|
|
def validate_blueprint(self, blueprint_path: str) -> Dict[str, Any]:
|
|
"""Validate a blueprint using the CLI"""
|
|
# First check if blueprint file exists
|
|
if not Path(blueprint_path).exists():
|
|
return {"valid": False, "error": "Blueprint file not found"}
|
|
|
|
# Try to parse the blueprint as JSON
|
|
try:
|
|
with open(blueprint_path, 'r') as f:
|
|
blueprint_data = json.load(f)
|
|
|
|
# Basic validation
|
|
required_fields = ["name", "version"]
|
|
missing_fields = [field for field in required_fields if field not in blueprint_data]
|
|
|
|
if missing_fields:
|
|
return {
|
|
"valid": False,
|
|
"error": f"Missing required fields: {missing_fields}"
|
|
}
|
|
|
|
# Try to use CLI to validate (if it supports validation)
|
|
try:
|
|
cmd = [
|
|
str(self.cli_binary), "build",
|
|
"--blueprint", blueprint_path,
|
|
"--dry-run",
|
|
"--data-dir", str(self.data_dir)
|
|
]
|
|
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
return {"valid": True, "cli_validation": "passed"}
|
|
|
|
except subprocess.CalledProcessError:
|
|
# CLI doesn't support dry-run, but JSON is valid
|
|
return {"valid": True, "json_validation": "passed"}
|
|
|
|
except json.JSONDecodeError as e:
|
|
return {"valid": False, "error": f"Invalid JSON: {e}"}
|
|
|
|
def create_debian_blueprint(self, name: str, version: str,
|
|
packages: List[str],
|
|
customizations: Optional[Dict[str, Any]] = None) -> str:
|
|
"""Create a Debian-specific blueprint for CLI use"""
|
|
blueprint = {
|
|
"name": name,
|
|
"version": version,
|
|
"description": f"Debian atomic blueprint for {name}",
|
|
"packages": packages,
|
|
"modules": [],
|
|
"groups": [],
|
|
"customizations": customizations or {}
|
|
}
|
|
|
|
# Add Debian-specific customizations
|
|
if "debian" not in blueprint["customizations"]:
|
|
blueprint["customizations"]["debian"] = {
|
|
"repositories": [
|
|
{
|
|
"name": "debian-main",
|
|
"baseurl": "http://deb.debian.org/debian",
|
|
"enabled": True
|
|
}
|
|
]
|
|
}
|
|
|
|
# Save blueprint to data directory
|
|
blueprint_path = self.data_dir / f"{name}-{version}.json"
|
|
with open(blueprint_path, 'w') as f:
|
|
json.dump(blueprint, f, indent=2)
|
|
|
|
return str(blueprint_path)
|
|
|
|
def test_cli_integration(self) -> Dict[str, Any]:
|
|
"""Test CLI integration functionality"""
|
|
results = {
|
|
"cli_binary_exists": self.cli_binary.exists(),
|
|
"cli_version": self.get_cli_version(),
|
|
"data_dir_created": self.data_dir.exists(),
|
|
"distros_available": len(self.list_distros()) > 0,
|
|
"repos_available": len(self.list_repositories()) > 0
|
|
}
|
|
|
|
# Test blueprint creation
|
|
try:
|
|
test_blueprint = self.create_debian_blueprint(
|
|
"test-integration", "1.0.0", ["bash", "coreutils"]
|
|
)
|
|
blueprint_validation = self.validate_blueprint(test_blueprint)
|
|
results["blueprint_creation"] = blueprint_validation["valid"]
|
|
|
|
# Clean up test blueprint
|
|
if Path(test_blueprint).exists():
|
|
os.remove(test_blueprint)
|
|
|
|
except Exception as e:
|
|
results["blueprint_creation"] = False
|
|
results["blueprint_error"] = str(e)
|
|
|
|
return results
|
|
|
|
# Import time module for build timing
|
|
import time
|