From 32bf77e351dff39d37dcb8217f3be1096648b939 Mon Sep 17 00:00:00 2001 From: bradjin8 Date: Wed, 8 Jul 2026 15:19:03 -0400 Subject: [PATCH 1/3] initial implementation of #68 --- .pre-commit-config.yaml | 2 +- cli/localci/core/orchestrator.py | 23 +++++++++++++----- cli/localci/core/progress.py | 25 ++++++++++--------- cli/localci/core/queue_builder.py | 8 +++---- cli/localci/core/registry.py | 8 +++---- cli/localci/core/results.py | 3 ++- cli/localci/core/serialization.py | 8 +++---- cli/localci/core/workflow.py | 20 +++++++++------- cli/localci/utils/docker.py | 3 ++- cli/localci/utils/resources.py | 17 ++++++------- cli/localci/utils/yq.py | 40 ++++++++++++++++++------------- cli/pyproject.toml | 20 ++++------------ 12 files changed, 96 insertions(+), 81 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index cd0ee9e..6856f21 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -33,7 +33,7 @@ repos: hooks: - id: mypy-cli name: mypy (cli) - entry: bash -c 'cd cli && mypy localci/' + entry: python -c "import pathlib, subprocess, sys; sys.exit(subprocess.call([sys.executable, '-m', 'mypy', 'localci/'], cwd=pathlib.Path('.').resolve() / 'cli'))" language: python pass_filenames: false always_run: true diff --git a/cli/localci/core/orchestrator.py b/cli/localci/core/orchestrator.py index b3c8e64..8895498 100644 --- a/cli/localci/core/orchestrator.py +++ b/cli/localci/core/orchestrator.py @@ -16,7 +16,7 @@ from enum import Enum from pathlib import Path from threading import Event as ThreadEvent -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from localci.core.cmake_cache import compute_cmake_input_digest from localci.core.command_builder import ActCommandBuilder @@ -174,7 +174,7 @@ def __init__( self._state = OrchestratorState.IDLE self._run: ExecutionRun | None = None self._pool: ThreadPoolExecutor | None = None - self._futures: dict[str, Future] = {} + self._futures: dict[str, Future[JobResult]] = {} self._shutdown_event = ThreadEvent() self._listeners: list[Callable[[JobEvent], None]] = [] @@ -291,9 +291,12 @@ def _dispatch_loop(self) -> None: time.sleep(self.config.dispatch_interval) continue logger.info("Dispatching: %s", job.matrix_entry.name) - future = self._pool.submit(self._execute_job, job) + pool = self._pool + if pool is None: + continue + future = pool.submit(self._execute_job, job) self._futures[job.queue_key] = future - future.add_done_callback(lambda f, j=job: self._on_job_done(j, f)) + future.add_done_callback(self._make_done_callback(job)) def _wait_for_completion(self) -> None: for key, future in list(self._futures.items()): @@ -549,7 +552,15 @@ def _prepare_image(self, job: QueuedJob) -> str | None: ) return None - def _on_job_done(self, job: QueuedJob, future: Future) -> None: + def _make_done_callback( + self, job: QueuedJob + ) -> Callable[[Future[JobResult]], None]: + def callback(future: Future[JobResult]) -> None: + self._on_job_done(job, future) + + return callback + + def _on_job_done(self, job: QueuedJob, future: Future[JobResult]) -> None: try: result = future.result() except Exception as e: @@ -635,7 +646,7 @@ def execution_id(self) -> str | None: def active_workers(self) -> int: return len([f for f in self._futures.values() if not f.done()]) - def get_status(self) -> dict: + def get_status(self) -> dict[str, Any]: if not self._run: return {"state": "idle"} snap = self._resource_monitor.snapshot() diff --git a/cli/localci/core/progress.py b/cli/localci/core/progress.py index a0bd714..020b74f 100644 --- a/cli/localci/core/progress.py +++ b/cli/localci/core/progress.py @@ -15,7 +15,7 @@ from dataclasses import dataclass, field from datetime import datetime from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from localci.core.executor import JobResult from localci.core.models import ( @@ -25,6 +25,8 @@ ) if TYPE_CHECKING: + from rich.live import Live + from localci.core.orchestrator import ExecutionRun from localci.core.queue import PriorityJobQueue @@ -209,7 +211,7 @@ def __init__( self._jobs: dict[str, JobProgress] = {} self._started_at: datetime | None = None self._execution_id: str | None = None - self._live = None # Rich Live instance + self._live: Live | None = None self._completed_durations: list[float] = [] self._last_status_write: float = 0.0 self._status_write_interval: float = 1.0 @@ -392,12 +394,13 @@ def start_live(self) -> None: """Start Rich Live display.""" from rich.live import Live - self._live = Live( + live = Live( self._render_live(), refresh_per_second=4, transient=True, ) - self._live.start() + self._live = live + live.start() def stop_live(self) -> None: """Stop Rich Live display.""" @@ -405,18 +408,18 @@ def stop_live(self) -> None: self._live.stop() self._live = None - def _render_live(self): + def _render_live(self) -> Any: """Render the complete live display.""" from rich.console import Group - parts = [] + parts: list[Any] = [] parts.append(self._render_header()) parts.append(self._render_progress_bar()) parts.append(self._render_priority_levels()) parts.append(self._render_job_table()) return Group(*parts) - def _render_header(self): + def _render_header(self) -> Any: """Render execution header panel.""" from rich.panel import Panel @@ -433,7 +436,7 @@ def _render_header(self): ) return Panel(content, title="Local CI Execution", expand=True) - def _render_progress_bar(self): + def _render_progress_bar(self) -> Any: """Render progress bar with ETA.""" from rich.text import Text @@ -451,7 +454,7 @@ def _render_progress_bar(self): f"Elapsed: {self._elapsed_display()} {eta_str}\n" ) - def _render_priority_levels(self): + def _render_priority_levels(self) -> Any: """Render priority level summary.""" from rich.text import Text @@ -468,7 +471,7 @@ def _render_priority_levels(self): lines.append(f"Priority {level.priority} {icon} {label}{suffix}") return Text("\n".join(lines) + "\n") - def _render_job_table(self): + def _render_job_table(self) -> Any: """Render per-job status table with current step when running.""" from rich.table import Table @@ -623,7 +626,7 @@ def print_summary(self, run: ExecutionRun) -> None: # JSON status (for localci status --format json) # ----------------------------------------------------------------------- - def get_status_dict(self) -> dict: + def get_status_dict(self) -> dict[str, Any]: """Get structured status for JSON output and status files.""" with self._lock: jobs = list(self._jobs.values()) diff --git a/cli/localci/core/queue_builder.py b/cli/localci/core/queue_builder.py index 1c34c82..da39b4e 100644 --- a/cli/localci/core/queue_builder.py +++ b/cli/localci/core/queue_builder.py @@ -5,7 +5,7 @@ import fnmatch import logging from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from localci.core.image_tag import derive_image_tag from localci.core.models import QueuedJob @@ -40,7 +40,7 @@ def _resolve_image_tag_and_build( return derived_tag, None, True -def _matches_filter(entry: MatrixEntry, filters: list[dict]) -> bool: +def _matches_filter(entry: MatrixEntry, filters: list[dict[str, Any]]) -> bool: """True if entry matches any of the filter dicts.""" for f in filters: match = True @@ -96,8 +96,8 @@ def build( platform_filter: Platform | None = None, job_filter: list[str] | None = None, compiler_filter: str | None = None, - matrix_include: list[dict] | None = None, - matrix_exclude: list[dict] | None = None, + matrix_include: list[dict[str, Any]] | None = None, + matrix_exclude: list[dict[str, Any]] | None = None, entries_include: set[tuple[str, int]] | None = None, registry_path: Path | None = None, ) -> PriorityJobQueue: diff --git a/cli/localci/core/registry.py b/cli/localci/core/registry.py index df32f7b..a4b94d0 100644 --- a/cli/localci/core/registry.py +++ b/cli/localci/core/registry.py @@ -47,7 +47,7 @@ class RegistryEntry: usage_count: int = 0 # Optional fields from existing registry (preserved on load/save) variants: list[str] = field(default_factory=list) - raw: dict = field(default_factory=dict) + raw: dict[str, Any] = field(default_factory=dict) @classmethod def from_dict(cls, d: dict[str, Any]) -> RegistryEntry: @@ -257,9 +257,9 @@ def select_image( ) best_ess = max(s[0] for s in scored) - candidates = [(ext, r) for ess, ext, r in scored if ess == best_ess] - best_ext = max(e for e, _ in candidates) - base_candidates = [r for e, r in candidates if e == best_ext] + scored_candidates = [(ext, r) for ess, ext, r in scored if ess == best_ess] + best_ext = max(e for e, _ in scored_candidates) + base_candidates = [r for e, r in scored_candidates if e == best_ext] best = base_candidates[0] for other in base_candidates[1:]: best = _tie_break(best, other) diff --git a/cli/localci/core/results.py b/cli/localci/core/results.py index a186ded..42611bf 100644 --- a/cli/localci/core/results.py +++ b/cli/localci/core/results.py @@ -10,6 +10,7 @@ from dataclasses import dataclass, field from datetime import datetime from pathlib import Path +from typing import Any from localci.core.executor import JobResult, JobStatus @@ -151,7 +152,7 @@ def summary_report(self) -> str: # Persistence # ----------------------------------------------------------------- - def to_dict(self) -> dict: + def to_dict(self) -> dict[str, Any]: """Serialise to a plain dict (JSON-safe).""" return { "execution_id": self.execution_id, diff --git a/cli/localci/core/serialization.py b/cli/localci/core/serialization.py index 2e4a29c..c281814 100644 --- a/cli/localci/core/serialization.py +++ b/cli/localci/core/serialization.py @@ -11,7 +11,7 @@ from dataclasses import asdict from enum import Enum from pathlib import Path -from typing import Any +from typing import Any, cast from localci.core.workflow import ( Workflow, @@ -52,12 +52,12 @@ def workflow_to_json(workflow: Workflow, indent: int = 2) -> str: return json.dumps(workflow, cls=WorkflowEncoder, indent=indent) -def workflow_to_dict(workflow: Workflow) -> dict: +def workflow_to_dict(workflow: Workflow) -> dict[str, Any]: """Convert *workflow* to a plain ``dict`` (JSON round-trip).""" - return json.loads(workflow_to_json(workflow)) + return cast(dict[str, Any], json.loads(workflow_to_json(workflow))) -def workflow_summary(workflow: Workflow) -> dict: +def workflow_summary(workflow: Workflow) -> dict[str, Any]: """Generate a concise summary dict for *workflow*.""" return { "name": workflow.name, diff --git a/cli/localci/core/workflow.py b/cli/localci/core/workflow.py index aa5f687..5810da8 100644 --- a/cli/localci/core/workflow.py +++ b/cli/localci/core/workflow.py @@ -199,7 +199,7 @@ class MatrixEntry: is_latest: bool = False is_earliest: bool = False timeout_minutes: int = 120 - raw: dict = field(default_factory=dict) + raw: dict[str, Any] = field(default_factory=dict) @property def image_requirements_key(self) -> str: @@ -247,12 +247,12 @@ class Job: container: ContainerInfo | None = None needs: list[str] = field(default_factory=list) condition: str | None = None - strategy: dict | None = None + strategy: dict[str, Any] | None = None matrix: list[MatrixEntry] = field(default_factory=list) steps: list[StepInfo] = field(default_factory=list) timeout_minutes: int = 60 env: dict[str, str] = field(default_factory=dict) - defaults: dict | None = None + defaults: dict[str, Any] | None = None @property def has_matrix(self) -> bool: @@ -277,7 +277,7 @@ class Workflow: file_path: Path events: list[str] env: dict[str, str] = field(default_factory=dict) - concurrency: dict | None = None + concurrency: dict[str, Any] | None = None jobs: dict[str, Job] = field(default_factory=dict) @property @@ -556,7 +556,7 @@ def _parse_job(self, file: Path, job_id: str) -> Job: # ----------------------------------------------------------------- def _parse_matrix_entry( - self, index: int, entry: dict, job_data: dict + self, index: int, entry: dict[str, Any], job_data: dict[str, Any] ) -> MatrixEntry: """Parse a single matrix include entry.""" compiler = self._parse_compiler(entry) @@ -603,7 +603,7 @@ def _parse_matrix_entry( # Field parsers # ----------------------------------------------------------------- - def _parse_compiler(self, entry: dict) -> CompilerInfo: + def _parse_compiler(self, entry: dict[str, Any]) -> CompilerInfo: family_str = entry.get("compiler", "unknown") family = self._classify_compiler(family_str) return CompilerInfo( @@ -628,7 +628,7 @@ def _parse_container(self, container_data: Any) -> ContainerInfo: ) return ContainerInfo() - def _parse_packages(self, entry: dict) -> PackageRequirements: + def _parse_packages(self, entry: dict[str, Any]) -> PackageRequirements: install_str = entry.get("install", "") apt_packages = ( [p.strip() for p in install_str.split() if p.strip()] if install_str else [] @@ -650,7 +650,7 @@ def _parse_packages(self, entry: dict) -> PackageRequirements: ccflags=entry.get("ccflags"), ) - def _parse_step(self, step: dict) -> StepInfo: + def _parse_step(self, step: dict[str, Any]) -> StepInfo: return StepInfo( name=step.get("name", ""), uses=step.get("uses"), @@ -697,7 +697,9 @@ def _classify_platform(self, runs_on: str, container: ContainerInfo) -> Platform def _classify_compiler(self, compiler_str: str) -> CompilerFamily: return self._COMPILER_MAP.get(compiler_str.lower(), CompilerFamily.UNKNOWN) - def _detect_build_system(self, entry: dict, job_data: dict) -> BuildSystem: + def _detect_build_system( + self, entry: dict[str, Any], job_data: dict[str, Any] + ) -> BuildSystem: has_b2 = False has_cmake = False diff --git a/cli/localci/utils/docker.py b/cli/localci/utils/docker.py index ea4c300..9ab8346 100644 --- a/cli/localci/utils/docker.py +++ b/cli/localci/utils/docker.py @@ -11,6 +11,7 @@ import subprocess import time from pathlib import Path +from typing import Any from localci.errors import DockerNotAvailableError @@ -204,7 +205,7 @@ def cleanup_act_containers(self) -> int: # Resource monitoring # ----------------------------------------------------------------- - def disk_usage(self) -> dict: + def disk_usage(self) -> dict[str, Any]: """Get Docker disk usage summary.""" result = subprocess.run( self._docker_cmd("system", "df", "--format", "{{json .}}"), diff --git a/cli/localci/utils/resources.py b/cli/localci/utils/resources.py index fc9858e..45a087b 100644 --- a/cli/localci/utils/resources.py +++ b/cli/localci/utils/resources.py @@ -11,6 +11,7 @@ import subprocess from dataclasses import dataclass from datetime import datetime +from typing import Any logger = logging.getLogger(__name__) @@ -51,9 +52,9 @@ class ResourceMonitor: """ def __init__(self) -> None: - self._psutil: object | None = self._try_import_psutil() + self._psutil: Any = self._try_import_psutil() - def _try_import_psutil(self) -> object | None: + def _try_import_psutil(self) -> Any: try: import psutil @@ -109,23 +110,23 @@ def check_thresholds( return ok, warnings def _get_cpu(self) -> float: - if self._psutil: - return self._psutil.cpu_percent(interval=0.1) + if self._psutil is not None: + return float(self._psutil.cpu_percent(interval=0.1)) return 50.0 def _get_memory(self) -> tuple[float, float]: - if self._psutil: + if self._psutil is not None: mem = self._psutil.virtual_memory() - return mem.percent, mem.available / (1024**3) + return float(mem.percent), mem.available / (1024**3) return 50.0, 8.0 def _get_disk_free(self) -> float: - if self._psutil: + if self._psutil is not None: if platform.system() == "Windows": usage = self._psutil.disk_usage("C:\\") else: usage = self._psutil.disk_usage("/") - return usage.free / (1024**3) + return float(usage.free / (1024**3)) return 50.0 def _get_container_count(self) -> int: diff --git a/cli/localci/utils/yq.py b/cli/localci/utils/yq.py index ff5adee..53bd4b1 100644 --- a/cli/localci/utils/yq.py +++ b/cli/localci/utils/yq.py @@ -50,7 +50,7 @@ class YqWrapper: def __init__(self) -> None: self._is_linux: bool = sys.platform.startswith("linux") - self._file_cache: dict[Path, dict] = {} + self._file_cache: dict[Path, dict[str, Any]] = {} raw_path: str | None = shutil.which("yq") self._yq_path: str | None = None @@ -164,8 +164,10 @@ def query(self, file: Path, expression: str) -> Any: def _query_yq(self, file: Path, expression: str) -> Any: """Execute *expression* via ``yq -o json``.""" + yq_path = self._yq_path + assert yq_path is not None result = subprocess.run( - [self._yq_path, "-o", "json", expression, str(file)], + [yq_path, "-o", "json", expression, str(file)], capture_output=True, text=True, timeout=30, @@ -188,7 +190,7 @@ def _query_yq(self, file: Path, expression: str) -> Any: # PyYAML fallback backend # ----------------------------------------------------------------- - def _load_yaml(self, file: Path) -> dict: + def _load_yaml(self, file: Path) -> dict[str, Any]: """Load *file* via PyYAML (result is cached). Caller must ensure file exists (e.g. query() checks before calling). @@ -342,16 +344,17 @@ def global_env(self, file: Path) -> dict[str, str]: return {str(k): str(v) for k, v in env.items()} return {} - def concurrency(self, file: Path) -> dict | None: + def concurrency(self, file: Path) -> dict[str, Any] | None: """Extract concurrency configuration.""" - return self.query(file, ".concurrency") + result = self.query(file, ".concurrency") + return result if isinstance(result, dict) else None def job_names(self, file: Path) -> list[str]: """Extract all job IDs.""" result = self.query(file, "(.jobs // {}) | keys") return result if isinstance(result, list) else [] - def job_data(self, file: Path, job_id: str) -> dict: + def job_data(self, file: Path, job_id: str) -> dict[str, Any]: """Extract full job data dict for *job_id*.""" result = self.query(file, f".jobs.{job_id}") return result if isinstance(result, dict) else {} @@ -367,26 +370,28 @@ def job_needs(self, file: Path, job_id: str) -> list[str]: def job_condition(self, file: Path, job_id: str) -> str | None: """Extract job ``if`` condition.""" - return self.query(file, f".jobs.{job_id}.if") + result = self.query(file, f".jobs.{job_id}.if") + return result if isinstance(result, str) else None def job_runs_on(self, file: Path, job_id: str) -> str: """Extract job ``runs-on``.""" return self.query(file, f".jobs.{job_id}.runs-on") or "ubuntu-latest" - def job_container(self, file: Path, job_id: str) -> dict | None: + def job_container(self, file: Path, job_id: str) -> dict[str, Any] | None: """Extract job container configuration.""" container = self.query(file, f".jobs.{job_id}.container") if isinstance(container, str): return {"image": container} - return container + return container if isinstance(container, dict) else None def job_timeout(self, file: Path, job_id: str) -> int: """Extract job timeout in minutes.""" return self.query(file, f".jobs.{job_id}.timeout-minutes") or 60 - def job_defaults(self, file: Path, job_id: str) -> dict | None: + def job_defaults(self, file: Path, job_id: str) -> dict[str, Any] | None: """Extract job defaults.""" - return self.query(file, f".jobs.{job_id}.defaults") + result = self.query(file, f".jobs.{job_id}.defaults") + return result if isinstance(result, dict) else None def job_env(self, file: Path, job_id: str) -> dict[str, str]: """Extract job environment variables.""" @@ -395,16 +400,17 @@ def job_env(self, file: Path, job_id: str) -> dict[str, str]: return {str(k): str(v) for k, v in env.items()} return {} - def matrix_strategy(self, file: Path, job_id: str) -> dict | None: + def matrix_strategy(self, file: Path, job_id: str) -> dict[str, Any] | None: """Extract matrix strategy.""" - return self.query(file, f".jobs.{job_id}.strategy") + result = self.query(file, f".jobs.{job_id}.strategy") + return result if isinstance(result, dict) else None - def matrix_include(self, file: Path, job_id: str) -> list[dict]: + def matrix_include(self, file: Path, job_id: str) -> list[dict[str, Any]]: """Extract matrix include entries.""" result = self.query(file, f".jobs.{job_id}.strategy.matrix.include") return result if isinstance(result, list) else [] - def matrix_entry(self, file: Path, job_id: str, index: int) -> dict: + def matrix_entry(self, file: Path, job_id: str, index: int) -> dict[str, Any]: """Extract specific matrix entry by *index*.""" result = self.query(file, f".jobs.{job_id}.strategy.matrix.include[{index}]") return result if isinstance(result, dict) else {} @@ -416,13 +422,13 @@ def matrix_count(self, file: Path, job_id: str) -> int: def matrix_filter_by_field( self, file: Path, job_id: str, field_name: str, value: str - ) -> list[dict]: + ) -> list[dict[str, Any]]: """Filter matrix entries by a field value.""" # For yq we can use select(); for fallback, use Python filtering entries = self.matrix_include(file, job_id) return [e for e in entries if str(e.get(field_name, "")) == value] - def steps(self, file: Path, job_id: str) -> list[dict]: + def steps(self, file: Path, job_id: str) -> list[dict[str, Any]]: """Extract job steps.""" result = self.query(file, f".jobs.{job_id}.steps") return result if isinstance(result, list) else [] diff --git a/cli/pyproject.toml b/cli/pyproject.toml index fceb7ae..fc9f2d9 100644 --- a/cli/pyproject.toml +++ b/cli/pyproject.toml @@ -25,7 +25,7 @@ dev = [ "pytest-cov>=4.0.0", "ruff>=0.5", "types-PyYAML>=6.0.0", - "types-psutil>=5.9.0", # for localci.utils.resources when it leaves ignore_errors + "types-psutil>=5.9.0", ] [project.scripts] @@ -42,10 +42,10 @@ markers = [ ] norecursedirs = ["integration"] -# Mypy baseline (2026-06-09): 85 errors in 18 legacy modules suppressed via -# per-module ignore_errors overrides below. localci.core.config and -# localci.core.models pass strict mypy with zero errors. New modules must not -# use ignore_errors; remove overrides as legacy modules are typed. +# Mypy baseline: legacy CLI modules suppressed via per-module ignore_errors +# overrides below. localci.core.* and localci.utils.* pass strict mypy with +# zero errors. New modules must not use ignore_errors; remove overrides as +# legacy CLI modules are typed. [tool.mypy] python_version = "3.10" strict = true @@ -64,16 +64,6 @@ module = [ "localci.cli.run.patcher", "localci.cli.run.run_flow", "localci.cli.status", - "localci.core.orchestrator", - "localci.core.progress", - "localci.core.queue_builder", - "localci.core.registry", - "localci.core.results", - "localci.core.serialization", - "localci.core.workflow", - "localci.utils.docker", - "localci.utils.resources", - "localci.utils.yq", ] ignore_errors = true From 2fb526b0f0d03cfeffda7ac6c7f6b9a1e4e6e4d7 Mon Sep 17 00:00:00 2001 From: bradjin8 Date: Wed, 8 Jul 2026 15:37:24 -0400 Subject: [PATCH 2/3] fix: nitpick comments --- cli/localci/core/orchestrator.py | 16 ++++++++++++++++ cli/localci/core/progress.py | 13 +++++++------ cli/localci/utils/yq.py | 19 +++++++++++++++++-- 3 files changed, 40 insertions(+), 8 deletions(-) diff --git a/cli/localci/core/orchestrator.py b/cli/localci/core/orchestrator.py index 8895498..09ce0ac 100644 --- a/cli/localci/core/orchestrator.py +++ b/cli/localci/core/orchestrator.py @@ -293,6 +293,22 @@ def _dispatch_loop(self) -> None: logger.info("Dispatching: %s", job.matrix_entry.name) pool = self._pool if pool is None: + logger.error( + "Thread pool unavailable; marking job as failed: %s", + job.matrix_entry.name, + ) + failed = Future[JobResult]() + failed.set_result( + JobResult( + job_id=job.job_id, + matrix_index=job.matrix_entry.index, + matrix_name=job.matrix_entry.name, + status=JobStatus.ERROR, + error_message="Orchestrator thread pool unavailable", + ) + ) + self._futures[job.queue_key] = failed + self._on_job_done(job, failed) continue future = pool.submit(self._execute_job, job) self._futures[job.queue_key] = future diff --git a/cli/localci/core/progress.py b/cli/localci/core/progress.py index 020b74f..51df614 100644 --- a/cli/localci/core/progress.py +++ b/cli/localci/core/progress.py @@ -25,6 +25,7 @@ ) if TYPE_CHECKING: + from rich.console import RenderableType from rich.live import Live from localci.core.orchestrator import ExecutionRun @@ -408,18 +409,18 @@ def stop_live(self) -> None: self._live.stop() self._live = None - def _render_live(self) -> Any: + def _render_live(self) -> RenderableType: """Render the complete live display.""" from rich.console import Group - parts: list[Any] = [] + parts: list[RenderableType] = [] parts.append(self._render_header()) parts.append(self._render_progress_bar()) parts.append(self._render_priority_levels()) parts.append(self._render_job_table()) return Group(*parts) - def _render_header(self) -> Any: + def _render_header(self) -> RenderableType: """Render execution header panel.""" from rich.panel import Panel @@ -436,7 +437,7 @@ def _render_header(self) -> Any: ) return Panel(content, title="Local CI Execution", expand=True) - def _render_progress_bar(self) -> Any: + def _render_progress_bar(self) -> RenderableType: """Render progress bar with ETA.""" from rich.text import Text @@ -454,7 +455,7 @@ def _render_progress_bar(self) -> Any: f"Elapsed: {self._elapsed_display()} {eta_str}\n" ) - def _render_priority_levels(self) -> Any: + def _render_priority_levels(self) -> RenderableType: """Render priority level summary.""" from rich.text import Text @@ -471,7 +472,7 @@ def _render_priority_levels(self) -> Any: lines.append(f"Priority {level.priority} {icon} {label}{suffix}") return Text("\n".join(lines) + "\n") - def _render_job_table(self) -> Any: + def _render_job_table(self) -> RenderableType: """Render per-job status table with current step when running.""" from rich.table import Table diff --git a/cli/localci/utils/yq.py b/cli/localci/utils/yq.py index 53bd4b1..2cad755 100644 --- a/cli/localci/utils/yq.py +++ b/cli/localci/utils/yq.py @@ -375,7 +375,10 @@ def job_condition(self, file: Path, job_id: str) -> str | None: def job_runs_on(self, file: Path, job_id: str) -> str: """Extract job ``runs-on``.""" - return self.query(file, f".jobs.{job_id}.runs-on") or "ubuntu-latest" + result = self.query(file, f".jobs.{job_id}.runs-on") + if isinstance(result, str): + return result + return "ubuntu-latest" def job_container(self, file: Path, job_id: str) -> dict[str, Any] | None: """Extract job container configuration.""" @@ -386,7 +389,19 @@ def job_container(self, file: Path, job_id: str) -> dict[str, Any] | None: def job_timeout(self, file: Path, job_id: str) -> int: """Extract job timeout in minutes.""" - return self.query(file, f".jobs.{job_id}.timeout-minutes") or 60 + result = self.query(file, f".jobs.{job_id}.timeout-minutes") + if isinstance(result, bool): + return 60 + if isinstance(result, int): + return result + if isinstance(result, float): + return int(result) + if isinstance(result, str): + try: + return int(result) + except ValueError: + return 60 + return 60 def job_defaults(self, file: Path, job_id: str) -> dict[str, Any] | None: """Extract job defaults.""" From fd9a720885bd3dc4c0b9c0bb5e88ae9b357b4f97 Mon Sep 17 00:00:00 2001 From: bradjin8 Date: Thu, 9 Jul 2026 13:09:30 -0400 Subject: [PATCH 3/3] fix: added small unit tests --- cli/tests/test_orchestrator.py | 41 +++++++++++++++++++++++++++++++++- 1 file changed, 40 insertions(+), 1 deletion(-) diff --git a/cli/tests/test_orchestrator.py b/cli/tests/test_orchestrator.py index 0106bb7..078f887 100644 --- a/cli/tests/test_orchestrator.py +++ b/cli/tests/test_orchestrator.py @@ -1,14 +1,16 @@ """Tests for parallel execution manager (Issue 7).""" import time +from datetime import datetime from pathlib import Path from unittest.mock import MagicMock, patch from localci.core.executor import JobResult, JobStatus -from localci.core.models import JobEventType +from localci.core.models import JobEventType, QueuedJobStatus from localci.core.orchestrator import ( ExecutionRun, OrchestratorConfig, + OrchestratorState, ParallelExecutionManager, ) from localci.core.queue import PriorityJobQueue @@ -286,3 +288,40 @@ def test_get_status(self, MockExecutor, MockDocker, MockMonitor, tmp_path): assert status["state"] in ("completed", "running") assert "total_jobs" in status assert "resources" in status + + @patch("localci.core.orchestrator.ResourceMonitor") + @patch("localci.core.orchestrator.DockerManager") + @patch("localci.core.orchestrator.JobExecutor") + def test_dispatch_pool_none_marks_job_failed( + self, MockExecutor, MockDocker, MockMonitor, tmp_path + ): + mock_monitor = MockMonitor.return_value + mock_monitor.check_thresholds.return_value = (True, []) + + queue = PriorityJobQueue() + job = make_job("GCC 15", priority=1) + queue.enqueue(job) + + orchestrator = ParallelExecutionManager( + queue=queue, + workflow_file=Path("ci.yml"), + config=OrchestratorConfig(max_parallel=4), + logs_dir=tmp_path / "logs", + ) + orchestrator._run = ExecutionRun( + execution_id="test", + started_at=datetime.now(), + state=OrchestratorState.RUNNING, + ) + orchestrator._state = OrchestratorState.RUNNING + orchestrator._pool = None + + orchestrator._dispatch_loop() + + result = orchestrator._run.results[job.queue_key] + assert result.status == JobStatus.ERROR + assert result.error_message == "Orchestrator thread pool unavailable" + failed_jobs = queue.get_jobs_by_status(QueuedJobStatus.FAILED) + assert len(failed_jobs) == 1 + assert failed_jobs[0].queue_key == job.queue_key + MockExecutor.return_value.run.assert_not_called()