diff --git a/cortex/benchmark.py b/cortex/benchmark.py index 92dc0382..a9812594 100644 --- a/cortex/benchmark.py +++ b/cortex/benchmark.py @@ -15,15 +15,14 @@ from dataclasses import asdict, dataclass, field from datetime import datetime from pathlib import Path -from typing import Any, List, Optional, Tuple +from typing import Any from rich import box -from rich.console import Console from rich.panel import Panel from rich.progress import BarColumn, Progress, SpinnerColumn, TextColumn, TimeElapsedColumn from rich.table import Table -from cortex.branding import CORTEX_CYAN, console, cx_header, cx_print +from cortex.branding import console, cx_header, cx_print # Model recommendations based on system capabilities MODEL_REQUIREMENTS = { @@ -118,7 +117,9 @@ def _get_system_info(self) -> dict: elif platform.system() == "Darwin": result = subprocess.run( ["sysctl", "-n", "machdep.cpu.brand_string"], - capture_output=True, text=True, timeout=5 + capture_output=True, + text=True, + timeout=5, ) if result.returncode == 0: info["cpu_model"] = result.stdout.strip() @@ -139,8 +140,7 @@ def _get_system_info(self) -> dict: break elif platform.system() == "Darwin": result = subprocess.run( - ["sysctl", "-n", "hw.memsize"], - capture_output=True, text=True, timeout=5 + ["sysctl", "-n", "hw.memsize"], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: mem_bytes = int(result.stdout.strip()) @@ -160,7 +160,9 @@ def _detect_nvidia_gpu(self) -> bool: try: result = subprocess.run( ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], - capture_output=True, text=True, timeout=10 + capture_output=True, + text=True, + timeout=10, ) return result.returncode == 0 and result.stdout.strip() != "" except Exception: @@ -171,7 +173,9 @@ def _get_nvidia_vram(self) -> int: try: result = subprocess.run( ["nvidia-smi", "--query-gpu=memory.total", "--format=csv,noheader,nounits"], - capture_output=True, text=True, timeout=10 + capture_output=True, + text=True, + timeout=10, ) if result.returncode == 0: return int(result.stdout.strip().split("\n")[0]) @@ -223,7 +227,7 @@ def _benchmark_cpu(self) -> BenchmarkResult: score=score, raw_value=round(avg_time * 1000, 2), unit="ms", - description="Matrix computation speed" + description="Matrix computation speed", ) def _benchmark_memory(self) -> BenchmarkResult: @@ -250,7 +254,7 @@ def _benchmark_memory(self) -> BenchmarkResult: # Calculate approximate bandwidth (bytes per second) bytes_processed = size * 8 * 2 # 8 bytes per int, 2 operations - bandwidth_gbps = (bytes_processed / avg_time) / (1024 ** 3) + bandwidth_gbps = (bytes_processed / avg_time) / (1024**3) # Score based on bandwidth # Baseline: 10 GB/s = 50, 50 GB/s = 100, 1 GB/s = 10 @@ -267,7 +271,7 @@ def _benchmark_memory(self) -> BenchmarkResult: score=score, raw_value=round(bandwidth_gbps, 2), unit="GB/s", - description="Memory throughput" + description="Memory throughput", ) def _benchmark_gpu(self, system_info: dict) -> BenchmarkResult: @@ -298,7 +302,7 @@ def _benchmark_gpu(self, system_info: dict) -> BenchmarkResult: score=score, raw_value=vram_mb, unit="MB", - description="NVIDIA GPU VRAM" + description="NVIDIA GPU VRAM", ) elif system_info.get("has_apple_silicon"): @@ -320,7 +324,7 @@ def _benchmark_gpu(self, system_info: dict) -> BenchmarkResult: score=score, raw_value=int(ram_gb * 1024), unit="MB (unified)", - description="Apple Silicon unified memory" + description="Apple Silicon unified memory", ) else: @@ -330,7 +334,7 @@ def _benchmark_gpu(self, system_info: dict) -> BenchmarkResult: score=15, raw_value=0, unit="MB", - description="No dedicated GPU detected" + description="No dedicated GPU detected", ) def _benchmark_inference_simulation(self) -> BenchmarkResult: @@ -348,9 +352,11 @@ def _benchmark_inference_simulation(self) -> BenchmarkResult: # Simulate embedding lookup (string hashing) embeddings = [hash(token) % 10000 for token in tokens] # Simulate attention (nested loops) - attention = sum(embeddings[i] * embeddings[j] - for i in range(min(50, len(embeddings))) - for j in range(min(50, len(embeddings)))) + attention = sum( + embeddings[i] * embeddings[j] + for i in range(min(50, len(embeddings))) + for j in range(min(50, len(embeddings))) + ) _ = attention elapsed = time.perf_counter() - start @@ -372,7 +378,7 @@ def _benchmark_inference_simulation(self) -> BenchmarkResult: score=score, raw_value=round(tokens_per_sec / 1000, 2), unit="K tok/s", - description="Simulated inference throughput" + description="Simulated inference throughput", ) def _benchmark_token_generation(self) -> BenchmarkResult: @@ -390,8 +396,10 @@ def _benchmark_token_generation(self) -> BenchmarkResult: context = [0] * 10 for _ in range(sequence_length): # Simulate softmax over vocabulary - logits = [(hash((i, tuple(context[-10:]))) % 1000) / 1000 - for i in range(min(1000, vocab_size))] + logits = [ + (hash((i, tuple(context[-10:]))) % 1000) / 1000 + for i in range(min(1000, vocab_size)) + ] next_token = max(range(len(logits)), key=lambda i: logits[i]) generated.append(next_token) context.append(next_token) @@ -415,7 +423,7 @@ def _benchmark_token_generation(self) -> BenchmarkResult: score=score, raw_value=round(tokens_per_sec, 1), unit="tok/s", - description="Simulated generation speed" + description="Simulated generation speed", ) def _calculate_overall_score(self, results: list[BenchmarkResult]) -> tuple[int, str]: @@ -579,8 +587,9 @@ def run(self, save_history: bool = True) -> BenchmarkReport: report.overall_score, report.rating = self._calculate_overall_score(report.results) # Get model recommendations - report.can_run, report.needs_upgrade, report.upgrade_suggestion = \ + report.can_run, report.needs_upgrade, report.upgrade_suggestion = ( self._get_model_recommendations(report.system_info, report.overall_score) + ) # Save to history if save_history: @@ -633,11 +642,7 @@ def display_report(self, report: BenchmarkReport): else: score_str = f"[red]{result.score}/100[/red]" - table.add_row( - result.name, - score_str, - f"{result.raw_value} {result.unit}" - ) + table.add_row(result.name, score_str, f"{result.raw_value} {result.unit}") console.print(table) console.print() @@ -650,12 +655,16 @@ def display_report(self, report: BenchmarkReport): else: score_color = "red" - score_content = f"[bold {score_color}]{report.overall_score}/100[/bold {score_color}] ({report.rating})" - console.print(Panel( - f"[bold]OVERALL SCORE:[/bold] {score_content}", - border_style="cyan", - box=box.ROUNDED, - )) + score_content = ( + f"[bold {score_color}]{report.overall_score}/100[/bold {score_color}] ({report.rating})" + ) + console.print( + Panel( + f"[bold]OVERALL SCORE:[/bold] {score_content}", + border_style="cyan", + box=box.ROUNDED, + ) + ) console.print() # Model recommendations diff --git a/cortex/branding.py b/cortex/branding.py index 84e3972c..ea85ab3f 100644 --- a/cortex/branding.py +++ b/cortex/branding.py @@ -11,8 +11,6 @@ - Consistent visual language """ -from typing import List, Optional, Tuple - from rich import box from rich.console import Console from rich.panel import Panel @@ -318,7 +316,9 @@ def cx_error(message: str) -> None: def cx_warning(message: str) -> None: """Print a warning message with warning icon.""" - console.print(f"[{CORTEX_WARNING}]⚠[/{CORTEX_WARNING}] [{CORTEX_WARNING}]{message}[/{CORTEX_WARNING}]") + console.print( + f"[{CORTEX_WARNING}]⚠[/{CORTEX_WARNING}] [{CORTEX_WARNING}]{message}[/{CORTEX_WARNING}]" + ) def cx_info(message: str) -> None: diff --git a/cortex/cli.py b/cortex/cli.py index d68d15c9..b1cfe4a1 100644 --- a/cortex/cli.py +++ b/cortex/cli.py @@ -1648,9 +1648,7 @@ def progress_callback(message: str, percent: float) -> None: "success", ) if result.duration_seconds: - console.print( - f"[dim]Completed in {result.duration_seconds:.1f}s[/dim]" - ) + console.print(f"[dim]Completed in {result.duration_seconds:.1f}s[/dim]") elif result.status == UpdateStatus.PENDING: # Dry run cx_print( @@ -2931,9 +2929,7 @@ def main(): f"[cyan]🔔 Cortex update available:[/cyan] " f"[green]{update_release.version}[/green]" ) - console.print( - " [dim]Run 'cortex update' to upgrade[/dim]" - ) + console.print(" [dim]Run 'cortex update' to upgrade[/dim]") console.print() except Exception: pass # Don't block CLI on update check failures @@ -2987,7 +2983,7 @@ def main(): nargs="?", default="status", choices=["status", "diagnose", "deps"], - help="Action: status (default), diagnose, deps" + help="Action: status (default), diagnose, deps", ) systemd_parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") @@ -2998,9 +2994,11 @@ def main(): nargs="?", default="status", choices=["status", "modes", "switch", "apps"], - help="Action: status (default), modes, switch, apps" + help="Action: status (default), modes, switch, apps", + ) + gpu_parser.add_argument( + "mode", nargs="?", help="Mode for switch action (integrated/hybrid/nvidia)" ) - gpu_parser.add_argument("mode", nargs="?", help="Mode for switch action (integrated/hybrid/nvidia)") gpu_parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") # Printer/Scanner setup command @@ -3010,7 +3008,7 @@ def main(): nargs="?", default="status", choices=["status", "detect"], - help="Action: status (default), detect" + help="Action: status (default), detect", ) printer_parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") @@ -3477,7 +3475,8 @@ def main(): help="Action to perform (default: status)", ) wifi_parser.add_argument( - "-v", "--verbose", + "-v", + "--verbose", action="store_true", help="Enable verbose output", ) @@ -3504,7 +3503,8 @@ def main(): help="Truncation mode for large input (default: middle)", ) stdin_parser.add_argument( - "-v", "--verbose", + "-v", + "--verbose", action="store_true", help="Enable verbose output", ) @@ -3524,7 +3524,8 @@ def main(): help="Package constraints (format: pkg:constraint:source)", ) deps_parser.add_argument( - "-v", "--verbose", + "-v", + "--verbose", action="store_true", help="Enable verbose output", ) @@ -3539,7 +3540,8 @@ def main(): help="Action to perform (default: check)", ) health_parser.add_argument( - "-v", "--verbose", + "-v", + "--verbose", action="store_true", help="Enable verbose output", ) @@ -3574,18 +3576,17 @@ def main(): return cli.systemd( args.service, action=getattr(args, "action", "status"), - verbose=getattr(args, "verbose", False) + verbose=getattr(args, "verbose", False), ) elif args.command == "gpu": return cli.gpu( action=getattr(args, "action", "status"), mode=getattr(args, "mode", None), - verbose=getattr(args, "verbose", False) + verbose=getattr(args, "verbose", False), ) elif args.command == "printer": return cli.printer( - action=getattr(args, "action", "status"), - verbose=getattr(args, "verbose", False) + action=getattr(args, "action", "status"), verbose=getattr(args, "verbose", False) ) elif args.command == "ask": return cli.ask(args.question) @@ -3624,25 +3625,30 @@ def main(): return cli.env(args) elif args.command == "upgrade": from cortex.licensing import open_upgrade_page + open_upgrade_page() return 0 elif args.command == "license": from cortex.licensing import show_license_status + show_license_status() return 0 elif args.command == "activate": from cortex.licensing import activate_license + return 0 if activate_license(args.license_key) else 1 elif args.command == "update": return cli.update(args) elif args.command == "wifi": from cortex.wifi_driver import run_wifi_driver + return run_wifi_driver( action=getattr(args, "action", "status"), verbose=getattr(args, "verbose", False), ) elif args.command == "stdin": from cortex.stdin_handler import run_stdin_handler + return run_stdin_handler( action=getattr(args, "action", "info"), max_lines=getattr(args, "max_lines", 1000), @@ -3651,6 +3657,7 @@ def main(): ) elif args.command == "deps": from cortex.semver_resolver import run_semver_resolver + return run_semver_resolver( action=getattr(args, "action", "analyze"), packages=getattr(args, "packages", None), @@ -3658,6 +3665,7 @@ def main(): ) elif args.command == "health": from cortex.health_score import run_health_check + return run_health_check( action=getattr(args, "action", "check"), verbose=getattr(args, "verbose", False), diff --git a/cortex/gpu_manager.py b/cortex/gpu_manager.py index 5135fb8c..8b25eeb2 100644 --- a/cortex/gpu_manager.py +++ b/cortex/gpu_manager.py @@ -7,16 +7,13 @@ Issue: #454 """ -import os import re import subprocess from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple from rich import box -from rich.console import Console from rich.panel import Panel from rich.table import Table @@ -131,12 +128,7 @@ def __init__(self, verbose: bool = False): def _run_command(self, cmd: list[str], timeout: int = 10) -> tuple[int, str, str]: """Run a command and return (returncode, stdout, stderr).""" try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=timeout - ) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) return result.returncode, result.stdout, result.stderr except FileNotFoundError: return 1, "", f"Command not found: {cmd[0]}" @@ -202,11 +194,13 @@ def _parse_lspci_line(self, line: str) -> GPUDevice | None: def _detect_nvidia_gpu(self) -> GPUDevice | None: """Detect NVIDIA GPU with detailed info.""" - returncode, stdout, _ = self._run_command([ - "nvidia-smi", - "--query-gpu=name,memory.total,power.draw", - "--format=csv,noheader,nounits" - ]) + returncode, stdout, _ = self._run_command( + [ + "nvidia-smi", + "--query-gpu=name,memory.total,power.draw", + "--format=csv,noheader,nounits", + ] + ) if returncode != 0 or not stdout.strip(): return None @@ -216,9 +210,9 @@ def _detect_nvidia_gpu(self) -> GPUDevice | None: memory = int(float(parts[1].strip())) if len(parts) > 1 else 0 # Check power state - power_returncode, power_stdout, _ = self._run_command([ - "cat", "/sys/bus/pci/devices/0000:01:00.0/power/runtime_status" - ]) + power_returncode, power_stdout, _ = self._run_command( + ["cat", "/sys/bus/pci/devices/0000:01:00.0/power/runtime_status"] + ) power_state = power_stdout.strip() if power_returncode == 0 else "unknown" return GPUDevice( @@ -278,10 +272,15 @@ def get_state(self, refresh: bool = False) -> GPUState: # Find active GPU for device in state.devices: - if device.is_active or (state.mode == GPUMode.NVIDIA and device.vendor == GPUVendor.NVIDIA): + if device.is_active or ( + state.mode == GPUMode.NVIDIA and device.vendor == GPUVendor.NVIDIA + ): state.active_gpu = device break - elif state.mode == GPUMode.INTEGRATED and device.vendor in [GPUVendor.INTEL, GPUVendor.AMD]: + elif state.mode == GPUMode.INTEGRATED and device.vendor in [ + GPUVendor.INTEL, + GPUVendor.AMD, + ]: state.active_gpu = device break @@ -347,7 +346,11 @@ def switch_mode(self, mode: GPUMode, apply: bool = False) -> tuple[bool, str, st command = f"sudo system76-power graphics {mode_map[mode]}" if not command: - return False, "No GPU switching tool found. Install prime-select, envycontrol, or system76-power.", None + return ( + False, + "No GPU switching tool found. Install prime-select, envycontrol, or system76-power.", + None, + ) if apply: # Actually run the command (would need sudo) @@ -444,12 +447,14 @@ def display_status(self): [dim]{mode_info['description']}[/dim] Battery Impact: {mode_info['impact']} """ - console.print(Panel( - mode_panel, - title="[bold cyan]GPU Mode[/bold cyan]", - border_style=CORTEX_CYAN, - padding=(1, 2), - )) + console.print( + Panel( + mode_panel, + title="[bold cyan]GPU Mode[/bold cyan]", + border_style=CORTEX_CYAN, + padding=(1, 2), + ) + ) if state.is_hybrid_system: console.print() @@ -517,11 +522,7 @@ def display_app_recommendations(self): console.print(table) -def run_gpu_manager( - action: str = "status", - mode: str | None = None, - verbose: bool = False -) -> int: +def run_gpu_manager(action: str = "status", mode: str | None = None, verbose: bool = False) -> int: """ Main entry point for cortex gpu command. diff --git a/cortex/health_score.py b/cortex/health_score.py index 8344e6aa..497859f8 100644 --- a/cortex/health_score.py +++ b/cortex/health_score.py @@ -7,15 +7,11 @@ """ import json -import os import subprocess -import time -from collections.abc import Callable from dataclasses import dataclass, field from datetime import datetime from enum import Enum from pathlib import Path -from typing import Optional from rich.console import Console from rich.panel import Panel @@ -143,9 +139,7 @@ def __init__(self, verbose: bool = False): self.verbose = verbose self.history_path = Path.home() / ".cortex" / "health_history.json" - def _run_command( - self, cmd: list[str], timeout: int = 30 - ) -> tuple[int, str, str]: + def _run_command(self, cmd: list[str], timeout: int = 30) -> tuple[int, str, str]: """Run a command and return exit code, stdout, stderr.""" try: result = subprocess.run( @@ -309,9 +303,7 @@ def check_security(self) -> HealthFactor: pass # Check for unattended upgrades - code, _, _ = self._run_command( - ["dpkg", "-l", "unattended-upgrades"] - ) + code, _, _ = self._run_command(["dpkg", "-l", "unattended-upgrades"]) if code != 0: issues.append("Automatic updates not configured") score -= 10 @@ -484,10 +476,7 @@ def save_history(self, report: HealthReport): entry = { "timestamp": report.timestamp.isoformat(), "overall_score": report.overall_score, - "factors": { - f.name: {"score": f.score, "details": f.details} - for f in report.factors - }, + "factors": {f.name: {"score": f.score, "details": f.details} for f in report.factors}, } history.append(entry) @@ -588,9 +577,7 @@ def display_history(self): else: trend = "→" - score_color = ( - "green" if score >= 75 else "yellow" if score >= 50 else "red" - ) + score_color = "green" if score >= 75 else "yellow" if score >= 50 else "red" table.add_row( ts.strftime("%Y-%m-%d %H:%M"), diff --git a/cortex/licensing.py b/cortex/licensing.py index b20f8616..714832f1 100644 --- a/cortex/licensing.py +++ b/cortex/licensing.py @@ -43,7 +43,6 @@ def level(tier: str) -> int: "parallel_ops": FeatureTier.PRO, "priority_support": FeatureTier.PRO, "usage_analytics": FeatureTier.PRO, - # Enterprise features ($99/month) "sso": FeatureTier.ENTERPRISE, "ldap": FeatureTier.ENTERPRISE, @@ -183,12 +182,15 @@ def require_feature(feature_name: str): Raises: FeatureNotAvailableError: If feature not available """ + def decorator(func): def wrapper(*args, **kwargs): if not check_feature(feature_name): raise FeatureNotAvailableError(feature_name) return func(*args, **kwargs) + return wrapper + return decorator @@ -199,7 +201,8 @@ def show_upgrade_prompt(feature: str, required_tier: str) -> None: price = "$20" if required_tier == FeatureTier.PRO else "$99" - print(f""" + print( + f""" ┌─────────────────────────────────────────────────────────┐ │ ⚡ UPGRADE REQUIRED │ ├─────────────────────────────────────────────────────────┤ @@ -213,7 +216,8 @@ def show_upgrade_prompt(feature: str, required_tier: str) -> None: │ 🌐 {PRICING_URL} │ │ └─────────────────────────────────────────────────────────┘ -""") +""" + ) def show_license_status() -> None: @@ -226,12 +230,14 @@ def show_license_status() -> None: FeatureTier.ENTERPRISE: "yellow", } - print(f""" + print( + f""" ┌─────────────────────────────────────────────────────────┐ │ CORTEX LICENSE STATUS │ ├─────────────────────────────────────────────────────────┤ │ Tier: {info.tier.upper():12} │ -│ Status: {"ACTIVE" if info.valid else "EXPIRED":12} │""") +│ Status: {"ACTIVE" if info.valid else "EXPIRED":12} │""" + ) if info.organization: print(f"│ Organization: {info.organization[:12]:12} │") @@ -280,14 +286,18 @@ def activate_license(license_key: str) -> bool: if data.get("success"): # Save license locally LICENSE_FILE.parent.mkdir(parents=True, exist_ok=True) - LICENSE_FILE.write_text(json.dumps({ - "key": license_key, - "tier": data["tier"], - "valid": True, - "expires": data.get("expires"), - "organization": data.get("organization"), - "email": data.get("email"), - })) + LICENSE_FILE.write_text( + json.dumps( + { + "key": license_key, + "tier": data["tier"], + "valid": True, + "expires": data.get("expires"), + "organization": data.get("organization"), + "email": data.get("email"), + } + ) + ) # Clear cache _cached_license = None @@ -316,6 +326,7 @@ def open_upgrade_page() -> None: def _get_hostname() -> str: """Get system hostname.""" import platform + return platform.node() diff --git a/cortex/output_formatter.py b/cortex/output_formatter.py index 476b11e1..ea1583ad 100644 --- a/cortex/output_formatter.py +++ b/cortex/output_formatter.py @@ -9,12 +9,11 @@ from collections.abc import Generator from contextlib import contextmanager -from dataclasses import dataclass, field +from dataclasses import dataclass from enum import Enum -from typing import Any, List, Optional, Tuple from rich import box -from rich.console import Console, Group +from rich.console import Console from rich.live import Live from rich.panel import Panel from rich.progress import ( @@ -28,11 +27,8 @@ TimeElapsedColumn, TimeRemainingColumn, ) -from rich.spinner import Spinner from rich.status import Status -from rich.style import Style from rich.table import Table -from rich.text import Text from rich.tree import Tree console = Console() diff --git a/cortex/printer_setup.py b/cortex/printer_setup.py index e405db98..86f16219 100644 --- a/cortex/printer_setup.py +++ b/cortex/printer_setup.py @@ -8,13 +8,10 @@ import re import subprocess -from dataclasses import dataclass, field +from dataclasses import dataclass from enum import Enum -from pathlib import Path -from typing import Dict, List, Optional, Tuple from rich import box -from rich.console import Console from rich.panel import Panel from rich.table import Table @@ -106,12 +103,7 @@ def __init__(self, verbose: bool = False): def _run_command(self, cmd: list[str], timeout: int = 30) -> tuple[int, str, str]: """Run a command and return (returncode, stdout, stderr).""" try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - timeout=timeout - ) + result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) return result.returncode, result.stdout, result.stderr except FileNotFoundError: return 1, "", f"Command not found: {cmd[0]}" @@ -161,13 +153,15 @@ def detect_usb_printers(self) -> list[PrinterDevice]: else: device_type = DeviceType.PRINTER - devices.append(PrinterDevice( - name=name, - device_type=device_type, - connection=ConnectionType.USB, - vendor=vendor, - usb_id=usb_id, - )) + devices.append( + PrinterDevice( + name=name, + device_type=device_type, + connection=ConnectionType.USB, + vendor=vendor, + usb_id=usb_id, + ) + ) return devices @@ -188,13 +182,15 @@ def detect_network_printers(self) -> list[PrinterDevice]: uri = parts[1] name = uri.split("/")[-1] if "/" in uri else uri - devices.append(PrinterDevice( - name=name, - device_type=DeviceType.PRINTER, - connection=ConnectionType.NETWORK, - uri=uri, - vendor=self._detect_vendor(name), - )) + devices.append( + PrinterDevice( + name=name, + device_type=DeviceType.PRINTER, + connection=ConnectionType.NETWORK, + uri=uri, + vendor=self._detect_vendor(name), + ) + ) return devices @@ -221,16 +217,26 @@ def detect_configured_printers(self) -> list[PrinterDevice]: parts = line.split() if len(parts) >= 2: name = parts[1] - state = "idle" if "is idle" in line else "printing" if "printing" in line else "disabled" if "disabled" in line else "unknown" - - devices.append(PrinterDevice( - name=name, - device_type=DeviceType.PRINTER, - connection=ConnectionType.UNKNOWN, - is_configured=True, - is_default=name == default_printer, - state=state, - )) + state = ( + "idle" + if "is idle" in line + else ( + "printing" + if "printing" in line + else "disabled" if "disabled" in line else "unknown" + ) + ) + + devices.append( + PrinterDevice( + name=name, + device_type=DeviceType.PRINTER, + connection=ConnectionType.UNKNOWN, + is_configured=True, + is_default=name == default_printer, + state=state, + ) + ) return devices @@ -256,14 +262,16 @@ def detect_scanners(self) -> list[PrinterDevice]: if "net:" in uri or "airscan:" in uri: connection = ConnectionType.NETWORK - devices.append(PrinterDevice( - name=name, - device_type=DeviceType.SCANNER, - connection=connection, - uri=uri, - vendor=self._detect_vendor(name), - is_configured=True, - )) + devices.append( + PrinterDevice( + name=name, + device_type=DeviceType.SCANNER, + connection=connection, + uri=uri, + vendor=self._detect_vendor(name), + is_configured=True, + ) + ) return devices @@ -360,7 +368,7 @@ def setup_printer( return False, f"Could not find driver for {device.name}" # Generate a safe printer name - printer_name = re.sub(r'[^a-zA-Z0-9_-]', '_', device.name)[:30] + printer_name = re.sub(r"[^a-zA-Z0-9_-]", "_", device.name)[:30] # Determine URI uri = device.uri @@ -379,9 +387,12 @@ def setup_printer( # Add printer cmd = [ "lpadmin", - "-p", printer_name, - "-v", uri, - "-m", driver.ppd_path, + "-p", + printer_name, + "-v", + uri, + "-m", + driver.ppd_path, "-E", # Enable ] @@ -401,10 +412,9 @@ def test_print(self, printer_name: str) -> tuple[bool, str]: return False, "CUPS is not installed" # Use CUPS test page - returncode, _, stderr = self._run_command([ - "lp", "-d", printer_name, - "/usr/share/cups/data/testprint" - ]) + returncode, _, stderr = self._run_command( + ["lp", "-d", printer_name, "/usr/share/cups/data/testprint"] + ) if returncode == 0: return True, "Test page sent to printer" @@ -454,11 +464,15 @@ def display_status(self): table.add_column("Default", style="green") for printer in configured: - status_color = "green" if printer.state == "idle" else "yellow" if printer.state == "printing" else "red" + status_color = ( + "green" + if printer.state == "idle" + else "yellow" if printer.state == "printing" else "red" + ) table.add_row( printer.name, f"[{status_color}]{printer.state}[/{status_color}]", - "✓" if printer.is_default else "" + "✓" if printer.is_default else "", ) console.print(table) @@ -469,7 +483,11 @@ def display_status(self): if usb_printers: console.print("[bold]Detected USB Devices:[/bold]") for printer in usb_printers: - icon = "🖨️" if printer.device_type == DeviceType.PRINTER else "📠" if printer.device_type == DeviceType.MULTIFUNCTION else "📷" + icon = ( + "🖨️" + if printer.device_type == DeviceType.PRINTER + else "📠" if printer.device_type == DeviceType.MULTIFUNCTION else "📷" + ) console.print(f" {icon} {printer.name} ({printer.vendor})") console.print() @@ -519,12 +537,14 @@ def display_setup_guide(self, device: PrinterDevice): if driver.recommended: content_lines.append("[green]✓ Recommended driver available[/green]") - console.print(Panel( - "\n".join(content_lines), - title="[bold cyan]Setup Information[/bold cyan]", - border_style=CORTEX_CYAN, - padding=(1, 2), - )) + console.print( + Panel( + "\n".join(content_lines), + title="[bold cyan]Setup Information[/bold cyan]", + border_style=CORTEX_CYAN, + padding=(1, 2), + ) + ) def run_printer_setup(action: str = "status", verbose: bool = False) -> int: diff --git a/cortex/semver_resolver.py b/cortex/semver_resolver.py index 27a51ca8..cec575f1 100644 --- a/cortex/semver_resolver.py +++ b/cortex/semver_resolver.py @@ -144,10 +144,7 @@ def satisfies(self, version: SemVer) -> bool: # ~1.2.3 means >=1.2.3 <1.3.0 if version < self.version: return False - return ( - version.major == self.version.major - and version.minor == self.version.minor - ) + return version.major == self.version.major and version.minor == self.version.minor elif self.constraint_type == ConstraintType.GREATER: return version > self.version @@ -203,9 +200,7 @@ def is_conflicting(self) -> bool: return True return False - def _constraints_compatible( - self, c1: VersionConstraint, c2: VersionConstraint - ) -> bool: + def _constraints_compatible(self, c1: VersionConstraint, c2: VersionConstraint) -> bool: """Check if two constraints can be satisfied simultaneously.""" if c1.constraint_type == ConstraintType.ANY: return True @@ -403,9 +398,7 @@ def parse_constraint(self, constraint_str: str) -> VersionConstraint | None: return None - def add_dependency( - self, package: str, constraint_str: str, source: str = "" - ) -> bool: + def add_dependency(self, package: str, constraint_str: str, source: str = "") -> bool: """Add a dependency constraint. Args: @@ -446,9 +439,7 @@ def detect_conflicts(self) -> list[VersionConflict]: return self.conflicts - def suggest_resolutions( - self, conflict: VersionConflict - ) -> list[ResolutionStrategy]: + def suggest_resolutions(self, conflict: VersionConflict) -> list[ResolutionStrategy]: """Suggest resolution strategies for a conflict. Args: @@ -512,9 +503,7 @@ def suggest_resolutions( return strategies - def _find_common_version_strategy( - self, conflict: VersionConflict - ) -> ResolutionStrategy | None: + def _find_common_version_strategy(self, conflict: VersionConflict) -> ResolutionStrategy | None: """Try to find a common version that satisfies all constraints.""" constraints = [d.constraint for d in conflict.dependencies] @@ -707,9 +696,7 @@ def run_semver_resolver( return 1 if constraint.satisfies(version): - console.print( - f"[green]Version {version} satisfies constraint {constraint_str}[/green]" - ) + console.print(f"[green]Version {version} satisfies constraint {constraint_str}[/green]") return 0 else: console.print( diff --git a/cortex/stdin_handler.py b/cortex/stdin_handler.py index d9e57103..bc61749c 100644 --- a/cortex/stdin_handler.py +++ b/cortex/stdin_handler.py @@ -141,11 +141,7 @@ def truncate(self, data: StdinData) -> StdinData: head = lines[:half] tail = lines[-half:] skipped = len(lines) - self.max_lines - truncated_lines = ( - head - + [f"\n... [{skipped} lines truncated] ...\n\n"] - + tail - ) + truncated_lines = head + [f"\n... [{skipped} lines truncated] ...\n\n"] + tail else: # SAMPLE step = max(1, len(lines) // self.max_lines) truncated_lines = lines[::step][: self.max_lines] @@ -155,9 +151,7 @@ def truncate(self, data: StdinData) -> StdinData: # Check byte limit content_bytes = content.encode("utf-8", errors="replace") if len(content_bytes) > self.max_bytes: - content = content_bytes[: self.max_bytes].decode( - "utf-8", errors="replace" - ) + content = content_bytes[: self.max_bytes].decode("utf-8", errors="replace") content += "\n... [truncated due to size limit] ..." new_lines = content.splitlines(keepends=True) @@ -230,21 +224,19 @@ def detect_content_type(content: str) -> str: return "json" # CSV - if "," in first_line and lines[0].count(",") == lines[1].count(",") if len(lines) > 1 else False: + if ( + "," in first_line and lines[0].count(",") == lines[1].count(",") + if len(lines) > 1 + else False + ): return "csv" # Docker/container logs - if any( - pattern in content - for pattern in ["container", "docker", "kubernetes", "pod"] - ): + if any(pattern in content for pattern in ["container", "docker", "kubernetes", "pod"]): return "container_log" # System logs - if any( - pattern in content - for pattern in ["systemd", "journald", "kernel", "syslog"] - ): + if any(pattern in content for pattern in ["systemd", "journald", "kernel", "syslog"]): return "system_log" return "text" diff --git a/cortex/systemd_helper.py b/cortex/systemd_helper.py index e837ddcb..e29d2b5b 100644 --- a/cortex/systemd_helper.py +++ b/cortex/systemd_helper.py @@ -7,16 +7,13 @@ Issue: #448 """ -import os import re import subprocess from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any, Dict, List, Optional, Tuple from rich import box -from rich.console import Console from rich.panel import Panel from rich.table import Table from rich.tree import Tree @@ -63,7 +60,10 @@ ("Verify dependencies are running", "systemctl list-dependencies {service}"), ], "signal": [ - ("Service was killed by a signal", "Check if OOM killer terminated it: dmesg | grep -i oom"), + ( + "Service was killed by a signal", + "Check if OOM killer terminated it: dmesg | grep -i oom", + ), ("Check resource limits", "systemctl show {service} | grep -i limit"), ], "timeout": [ @@ -75,8 +75,14 @@ ("Review application logs", "The application has a bug or invalid input."), ], "start-limit-hit": [ - ("Service crashed too many times", "Reset the failure count: systemctl reset-failed {service}"), - ("Fix the underlying issue", "Check logs before restarting: journalctl -u {service} -n 100"), + ( + "Service crashed too many times", + "Reset the failure count: systemctl reset-failed {service}", + ), + ( + "Fix the underlying issue", + "Check logs before restarting: journalctl -u {service} -n 100", + ), ], } @@ -154,12 +160,7 @@ def _run_systemctl(self, *args, capture: bool = True) -> tuple[int, str, str]: """Run a systemctl command and return (returncode, stdout, stderr).""" cmd = ["systemctl"] + list(args) try: - result = subprocess.run( - cmd, - capture_output=capture, - text=True, - timeout=30 - ) + result = subprocess.run(cmd, capture_output=capture, text=True, timeout=30) return result.returncode, result.stdout, result.stderr except FileNotFoundError: return 1, "", "systemctl not found. Is systemd installed?" @@ -173,7 +174,7 @@ def _run_journalctl(self, service: str, lines: int = 50) -> str: ["journalctl", "-u", service, "-n", str(lines), "--no-pager"], capture_output=True, text=True, - timeout=30 + timeout=30, ) return result.stdout except Exception: @@ -252,15 +253,17 @@ def explain_status(self, service: str) -> tuple[bool, str]: return False, f"Service '{service}' is not installed on this system." if status.load_state == "masked": - return True, f"Service '{service}' is MASKED (disabled by administrator and cannot be started)." + return ( + True, + f"Service '{service}' is MASKED (disabled by administrator and cannot be started).", + ) # Build explanation parts = [] # Main state state_explanation = SERVICE_STATE_EXPLANATIONS.get( - status.active_state, - f"in an unknown state ({status.active_state})" + status.active_state, f"in an unknown state ({status.active_state})" ) parts.append(f"**{service}** is **{status.active_state}**: {state_explanation}") @@ -328,7 +331,9 @@ def diagnose_failure(self, service: str) -> tuple[bool, str, list[str]]: # Analyze logs for common issues log_text = logs.lower() if "permission denied" in log_text: - recommendations.append("- **Permission issue detected**: Check file permissions and service user") + recommendations.append( + "- **Permission issue detected**: Check file permissions and service user" + ) if "address already in use" in log_text: recommendations.append("- **Port conflict**: Another process is using the same port") recommendations.append(" Run: `ss -tlnp | grep ` to find conflicting process") @@ -365,9 +370,9 @@ def get_dependencies(self, service: str) -> dict[str, list[str]]: service = f"{service}.service" # Get dependency info - returncode, stdout, _ = self._run_systemctl("show", service, - "-p", "Wants,Requires,After,Before,WantedBy,RequiredBy", - "--no-pager") + returncode, stdout, _ = self._run_systemctl( + "show", service, "-p", "Wants,Requires,After,Before,WantedBy,RequiredBy", "--no-pager" + ) if returncode == 0: for line in stdout.split("\n"): @@ -489,8 +494,8 @@ def create_unit_from_description( """ # Auto-generate name from description if not provided if not name: - name = re.sub(r'[^a-z0-9]+', '-', description.lower())[:40] - name = name.strip('-') + name = re.sub(r"[^a-z0-9]+", "-", description.lower())[:40] + name = name.strip("-") # Detect service type service_type = ServiceType.SIMPLE @@ -562,12 +567,14 @@ def display_status(self, service: str): console.print() success, explanation = self.explain_status(service) if success: - console.print(Panel( - explanation, - title="[bold cyan]Plain English Explanation[/bold cyan]", - border_style=CORTEX_CYAN, - padding=(1, 2), - )) + console.print( + Panel( + explanation, + title="[bold cyan]Plain English Explanation[/bold cyan]", + border_style=CORTEX_CYAN, + padding=(1, 2), + ) + ) def display_diagnosis(self, service: str): """Display failure diagnosis for a service.""" @@ -576,12 +583,14 @@ def display_diagnosis(self, service: str): found_issues, explanation, logs = self.diagnose_failure(service) if explanation: - console.print(Panel( - explanation, - title="[bold yellow]Diagnosis[/bold yellow]", - border_style="yellow", - padding=(1, 2), - )) + console.print( + Panel( + explanation, + title="[bold yellow]Diagnosis[/bold yellow]", + border_style="yellow", + padding=(1, 2), + ) + ) if logs: console.print() @@ -595,11 +604,7 @@ def display_diagnosis(self, service: str): console.print(f"[dim]{line}[/dim]") -def run_systemd_helper( - service: str, - action: str = "status", - verbose: bool = False -) -> int: +def run_systemd_helper(service: str, action: str = "status", verbose: bool = False) -> int: """ Main entry point for cortex systemd command. diff --git a/cortex/update_checker.py b/cortex/update_checker.py index 32c64e1a..31298e96 100644 --- a/cortex/update_checker.py +++ b/cortex/update_checker.py @@ -16,12 +16,7 @@ import requests -from cortex.version_manager import ( - SemanticVersion, - UpdateChannel, - get_current_version, - is_newer, -) +from cortex.version_manager import SemanticVersion, UpdateChannel, get_current_version, is_newer logger = logging.getLogger(__name__) @@ -228,8 +223,8 @@ def check(self, force: bool = False) -> UpdateCheckResult: if cached: # Update current version in case we've upgraded cached.current_version = current - cached.update_available = ( - cached.latest_version is not None and is_newer(cached.latest_version, current) + cached.update_available = cached.latest_version is not None and is_newer( + cached.latest_version, current ) return cached @@ -327,7 +322,11 @@ def _filter_by_channel(self, releases: list[ReleaseInfo]) -> list[ReleaseInfo]: if self.channel == UpdateChannel.BETA: # Stable + beta releases - return [r for r in releases if r.version.channel in (UpdateChannel.STABLE, UpdateChannel.BETA)] + return [ + r + for r in releases + if r.version.channel in (UpdateChannel.STABLE, UpdateChannel.BETA) + ] # DEV channel - all releases return releases diff --git a/cortex/updater.py b/cortex/updater.py index 5c65a314..02c39116 100644 --- a/cortex/updater.py +++ b/cortex/updater.py @@ -12,23 +12,16 @@ import shutil import subprocess import sys -import tempfile from collections.abc import Callable from dataclasses import dataclass from datetime import datetime from enum import Enum from pathlib import Path -from typing import Optional import requests from cortex.update_checker import ReleaseInfo, UpdateCheckResult, check_for_updates -from cortex.version_manager import ( - SemanticVersion, - UpdateChannel, - get_current_version, - get_version_string, -) +from cortex.version_manager import SemanticVersion, UpdateChannel, get_version_string logger = logging.getLogger(__name__) diff --git a/cortex/version_manager.py b/cortex/version_manager.py index 676c5b2e..294ee8a7 100644 --- a/cortex/version_manager.py +++ b/cortex/version_manager.py @@ -14,6 +14,7 @@ # Single source of truth for version __version__ = "0.1.0" + # Update channels class UpdateChannel(Enum): STABLE = "stable" diff --git a/cortex/wifi_driver.py b/cortex/wifi_driver.py index 0013e42d..c71480cf 100644 --- a/cortex/wifi_driver.py +++ b/cortex/wifi_driver.py @@ -190,9 +190,7 @@ def __init__(self, verbose: bool = False): self.verbose = verbose self.devices: list[WirelessDevice] = [] - def _run_command( - self, cmd: list[str], timeout: int = 30 - ) -> tuple[int, str, str]: + def _run_command(self, cmd: list[str], timeout: int = 30) -> tuple[int, str, str]: """Run a command and return exit code, stdout, stderr.""" try: result = subprocess.run( @@ -252,12 +250,8 @@ def detect_pci_devices(self) -> list[WirelessDevice]: driver = "" pci_addr = line.split()[0] if line.split() else "" if pci_addr: - _, drv_out, _ = self._run_command( - ["lspci", "-k", "-s", pci_addr] - ) - drv_match = re.search( - r"Kernel driver in use:\s*(\S+)", drv_out - ) + _, drv_out, _ = self._run_command(["lspci", "-k", "-s", pci_addr]) + drv_match = re.search(r"Kernel driver in use:\s*(\S+)", drv_out) if drv_match: driver = drv_match.group(1) @@ -447,12 +441,20 @@ def display_status(self): conn_table.add_column("Item", style="cyan") conn_table.add_column("Value") - wifi_status = "[green]Connected[/green]" if connectivity["wifi_connected"] else "[red]Not connected[/red]" + wifi_status = ( + "[green]Connected[/green]" + if connectivity["wifi_connected"] + else "[red]Not connected[/red]" + ) if connectivity["wifi_ssid"]: wifi_status += f" ({connectivity['wifi_ssid']})" conn_table.add_row("WiFi", wifi_status) - bt_status = "[green]Available[/green]" if connectivity["bluetooth_available"] else "[red]Not available[/red]" + bt_status = ( + "[green]Available[/green]" + if connectivity["bluetooth_available"] + else "[red]Not available[/red]" + ) if connectivity["bluetooth_powered"]: bt_status += " (Powered)" conn_table.add_row("Bluetooth", bt_status) @@ -597,7 +599,9 @@ def run_wifi_driver( console.print(f"WiFi: {'Connected' if status['wifi_connected'] else 'Not connected'}") if status["wifi_ssid"]: console.print(f" SSID: {status['wifi_ssid']}") - console.print(f"Bluetooth: {'Available' if status['bluetooth_available'] else 'Not available'}") + console.print( + f"Bluetooth: {'Available' if status['bluetooth_available'] else 'Not available'}" + ) return 0 else: diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py index c6bda82f..23bc2bbc 100644 --- a/tests/test_benchmark.py +++ b/tests/test_benchmark.py @@ -8,15 +8,15 @@ import os import tempfile from pathlib import Path -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.benchmark import ( - BenchmarkResult, + MODEL_REQUIREMENTS, BenchmarkReport, + BenchmarkResult, CortexBenchmark, - MODEL_REQUIREMENTS, run_benchmark, ) @@ -27,11 +27,7 @@ class TestBenchmarkResult: def test_result_creation(self): """Test creating a benchmark result.""" result = BenchmarkResult( - name="Test", - score=75, - raw_value=100.5, - unit="ms", - description="Test benchmark" + name="Test", score=75, raw_value=100.5, unit="ms", description="Test benchmark" ) assert result.name == "Test" assert result.score == 75 @@ -40,12 +36,7 @@ def test_result_creation(self): def test_result_default_description(self): """Test default description is empty.""" - result = BenchmarkResult( - name="Test", - score=50, - raw_value=10.0, - unit="s" - ) + result = BenchmarkResult(name="Test", score=50, raw_value=10.0, unit="s") assert result.description == "" @@ -64,11 +55,7 @@ def test_report_defaults(self): def test_report_to_dict(self): """Test report serialization.""" - report = BenchmarkReport( - timestamp="2025-01-01T00:00:00", - overall_score=75, - rating="Good" - ) + report = BenchmarkReport(timestamp="2025-01-01T00:00:00", overall_score=75, rating="Good") result = report.to_dict() assert result["timestamp"] == "2025-01-01T00:00:00" assert result["overall_score"] == 75 @@ -250,9 +237,7 @@ def test_save_to_history(self, benchmark): benchmark.HISTORY_FILE = Path(tmpdir) / "benchmark_history.json" report = BenchmarkReport( - timestamp="2025-01-01T00:00:00", - overall_score=75, - rating="Good" + timestamp="2025-01-01T00:00:00", overall_score=75, rating="Good" ) benchmark._save_to_history(report) @@ -299,19 +284,13 @@ def test_detect_nvidia_gpu_not_available(self, benchmark): def test_detect_nvidia_gpu_available(self, benchmark): """Test when NVIDIA GPU is detected.""" with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock( - returncode=0, - stdout="NVIDIA GeForce RTX 3080" - ) + mock_run.return_value = MagicMock(returncode=0, stdout="NVIDIA GeForce RTX 3080") assert benchmark._detect_nvidia_gpu() is True def test_get_nvidia_vram(self, benchmark): """Test getting NVIDIA VRAM.""" with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock( - returncode=0, - stdout="10240" - ) + mock_run.return_value = MagicMock(returncode=0, stdout="10240") assert benchmark._get_nvidia_vram() == 10240 diff --git a/tests/test_gpu_manager.py b/tests/test_gpu_manager.py index ce50c669..be7440b6 100644 --- a/tests/test_gpu_manager.py +++ b/tests/test_gpu_manager.py @@ -4,18 +4,18 @@ Issue: #454 - Hybrid GPU (Optimus) Manager """ -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.gpu_manager import ( + APP_GPU_RECOMMENDATIONS, + BATTERY_IMPACT, GPUDevice, GPUMode, GPUState, GPUVendor, HybridGPUManager, - BATTERY_IMPACT, - APP_GPU_RECOMMENDATIONS, run_gpu_manager, ) @@ -47,10 +47,7 @@ class TestGPUDevice: def test_default_values(self): """Test default device values.""" - device = GPUDevice( - vendor=GPUVendor.INTEL, - name="Intel HD Graphics" - ) + device = GPUDevice(vendor=GPUVendor.INTEL, name="Intel HD Graphics") assert device.vendor == GPUVendor.INTEL assert device.name == "Intel HD Graphics" assert device.driver == "" @@ -167,11 +164,7 @@ def test_run_command_not_found(self, manager): def test_run_command_success(self, manager): """Test successful command execution.""" with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock( - returncode=0, - stdout="output", - stderr="" - ) + mock_run.return_value = MagicMock(returncode=0, stdout="output", stderr="") code, stdout, stderr = manager._run_command(["test"]) assert code == 0 assert stdout == "output" @@ -298,9 +291,7 @@ def manager(self): def test_switch_mode_non_hybrid(self, manager): """Test switching on non-hybrid system.""" - state = GPUState(devices=[ - GPUDevice(vendor=GPUVendor.INTEL, name="Intel") - ]) + state = GPUState(devices=[GPUDevice(vendor=GPUVendor.INTEL, name="Intel")]) with patch.object(manager, "get_state") as mock_state: mock_state.return_value = state @@ -310,10 +301,12 @@ def test_switch_mode_non_hybrid(self, manager): def test_switch_mode_with_prime_select(self, manager): """Test switching with prime-select available.""" - state = GPUState(devices=[ - GPUDevice(vendor=GPUVendor.INTEL, name="Intel"), - GPUDevice(vendor=GPUVendor.NVIDIA, name="NVIDIA"), - ]) + state = GPUState( + devices=[ + GPUDevice(vendor=GPUVendor.INTEL, name="Intel"), + GPUDevice(vendor=GPUVendor.NVIDIA, name="NVIDIA"), + ] + ) with patch.object(manager, "get_state") as mock_state: mock_state.return_value = state diff --git a/tests/test_health_score.py b/tests/test_health_score.py index 1db8d073..890008e7 100644 --- a/tests/test_health_score.py +++ b/tests/test_health_score.py @@ -9,7 +9,7 @@ import tempfile from datetime import datetime from pathlib import Path -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest diff --git a/tests/test_printer_setup.py b/tests/test_printer_setup.py index ffe5941d..6e7eded0 100644 --- a/tests/test_printer_setup.py +++ b/tests/test_printer_setup.py @@ -4,18 +4,18 @@ Issue: #451 - Printer/Scanner Auto-Setup """ -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.printer_setup import ( + DRIVER_PACKAGES, + SCANNER_PACKAGES, ConnectionType, DeviceType, DriverInfo, PrinterDevice, PrinterSetup, - DRIVER_PACKAGES, - SCANNER_PACKAGES, run_printer_setup, ) @@ -165,7 +165,11 @@ def test_detect_usb_printers_parses_lsusb(self, setup): def test_detect_usb_printers_empty(self, setup): """Test when no printers detected.""" with patch.object(setup, "_run_command") as mock_cmd: - mock_cmd.return_value = (0, "Bus 001 Device 001: ID 1d6b:0002 Linux Foundation 2.0 root hub", "") + mock_cmd.return_value = ( + 0, + "Bus 001 Device 001: ID 1d6b:0002 Linux Foundation 2.0 root hub", + "", + ) devices = setup.detect_usb_printers() assert devices == [] diff --git a/tests/test_stdin_handler.py b/tests/test_stdin_handler.py index 9af2d488..51524ea7 100644 --- a/tests/test_stdin_handler.py +++ b/tests/test_stdin_handler.py @@ -4,10 +4,9 @@ Issue: #271 - Stdin Piping Support for Log Analysis """ -import io import json import sys -from unittest.mock import patch, MagicMock +from unittest.mock import patch import pytest @@ -188,7 +187,7 @@ def test_detect_json(self): content = '{"key": "value"}' assert detect_content_type(content) == "json" - content = '[1, 2, 3]' + content = "[1, 2, 3]" assert detect_content_type(content) == "json" def test_detect_python_traceback(self): @@ -260,7 +259,7 @@ def test_analyze_git_diff(self): def test_analyze_json_array(self): """Test JSON array analysis.""" data = StdinData( - content='[1, 2, 3, 4, 5]', + content="[1, 2, 3, 4, 5]", line_count=1, byte_count=15, ) @@ -338,9 +337,7 @@ def test_run_info_action(self, capsys): with patch.object( handler, "read_and_truncate", - return_value=StdinData( - content="test\n", line_count=1, byte_count=5 - ), + return_value=StdinData(content="test\n", line_count=1, byte_count=5), ): with patch( "cortex.stdin_handler.StdinHandler", @@ -358,9 +355,7 @@ def test_run_unknown_action(self, capsys): with patch.object( handler, "read_and_truncate", - return_value=StdinData( - content="test\n", line_count=1, byte_count=5 - ), + return_value=StdinData(content="test\n", line_count=1, byte_count=5), ): with patch( "cortex.stdin_handler.StdinHandler", @@ -380,9 +375,7 @@ def test_run_passthrough_action(self, capsys): with patch.object( handler, "read_and_truncate", - return_value=StdinData( - content="hello world", line_count=1, byte_count=11 - ), + return_value=StdinData(content="hello world", line_count=1, byte_count=11), ): with patch( "cortex.stdin_handler.StdinHandler", @@ -402,9 +395,7 @@ def test_run_stats_action(self, capsys): with patch.object( handler, "read_and_truncate", - return_value=StdinData( - content="test\n", line_count=1, byte_count=5 - ), + return_value=StdinData(content="test\n", line_count=1, byte_count=5), ): with patch( "cortex.stdin_handler.StdinHandler", @@ -484,7 +475,7 @@ def test_read_error(self): handler = StdinHandler() with patch("sys.stdin.isatty", return_value=False): - with patch("sys.stdin.read", side_effect=IOError("Read error")): + with patch("sys.stdin.read", side_effect=OSError("Read error")): data = handler.read_stdin() assert data.is_empty diff --git a/tests/test_systemd_helper.py b/tests/test_systemd_helper.py index 681a28ed..4ee3491c 100644 --- a/tests/test_systemd_helper.py +++ b/tests/test_systemd_helper.py @@ -4,18 +4,18 @@ Issue: #448 - Systemd Service Helper (Plain English) """ -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.systemd_helper import ( + FAILURE_SOLUTIONS, + SERVICE_STATE_EXPLANATIONS, + SUB_STATE_EXPLANATIONS, ServiceConfig, ServiceStatus, ServiceType, SystemdHelper, - SERVICE_STATE_EXPLANATIONS, - SUB_STATE_EXPLANATIONS, - FAILURE_SOLUTIONS, run_systemd_helper, ) @@ -25,11 +25,7 @@ class TestServiceConfig: def test_default_values(self): """Test default configuration values.""" - config = ServiceConfig( - name="test", - description="Test service", - exec_start="/usr/bin/test" - ) + config = ServiceConfig(name="test", description="Test service", exec_start="/usr/bin/test") assert config.name == "test" assert config.service_type == ServiceType.SIMPLE assert config.restart == "on-failure" @@ -137,11 +133,7 @@ def test_run_systemctl_not_found(self, helper): def test_run_systemctl_success(self, helper): """Test successful systemctl command.""" with patch("subprocess.run") as mock_run: - mock_run.return_value = MagicMock( - returncode=0, - stdout="ActiveState=active", - stderr="" - ) + mock_run.return_value = MagicMock(returncode=0, stdout="ActiveState=active", stderr="") code, stdout, stderr = helper._run_systemctl("status", "test") assert code == 0 assert "active" in stdout.lower() @@ -149,6 +141,7 @@ def test_run_systemctl_success(self, helper): def test_run_systemctl_timeout(self, helper): """Test systemctl timeout handling.""" import subprocess + with patch("subprocess.run") as mock_run: mock_run.side_effect = subprocess.TimeoutExpired("cmd", 30) code, stdout, stderr = helper._run_systemctl("status", "test") diff --git a/tests/test_update_checker.py b/tests/test_update_checker.py index b9026a8a..7356e084 100644 --- a/tests/test_update_checker.py +++ b/tests/test_update_checker.py @@ -13,8 +13,8 @@ from cortex.update_checker import ( CACHE_TTL_SECONDS, ReleaseInfo, - UpdateCheckResult, UpdateChecker, + UpdateCheckResult, check_for_updates, should_notify_update, ) @@ -99,9 +99,7 @@ def setUp(self): """Set up test fixtures.""" # Use temp directory for cache self.temp_dir = tempfile.mkdtemp() - self.cache_patch = patch( - "cortex.update_checker.CACHE_DIR", Path(self.temp_dir) - ) + self.cache_patch = patch("cortex.update_checker.CACHE_DIR", Path(self.temp_dir)) self.cache_patch.start() self.cache_file_patch = patch( "cortex.update_checker.UPDATE_CACHE_FILE", diff --git a/tests/test_updater.py b/tests/test_updater.py index 9219d9c3..1b8b0eb9 100644 --- a/tests/test_updater.py +++ b/tests/test_updater.py @@ -14,9 +14,9 @@ from cortex.update_checker import ReleaseInfo, UpdateCheckResult from cortex.updater import ( BackupInfo, + Updater, UpdateResult, UpdateStatus, - Updater, download_with_progress, verify_checksum, ) diff --git a/tests/test_wifi_driver.py b/tests/test_wifi_driver.py index 530ebacf..c52df261 100644 --- a/tests/test_wifi_driver.py +++ b/tests/test_wifi_driver.py @@ -4,19 +4,19 @@ Issue: #444 - WiFi/Bluetooth Driver Auto-Matcher """ -from unittest.mock import patch, MagicMock +from unittest.mock import MagicMock, patch import pytest from cortex.wifi_driver import ( + BLUETOOTH_DRIVERS, + DRIVER_DATABASE, ConnectionType, DeviceType, DriverInfo, DriverSource, WirelessDevice, WirelessDriverMatcher, - DRIVER_DATABASE, - BLUETOOTH_DRIVERS, run_wifi_driver, )