did stuff
This commit is contained in:
robojerk
parent
3f2346b201
commit
ee02c74250
10 changed files with 1511 additions and 0 deletions
332
cli_integration.py
Normal file
332
cli_integration.py
Normal file
|
|
@ -0,0 +1,332 @@
|
|||
#!/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
|
||||
Loading…
Add table
Add a link
Reference in a new issue