Coverage for fastblocks / mcp / cli.py: 0%
351 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-19 12:49 -0800
1"""Command-line interface for FastBlocks MCP adapter management."""
3import asyncio
4import json
5import typing as t
6from pathlib import Path
8import click
10from .config_audit import ConfigurationAuditor
11from .config_cli import config_cli
12from .config_health import ConfigurationHealthChecker
13from .config_migration import ConfigurationMigrationManager
14from .configuration import ConfigurationManager
15from .env_manager import EnvironmentManager
16from .health import HealthCheckSystem
17from .registry import AdapterRegistry
19# Audit output formatting helpers
20_SEVERITY_COLORS = {
21 "critical": "red",
22 "high": "red",
23 "medium": "yellow",
24 "low": "white",
25 "info": "cyan",
26}
28# Health check formatting helpers
29_HEALTH_STATUS_COLORS = {
30 "healthy": "green",
31 "warning": "yellow",
32 "error": "red",
33}
36def _display_health_result_summary(name: str, result: t.Any) -> None:
37 """Display a single health check result in summary format."""
38 status_color = _HEALTH_STATUS_COLORS.get(result.status, "white")
39 click.echo(f"{name:<20} ", nl=False)
40 click.secho(f"{result.status.upper():<8}", fg=status_color, nl=False)
41 click.echo(f" {result.message}")
44def _display_health_result_detail(adapter_name: str, result: t.Any) -> None:
45 """Display a single health check result with full details."""
46 status_color = _HEALTH_STATUS_COLORS.get(result.status, "white")
48 click.echo(f"Health Check: {adapter_name}")
49 click.secho(f"Status: {result.status.upper()}", fg=status_color)
50 click.echo(f"Message: {result.message}")
51 click.echo(f"Duration: {result.duration_ms:.2f}ms")
53 if result.details:
54 click.echo("Details:")
55 for key, value in result.details.items():
56 if isinstance(value, list):
57 click.echo(f" {key}: {', '.join(value) if value else 'None'}")
58 else:
59 click.echo(f" {key}: {value}")
62def _display_system_health_summary(summary: dict[str, t.Any]) -> None:
63 """Display overall system health summary."""
64 total = summary["total_adapters"]
65 click.echo("System Health Summary:")
66 click.echo(f"Total Adapters: {total}")
67 if total > 0:
68 click.secho(f"Healthy: {summary['healthy_adapters']}", fg="green")
69 click.secho(f"Warnings: {summary['warning_adapters']}", fg="yellow")
70 click.secho(f"Errors: {summary['error_adapters']}", fg="red")
71 click.echo(f"Unknown: {summary['unknown_adapters']}")
74def _display_migration_compatibility(
75 compatibility: dict[str, t.Any], target_version: str
76) -> None:
77 """Display migration compatibility information."""
78 click.echo(f"Migration from {compatibility['current_version']} to {target_version}")
79 click.echo(f"Steps: {' -> '.join(compatibility['migration_path'])}")
81 if compatibility["warnings"]:
82 click.echo("\nWarnings:")
83 for warning in compatibility["warnings"]:
84 click.echo(f" - {warning}")
87def _display_migration_incompatibility(compatibility: dict[str, t.Any]) -> None:
88 """Display migration incompatibility errors."""
89 click.echo("Migration not possible:", err=True)
90 for warning in compatibility["warnings"]:
91 click.echo(f" - {warning}", err=True)
94def _display_migration_success(result: t.Any) -> None:
95 """Display successful migration results."""
96 click.secho("✓ Migration completed successfully", fg="green")
97 click.echo(f"Steps applied: {', '.join(result.steps_applied)}")
98 if result.warnings:
99 click.echo("Warnings:")
100 for warning in result.warnings:
101 click.secho(f" - {warning}", fg="yellow")
104def _display_migration_failure(result: t.Any) -> None:
105 """Display migration failure errors."""
106 click.secho("✗ Migration failed", fg="red")
107 for error in result.errors:
108 click.echo(f" - {error}")
111def _format_finding_for_json(finding: t.Any) -> dict[str, t.Any]:
112 """Format a single finding for JSON output."""
113 return {
114 "id": finding.id,
115 "category": finding.category.value,
116 "severity": finding.severity.value,
117 "title": finding.title,
118 "description": finding.description,
119 "recommendation": finding.recommendation,
120 "affected_items": finding.affected_items,
121 }
124def _write_json_audit_report(report: t.Any, output_path: str | None) -> None:
125 """Write audit report in JSON format."""
126 import json
128 output_data = {
129 "configuration_name": report.configuration_name,
130 "profile": report.profile,
131 "score": report.score,
132 "summary": report.summary,
133 "findings": [_format_finding_for_json(f) for f in report.findings],
134 }
136 if output_path:
137 with open(output_path, "w") as f:
138 json.dump(output_data, f, indent=2)
139 else:
140 click.echo(json.dumps(output_data, indent=2))
143def _display_audit_finding(finding: t.Any) -> None:
144 """Display a single audit finding to console."""
145 severity_color = _SEVERITY_COLORS.get(finding.severity.value, "white")
147 click.secho(
148 f"\n[{finding.severity.value.upper()}] {finding.title}",
149 fg=severity_color,
150 )
151 click.echo(f" {finding.description}")
152 click.echo(f" Recommendation: {finding.recommendation}")
153 if finding.affected_items:
154 click.echo(f" Affected: {', '.join(finding.affected_items)}")
157def _display_text_audit_report(report: t.Any) -> None:
158 """Display audit report in text format to console."""
159 click.echo(f"Configuration Audit Report: {report.configuration_name}")
160 click.echo(f"Profile: {report.profile}")
161 click.echo(f"Score: {report.score}/100")
162 click.echo(f"Total Findings: {report.summary['total_findings']}")
164 if report.findings:
165 click.echo("\nFindings:")
166 for finding in report.findings:
167 _display_audit_finding(finding)
169 if report.recommendations:
170 click.echo("\nRecommendations:")
171 for rec in report.recommendations:
172 click.echo(f" • {rec}")
175def _write_text_audit_report(report: t.Any, output_path: str) -> None:
176 """Write audit report in text format to file."""
177 with open(output_path, "w") as f:
178 f.write("Configuration Audit Report\n")
179 f.write("========================\n\n")
180 f.write(f"Configuration: {report.configuration_name}\n")
181 f.write(f"Profile: {report.profile}\n")
182 f.write(f"Score: {report.score}/100\n\n")
185async def get_registry_and_health() -> tuple[AdapterRegistry, HealthCheckSystem]:
186 """Get initialized registry and health check system."""
187 registry = AdapterRegistry()
188 await registry.initialize()
189 health = HealthCheckSystem(registry)
190 return registry, health
193@click.group()
194def cli() -> None:
195 """FastBlocks MCP Adapter Management CLI."""
196 pass
199# Add configuration management commands
200cli.add_command(config_cli, name="config")
203@cli.command()
204@click.option("--format", type=click.Choice(["json", "table"]), default="table")
205def list_adapters(format: str) -> None:
206 """List all available adapters."""
208 async def _list() -> None:
209 registry, _ = await get_registry_and_health()
210 adapters = await registry.list_available_adapters()
212 if format == "json":
213 output = {name: info.to_dict() for name, info in adapters.items()}
214 click.echo(json.dumps(output, indent=2))
215 else:
216 click.echo("Available Adapters:")
217 click.echo("-" * 50)
218 for name, info in adapters.items():
219 status_color = "green" if info.module_status == "stable" else "yellow"
220 click.echo(f" {name:<20} {info.category:<12} ", nl=False)
221 click.secho(info.module_status, fg=status_color)
222 if info.description:
223 click.echo(f" {info.description}")
225 asyncio.run(_list())
228@cli.command()
229@click.option("--category", help="Filter by category")
230def list_categories(category: str | None = None) -> None:
231 """List adapter categories."""
233 async def _list_categories() -> None:
234 registry, _ = await get_registry_and_health()
236 if category:
237 adapters = await registry.get_adapters_by_category(category)
238 click.echo(f"Adapters in category '{category}':")
239 for adapter in adapters:
240 click.echo(f" - {adapter.name}")
241 else:
242 categories = await registry.get_categories()
243 click.echo("Available Categories:")
244 for cat in sorted(categories):
245 click.echo(f" - {cat}")
247 asyncio.run(_list_categories())
250@cli.command()
251@click.argument("adapter_name")
252@click.option("--format", type=click.Choice(["json", "text"]), default="text")
253def inspect(adapter_name: str, format: str) -> None:
254 """Inspect a specific adapter."""
256 async def _inspect() -> None:
257 registry, _ = await get_registry_and_health()
258 info = await registry.get_adapter_info(adapter_name)
260 if not info:
261 click.echo(f"Adapter '{adapter_name}' not found", err=True)
262 return
264 if format == "json":
265 click.echo(json.dumps(info.to_dict(), indent=2))
266 else:
267 click.echo(f"Adapter: {info.name}")
268 click.echo(f"Class: {info.class_name}")
269 click.echo(f"Module: {info.module_path}")
270 click.echo(f"Category: {info.category}")
271 click.echo(f"Status: {info.module_status}")
272 click.echo(f"ID: {info.module_id}")
273 if info.description:
274 click.echo(f"Description: {info.description}")
275 if info.protocols:
276 click.echo(f"Protocols: {', '.join(info.protocols)}")
277 if info.settings_class:
278 click.echo(f"Settings Class: {info.settings_class}")
280 asyncio.run(_inspect())
283@cli.command()
284@click.argument("adapter_name")
285def validate(adapter_name: str) -> None:
286 """Validate an adapter configuration."""
288 async def _validate() -> None:
289 registry, _ = await get_registry_and_health()
290 result = await registry.validate_adapter(adapter_name)
292 if result["valid"]:
293 click.secho(f"✓ Adapter '{adapter_name}' is valid", fg="green")
294 else:
295 click.secho(f"✗ Adapter '{adapter_name}' has issues", fg="red")
297 if result["errors"]:
298 click.echo("Errors:")
299 for error in result["errors"]:
300 click.secho(f" - {error}", fg="red")
302 if result["warnings"]:
303 click.echo("Warnings:")
304 for warning in result["warnings"]:
305 click.secho(f" - {warning}", fg="yellow")
307 asyncio.run(_validate())
310async def _display_all_adapters_health(health_system: t.Any, format: str) -> None:
311 """Display health results for all adapters."""
312 results = await health_system.check_all_adapters()
313 if format == "json":
314 output = {name: result.to_dict() for name, result in results.items()}
315 click.echo(json.dumps(output, indent=2))
316 else:
317 click.echo("Health Check Results:")
318 click.echo("-" * 50)
319 for name, result in results.items():
320 _display_health_result_summary(name, result)
323async def _display_single_adapter_health(
324 health_system: t.Any, adapter_name: str, format: str
325) -> None:
326 """Display health result for a single adapter."""
327 result = await health_system.check_adapter_health(adapter_name)
328 if format == "json":
329 click.echo(json.dumps(result.to_dict(), indent=2))
330 else:
331 _display_health_result_detail(adapter_name, result)
334async def _display_system_health_summary_cli(health_system: t.Any, format: str) -> None:
335 """Display system health summary."""
336 summary = health_system.get_system_health_summary()
337 if format == "json":
338 click.echo(json.dumps(summary, indent=2))
339 else:
340 _display_system_health_summary(summary)
343@cli.command()
344@click.argument("adapter_name", required=False)
345@click.option("--all", is_flag=True, help="Check all adapters")
346@click.option("--format", type=click.Choice(["json", "text"]), default="text")
347def health(
348 adapter_name: str | None = None, all: bool = False, format: str = "text"
349) -> None:
350 """Perform health checks on adapters."""
352 async def _health() -> None:
353 _registry, health_system = await get_registry_and_health()
355 if all:
356 await _display_all_adapters_health(health_system, format)
357 elif adapter_name:
358 await _display_single_adapter_health(health_system, adapter_name, format)
359 else:
360 await _display_system_health_summary_cli(health_system, format)
362 asyncio.run(_health())
365@cli.command()
366def statistics() -> None:
367 """Show adapter statistics."""
369 async def _stats() -> None:
370 registry, _ = await get_registry_and_health()
371 stats = await registry.get_adapter_statistics()
373 click.echo("Adapter Statistics:")
374 click.echo(f"Total Available: {stats['total_available']}")
375 click.echo(f"Total Active: {stats['total_active']}")
376 click.echo(f"Categories: {stats['total_categories']}")
378 click.echo("\nBy Category:")
379 for category, info in stats["categories"].items():
380 click.echo(f" {category}: {info['total']} adapters")
382 click.echo("\nBy Status:")
383 for status, count in stats["status_breakdown"].items():
384 click.echo(f" {status}: {count}")
386 if stats["active_adapters"]:
387 click.echo(f"\nActive Adapters: {', '.join(stats['active_adapters'])}")
389 asyncio.run(_stats())
392@cli.command()
393@click.option(
394 "--auto-register", is_flag=True, help="Automatically register all adapters"
395)
396def register(auto_register: bool = False) -> None:
397 """Register adapters with the system."""
399 async def _register() -> None:
400 registry, _ = await get_registry_and_health()
402 if auto_register:
403 results = await registry.auto_register_available_adapters()
405 success_count = sum(1 for success in results.values() if success)
406 total_count = len(results)
408 click.echo(
409 f"Auto-registration completed: {success_count}/{total_count} successful"
410 )
412 for name, success in results.items():
413 status = "✓" if success else "✗"
414 color = "green" if success else "red"
415 click.secho(f" {status} {name}", fg=color)
416 else:
417 click.echo("Use --auto-register to register all available adapters")
419 asyncio.run(_register())
422@cli.command()
423@click.argument("config_file")
424@click.option("--output", "-o", help="Output report file")
425@click.option("--format", type=click.Choice(["json", "text"]), default="text")
426def audit(config_file: str, output: str | None, format: str) -> None:
427 """Audit configuration for security and compliance."""
429 async def _audit() -> None:
430 registry = AdapterRegistry()
431 env_manager = EnvironmentManager()
432 auditor = ConfigurationAuditor(env_manager)
434 await registry.initialize()
436 config_manager = ConfigurationManager(registry)
437 await config_manager.initialize()
439 try:
440 config = await config_manager.load_configuration(config_file)
441 report = await auditor.audit_configuration(config)
443 if format == "json":
444 _write_json_audit_report(report, output)
445 else:
446 _display_text_audit_report(report)
447 if output:
448 _write_text_audit_report(report, output)
450 except Exception as e:
451 click.echo(f"Error: {e}", err=True)
453 asyncio.run(_audit())
456@cli.command()
457@click.argument("config_file")
458@click.argument("target_version")
459@click.option(
460 "--backup/--no-backup", default=True, help="Create backup before migration"
461)
462@click.option("--output", "-o", help="Output file for migrated configuration")
463def migrate(
464 config_file: str, target_version: str, backup: bool, output: str | None
465) -> None:
466 """Migrate configuration to target version."""
468 async def _migrate() -> None:
469 migration_manager = ConfigurationMigrationManager()
471 config_path = Path(config_file)
472 if not config_path.exists():
473 click.echo(f"Configuration file not found: {config_file}", err=True)
474 return
476 # Validate migration compatibility
477 compatibility = await migration_manager.validate_migration_compatibility(
478 config_path, target_version
479 )
481 if not compatibility["compatible"]:
482 _display_migration_incompatibility(compatibility)
483 return
485 _display_migration_compatibility(compatibility, target_version)
487 if not click.confirm("Continue with migration?"):
488 return
490 # Create backup if requested
491 if backup:
492 backup_path = await migration_manager.create_migration_backup(
493 config_path, target_version
494 )
495 click.echo(f"Backup created: {backup_path}")
497 # Perform migration
498 result = await migration_manager.migrate_configuration_file(
499 config_path, target_version, Path(output) if output else None
500 )
502 if result.success:
503 _display_migration_success(result)
504 else:
505 _display_migration_failure(result)
507 asyncio.run(_migrate())
510def _parse_test_types(test_types: str | None) -> list[t.Any] | None:
511 """Parse comma-separated test types into a list."""
512 if not test_types:
513 return None
515 from .config_health import ConfigurationTestType
517 return [
518 ConfigurationTestType(test_type.strip()) for test_type in test_types.split(",")
519 ]
522def _display_health_summary(report: t.Any, config_file: str) -> None:
523 """Display health check summary information."""
524 status_color = {"valid": "green", "warning": "yellow", "error": "red"}.get(
525 report.overall_status.value, "white"
526 )
528 click.echo(f"Configuration Health Check: {config_file}")
529 click.secho(
530 f"Overall Status: {report.overall_status.value.upper()}",
531 fg=status_color,
532 )
533 click.echo(f"Total Tests: {report.summary['total_tests']}")
534 click.echo(f"Passed: {report.summary['passed_tests']}")
535 click.echo(f"Failed: {report.summary['failed_tests']}")
536 click.echo(f"Pass Rate: {report.summary['pass_rate']:.1f}%")
539def _display_failed_tests(failed_tests: list[t.Any]) -> None:
540 """Display failed test details."""
541 if not failed_tests:
542 return
544 click.echo(f"\nFailed Tests ({len(failed_tests)}):")
545 for test in failed_tests:
546 severity_color = {
547 "critical": "red",
548 "high": "red",
549 "medium": "yellow",
550 "low": "white",
551 }.get(test.severity.value, "white")
553 click.secho(
554 f" [{test.severity.value.upper()}] {test.test_name}",
555 fg=severity_color,
556 )
557 click.echo(f" {test.message}")
560def _display_recommendations(recommendations: list[str]) -> None:
561 """Display health check recommendations."""
562 if not recommendations:
563 return
565 click.echo("\nRecommendations:")
566 for rec in recommendations:
567 click.echo(f" • {rec}")
570async def _save_health_report_if_requested(
571 health_checker: t.Any, report: t.Any, output: str | None
572) -> None:
573 """Save health report to file if output path is provided."""
574 if output:
575 await health_checker._save_health_report(report, Path(output))
576 click.echo(f"\nReport saved to: {output}")
579@cli.command()
580@click.argument("config_file")
581@click.option("--test-types", help="Comma-separated test types to run")
582@click.option("--output", "-o", help="Output report file")
583def health_check(config_file: str, test_types: str | None, output: str | None) -> None:
584 """Run comprehensive health check on configuration."""
586 async def _health_check() -> None:
587 registry = AdapterRegistry()
588 env_manager = EnvironmentManager()
589 health_checker = ConfigurationHealthChecker(registry, env_manager)
591 await registry.initialize()
593 config_manager = ConfigurationManager(registry)
594 await config_manager.initialize()
596 try:
597 config = await config_manager.load_configuration(config_file)
599 # Parse test types and run health check
600 test_type_list = _parse_test_types(test_types)
601 report = await health_checker.run_comprehensive_health_check(
602 config, test_type_list
603 )
605 # Display results
606 _display_health_summary(report, config_file)
608 # Show failed tests
609 failed_tests = [r for r in report.test_results if not r.passed]
610 _display_failed_tests(failed_tests)
612 # Show recommendations
613 _display_recommendations(report.recommendations)
615 # Save report if requested
616 await _save_health_report_if_requested(health_checker, report, output)
618 except Exception as e:
619 click.echo(f"Error: {e}", err=True)
621 asyncio.run(_health_check())
624if __name__ == "__main__":
625 cli()