admin/OpenJarvis
1
0
Fork 0
mirror of https://github.com/open-jarvis/OpenJarvis.git synced 2026-07-31 03:12:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #115 from open-jarvis/feat/init-onboarding-and-privacy-scanner

Browse Source 
feat: interactive model download in init + privacy environment scanner
This commit is contained in:
Jon Saad-Falcon
2026-03-25 08:12:45 -07:00
committed by GitHub
parent a5949052b6 68a8f5646c
commit 14c81bba98
11 changed files with 1228 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openjarvis/cli/__init__.py
+2
View File
@@ -26,6 +26,7 @@ from openjarvis.cli.operators_cmd import operators
from openjarvis.cli.optimize_cmd import optimize_group
from openjarvis.cli.quickstart_cmd import quickstart
from openjarvis.cli.registry_cmd import registry
from openjarvis.cli.scan_cmd import scan
from openjarvis.cli.scheduler_cmd import scheduler
from openjarvis.cli.serve import serve
from openjarvis.cli.skill_cmd import skill
@@ -87,6 +88,7 @@ cli.add_command(gateway, "gateway")
cli.add_command(tool, "tool")
cli.add_command(registry, "registry")
cli.add_command(config, "config")
cli.add_command(scan, "scan")
def main() -> None:
src/openjarvis/cli/init_cmd.py
+91 -2
View File
@@ -10,10 +10,13 @@ from rich.console import Console
from rich.markup import escape
from rich.panel import Panel
from openjarvis.cli.model import find_model_spec, hf_download, ollama_pull
from openjarvis.cli.scan_cmd import PrivacyScanner
from openjarvis.core.config import (
DEFAULT_CONFIG_DIR,
DEFAULT_CONFIG_PATH,
detect_hardware,
estimated_download_gb,
generate_default_toml,
generate_minimal_toml,
recommend_engine,
@@ -132,10 +135,75 @@ def _next_steps_text(engine: str, model: str = "") -> str:
"\n"
" Run `jarvis doctor` to verify your setup."
),
"exo": (
"Next steps:\n\n"
" 1. Install and start Exo:\n"
" pip install exo\n"
" exo\n\n"
" 2. Try it out:\n"
" jarvis ask \"Hello\"\n\n"
" Run `jarvis doctor` to verify your setup."
),
"nexa": (
"Next steps:\n\n"
" 1. Install and start Nexa:\n"
" pip install nexaai\n"
" nexa server\n\n"
" 2. Try it out:\n"
" jarvis ask \"Hello\"\n\n"
" Run `jarvis doctor` to verify your setup."
),
}
return steps.get(engine, steps["ollama"])
def _quick_privacy_check(console: Console) -> None:
"""Run critical privacy checks and print compact summary."""
scanner = PrivacyScanner()
results = scanner.run_quick()
if results:
console.print(" [bold]Privacy check:[/bold]")
for r in results:
if r.status == "ok":
console.print(f" [green]\u2713[/green] {r.message}")
elif r.status == "warn":
console.print(f" [yellow]![/yellow] {r.message}")
elif r.status == "fail":
console.print(f" [red]\u2717[/red] {r.message}")
console.print()
console.print(" Run [cyan]jarvis scan[/cyan] for a full environment audit.")
def _do_download(engine: str, model: str, spec, console: Console) -> None:
"""Dispatch model download based on engine type."""
import os
if engine == "ollama":
host = os.environ.get("OLLAMA_HOST", "http://localhost:11434").rstrip("/")
ollama_pull(host, model, console)
elif engine == "llamacpp":
repo = spec.metadata.get("hf_repo", "")
gguf = spec.metadata.get("gguf_file", "")
if repo and gguf:
console.print(f" Downloading [cyan]{gguf}[/cyan] from {repo}...")
hf_download(repo, gguf, console)
else:
console.print(f" [yellow]No GGUF download info for {model}[/yellow]")
elif engine == "mlx":
mlx_repo = spec.metadata.get("mlx_repo", "")
if mlx_repo:
console.print(f" Downloading [cyan]{mlx_repo}[/cyan]...")
hf_download(mlx_repo, None, console)
else:
console.print(f" [yellow]No MLX repo info for {model}[/yellow]")
elif engine in ("vllm", "sglang"):
console.print(
f" [cyan]{model}[/cyan] will download automatically when "
f"{engine} starts serving it."
)
else:
console.print(f" Download {model} through the {engine} interface.")
@click.command()
@click.option(
"--force", is_flag=True, help="Overwrite existing config without prompting."
@@ -157,11 +225,15 @@ def _next_steps_text(engine: str, model: str = "") -> str:
default=None,
help="Inference engine to use (skips interactive selection).",
)
@click.option(
"--no-download", is_flag=True, default=False, help="Skip the model download prompt."
)
def init(
force: bool,
config: Optional[Path],
full_config: bool = False,
engine: Optional[str] = None,
no_download: bool = False,
) -> None:
"""Detect hardware and generate ~/.openjarvis/config.toml."""
console = Console()
@@ -286,8 +358,25 @@ def init(
selected_engine = engine or recommend_engine(hw)
model = recommend_model(hw, selected_engine)
if model:
console.print(f"\n [bold]Recommended model:[/bold] {model}")
if not model:
console.print(
"\n [yellow]! Not enough memory to run any local model.[/yellow]\n"
" Consider a cloud engine or a machine with more RAM."
)
else:
spec = find_model_spec(model)
size_gb = estimated_download_gb(spec.parameter_count_b) if spec else 0
console.print(
f"\n [bold]Recommended model:[/bold] {model} (~{size_gb:.1f} GB estimated)"
)
if not no_download and spec:
prompt = f" Download {model} (~{size_gb:.1f} GB estimated) now?"
if click.confirm(prompt, default=True):
_do_download(selected_engine, model, spec, console)
_quick_privacy_check(console)
console.print()
console.print(
Panel(
src/openjarvis/cli/model.py
+89 -15
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import os
import subprocess
import sys
import click
@@ -15,6 +16,7 @@ from openjarvis.core.config import load_config
from openjarvis.core.registry import ModelRegistry
from openjarvis.engine import discover_engines, discover_models
from openjarvis.intelligence import merge_discovered_models, register_builtin_models
from openjarvis.intelligence.model_catalog import BUILTIN_MODELS
@click.group()
@@ -149,18 +151,8 @@ def info(model_name: str) -> None:
console.print(Panel("\n".join(lines), title=spec.name, border_style="blue"))
@model.command()
@click.argument("model_name")
def pull(model_name: str) -> None:
"""Download a model (Ollama only)."""
console = Console()
config = load_config()
host = (
config.engine.ollama_host
or os.environ.get("OLLAMA_HOST")
or "http://localhost:11434"
).rstrip("/")
def ollama_pull(host: str, model_name: str, console: Console) -> bool:
"""Pull a model via Ollama API. Returns True on success."""
console.print(f"Pulling [cyan]{model_name}[/cyan] via Ollama...")
try:
with httpx.stream(
@@ -171,7 +163,6 @@ def pull(model_name: str) -> None:
) as resp:
resp.raise_for_status()
import json
for line in resp.iter_lines():
if not line.strip():
continue
@@ -188,9 +179,92 @@ def pull(model_name: str) -> None:
elif status:
console.print(f" {status}")
console.print(f"\n[green]Successfully pulled {model_name}[/green]")
return True
except httpx.ConnectError:
console.print("[red]Cannot connect to Ollama.[/red] Is it running?")
sys.exit(1)
return False
except httpx.HTTPStatusError as exc:
console.print(f"[red]Ollama error:[/red] {exc.response.status_code}")
sys.exit(1)
return False
def find_model_spec(model_name: str):
"""Look up a model in the builtin catalog. Returns None if not found."""
for spec in BUILTIN_MODELS:
if spec.model_id == model_name:
return spec
return None
def hf_download(repo: str, filename: str | None, console: Console) -> bool:
"""Download from HuggingFace via huggingface-cli. Returns True on success."""
cmd = ["huggingface-cli", "download", repo]
if filename:
cmd.append(filename)
try:
subprocess.run(cmd, check=True)
console.print("[green]Download complete.[/green]")
return True
except FileNotFoundError:
console.print(
"[red]huggingface-cli not found.[/red]\n"
"Install it: [cyan]pip install huggingface_hub[/cyan]\n"
f"Or download manually: https://huggingface.co/{repo}"
)
return False
except subprocess.CalledProcessError:
console.print("[red]Download failed.[/red]")
return False
@model.command()
@click.argument("model_name")
@click.option(
"--engine", default=None, help="Engine to download for."
)
def pull(model_name: str, engine: str | None) -> None:
"""Download a model."""
console = Console()
config = load_config()
engine = engine or config.engine.default or "ollama"
if engine == "ollama":
host = (
config.engine.ollama_host
or os.environ.get("OLLAMA_HOST")
or "http://localhost:11434"
).rstrip("/")
if not ollama_pull(host, model_name, console):
sys.exit(1)
elif engine in ("llamacpp", "mlx"):
spec = find_model_spec(model_name)
if not spec:
console.print(f"[red]Model not in catalog:[/red] {model_name}")
sys.exit(1)
if engine == "llamacpp":
repo = spec.metadata.get("hf_repo", "")
gguf = spec.metadata.get("gguf_file", "")
if not repo or not gguf:
console.print(f"[red]No GGUF download info for {model_name}[/red]")
sys.exit(1)
console.print(f"Downloading [cyan]{gguf}[/cyan] from {repo}...")
if not hf_download(repo, gguf, console):
sys.exit(1)
else: # mlx
mlx_repo = spec.metadata.get("mlx_repo", "")
if not mlx_repo:
console.print(f"[red]No MLX repo info for {model_name}[/red]")
sys.exit(1)
console.print(f"Downloading [cyan]{mlx_repo}[/cyan]...")
if not hf_download(mlx_repo, None, console):
sys.exit(1)
elif engine in ("vllm", "sglang"):
console.print(
f"[cyan]{model_name}[/cyan] will download automatically when "
f"{engine} starts serving it."
)
else:
console.print(
f"Manual download required for engine [cyan]{engine}[/cyan].\n"
f"Check the engine documentation for instructions."
)
src/openjarvis/cli/scan_cmd.py
+386
View File
@@ -0,0 +1,386 @@
"""``jarvis scan`` — audit your environment for privacy and security risks."""
from __future__ import annotations
import json
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, List
import click
# Engine ports that should only be listening on localhost.
_ENGINE_PORTS = {11434, 8080, 8000, 30000, 1234, 52415, 18181}
# Processes associated with cloud-sync agents.
_CLOUD_SYNC_PROCS = ["Dropbox", "OneDrive", "Google Drive", "iCloudDrive"]
# Screen-recording / remote-access processes (macOS).
_SCREEN_RECORDING_PROCS = [
"TeamViewer", "AnyDesk", "ScreenConnect", "vncviewer", "Vine"
]
# Remote-access processes (Linux).
_REMOTE_ACCESS_PROCS = ["xrdp", "x11vnc", "vncserver", "AnyDesk"]
# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------
@dataclass(slots=True)
class ScanResult:
"""Result of a single privacy/security check."""
name: str
status: str # "ok" | "warn" | "fail" | "skip"
message: str
platform: str # "darwin" | "linux" | "all"
# ---------------------------------------------------------------------------
# Scanner
# ---------------------------------------------------------------------------
class PrivacyScanner:
"""Collection of environment privacy checks."""
# -- Subprocess helper ---------------------------------------------------
def _run(self, cmd: list[str]) -> subprocess.CompletedProcess: # type: ignore[type-arg]
return subprocess.run(cmd, capture_output=True, text=True, timeout=10)
# -- Individual checks ---------------------------------------------------
def check_filevault(self) -> ScanResult:
"""Check whether FileVault disk encryption is enabled (macOS)."""
try:
proc = self._run(["fdesetup", "status"])
if "On" in proc.stdout:
return ScanResult(
name="FileVault",
status="ok",
message="FileVault is enabled.",
platform="darwin",
)
return ScanResult(
name="FileVault",
status="fail",
message="FileVault is NOT enabled. Enable full-disk encryption.",
platform="darwin",
)
except Exception:
return ScanResult(
name="FileVault",
status="skip",
message="fdesetup not available.",
platform="darwin",
)
def check_mdm(self) -> ScanResult:
"""Check whether the device is enrolled in an MDM profile (macOS)."""
try:
proc = self._run(["profiles", "status", "-type", "enrollment"])
output = proc.stdout + proc.stderr
lower = output.lower()
# "not enrolled" is a strong negative signal — check it first.
not_enrolled = "not enrolled" in lower or "no" in lower
# Positive signals: explicit Yes/enrolled without a negation.
enrolled_yes = (
"mdm enrollment: yes" in lower
or "enrolled via dep: yes" in lower
or ("enrolled" in lower and not not_enrolled)
)
if enrolled_yes:
return ScanResult(
name="MDM Enrollment",
status="warn",
message="Device appears to be enrolled in an MDM profile.",
platform="darwin",
)
return ScanResult(
name="MDM Enrollment",
status="ok",
message="Device is not enrolled in an MDM profile.",
platform="darwin",
)
except Exception:
return ScanResult(
name="MDM Enrollment",
status="skip",
message="profiles command not available.",
platform="darwin",
)
def check_icloud_sync(self) -> ScanResult:
"""Check whether ~/.openjarvis is inside iCloud Drive sync scope."""
try:
config_path = Path("~/.openjarvis").expanduser().resolve()
icloud_path = Path("~/Library/Mobile Documents/").expanduser().resolve()
if str(config_path).startswith(str(icloud_path)):
return ScanResult(
name="iCloud Sync",
status="warn",
message="~/.openjarvis may be synced to iCloud.",
platform="darwin",
)
# Also probe defaults for com.apple.bird (iCloud daemon)
try:
proc = self._run(
["defaults", "read", "com.apple.bird", "optout_preference"]
)
val = proc.stdout.strip()
if val == "0":
return ScanResult(
name="iCloud Sync",
status="warn",
message="iCloud Desktop/Documents sync may be active.",
platform="darwin",
)
except Exception:
pass
return ScanResult(
name="iCloud Sync",
status="ok",
message="~/.openjarvis is not inside iCloud Drive.",
platform="darwin",
)
except Exception:
return ScanResult(
name="iCloud Sync",
status="skip",
message="Could not determine iCloud sync status.",
platform="darwin",
)
def check_luks(self) -> ScanResult:
"""Check whether any block device uses LUKS encryption (Linux)."""
try:
proc = self._run(["lsblk", "-o", "NAME,TYPE,FSTYPE", "-J"])
data = json.loads(proc.stdout)
except Exception:
return ScanResult(
name="LUKS Encryption",
status="skip",
message="lsblk not available or returned unexpected output.",
platform="linux",
)
def _has_luks(devices: list) -> bool: # type: ignore[type-arg]
for dev in devices:
if dev.get("fstype") == "crypto_LUKS":
return True
children = dev.get("children") or []
if _has_luks(children):
return True
return False
try:
devices = data.get("blockdevices", [])
if _has_luks(devices):
return ScanResult(
name="LUKS Encryption",
status="ok",
message="At least one LUKS-encrypted device found.",
platform="linux",
)
return ScanResult(
name="LUKS Encryption",
status="fail",
message="No LUKS-encrypted block devices found.",
platform="linux",
)
except Exception:
return ScanResult(
name="LUKS Encryption",
status="skip",
message="Could not parse lsblk output.",
platform="linux",
)
def _check_processes(
self,
names: list[str],
check_name: str,
warn_msg: str,
platform: str,
) -> ScanResult:
"""Shared helper: pgrep for any of the given process names."""
try:
for name in names:
try:
proc = self._run(["pgrep", "-x", name])
if proc.returncode == 0:
return ScanResult(
name=check_name,
status="warn",
message=warn_msg.format(name=name),
platform=platform,
)
except Exception:
continue
return ScanResult(
name=check_name,
status="ok",
message=f"No {check_name.lower()} processes detected.",
platform=platform,
)
except Exception:
return ScanResult(
name=check_name,
status="skip",
message="pgrep not available.",
platform=platform,
)
def check_cloud_sync_agents(self) -> ScanResult:
"""Check for running cloud-sync agent processes."""
return self._check_processes(
names=_CLOUD_SYNC_PROCS,
check_name="Cloud Sync Agents",
warn_msg="{name} sync agent is running — weights may be uploaded to cloud.",
platform="all",
)
def check_network_exposure(self) -> ScanResult:
"""Check if engine ports are exposed on 0.0.0.0 rather than localhost."""
try:
if sys.platform == "darwin":
proc = self._run(["lsof", "-iTCP", "-sTCP:LISTEN", "-n", "-P"])
else:
proc = self._run(["ss", "-tlnp"])
output = proc.stdout
exposed: list[int] = []
for port in _ENGINE_PORTS:
# Look for patterns like *:PORT or 0.0.0.0:PORT
for token in (f"*:{port}", f"0.0.0.0:{port}", f":::{port}"):
if token.replace(" ", "") in output.replace(" ", ""):
exposed.append(port)
break
if exposed:
ports_str = ", ".join(str(p) for p in sorted(exposed))
return ScanResult(
name="Network Exposure",
status="warn",
message=f"Engine port(s) {ports_str} exposed on all interfaces.",
platform="all",
)
return ScanResult(
name="Network Exposure",
status="ok",
message="All engine ports appear to be bound to localhost only.",
platform="all",
)
except Exception:
return ScanResult(
name="Network Exposure",
status="skip",
message="Could not determine network exposure (lsof/ss unavailable).",
platform="all",
)
def check_screen_recording(self) -> ScanResult:
"""Check for running screen-recording / remote-desktop processes (macOS)."""
return self._check_processes(
names=_SCREEN_RECORDING_PROCS,
check_name="Screen Recording",
warn_msg="{name} is running — screen may be accessible remotely.",
platform="darwin",
)
def check_remote_access(self) -> ScanResult:
"""Check for running remote-access processes (Linux)."""
return self._check_processes(
names=_REMOTE_ACCESS_PROCS,
check_name="Remote Access",
warn_msg="{name} is running — system may be accessible remotely.",
platform="linux",
)
# -- Orchestration -------------------------------------------------------
def _get_all_checks(self) -> list[Callable[[], ScanResult]]:
return [
self.check_filevault,
self.check_mdm,
self.check_icloud_sync,
self.check_cloud_sync_agents,
self.check_network_exposure,
self.check_luks,
self.check_screen_recording,
self.check_remote_access,
]
def run_all(self) -> list[ScanResult]:
"""Run all checks, filter to the current platform, hide 'skip' results."""
current_plat = "darwin" if sys.platform == "darwin" else "linux"
results: list[ScanResult] = []
for check_fn in self._get_all_checks():
result = check_fn()
if result.platform not in (current_plat, "all"):
continue
if result.status == "skip":
continue
results.append(result)
return results
def run_quick(self) -> list[ScanResult]:
"""Run only critical checks: disk encryption + cloud sync agents."""
current_plat = "darwin" if sys.platform == "darwin" else "linux"
quick_checks: list[Callable[[], ScanResult]]
if current_plat == "darwin":
quick_checks = [self.check_filevault, self.check_cloud_sync_agents]
else:
quick_checks = [self.check_luks, self.check_cloud_sync_agents]
results = []
for check_fn in quick_checks:
result = check_fn()
if result.status != "skip":
results.append(result)
return results
# ---------------------------------------------------------------------------
# CLI command
# ---------------------------------------------------------------------------
_STATUS_ICONS = {"ok": "✓", "warn": "!", "fail": "✗", "skip": "-"}
@click.command()
@click.option("--quick", is_flag=True, default=False, help="Run only critical checks.")
def scan(quick: bool) -> None:
"""Audit your environment for privacy and security risks."""
scanner = PrivacyScanner()
results: List[ScanResult] = scanner.run_quick() if quick else scanner.run_all()
if not results:
click.echo("No applicable checks for this platform.")
return
warnings = 0
failures = 0
for r in results:
icon = _STATUS_ICONS.get(r.status, "?")
click.echo(f" [{icon}] {r.name}: {r.message}")
if r.status == "warn":
warnings += 1
elif r.status == "fail":
failures += 1
click.echo("")
parts = []
if warnings:
parts.append(f"{warnings} warning(s)")
if failures:
parts.append(f"{failures} issue(s)")
if parts:
click.echo("Summary: " + ", ".join(parts) + ".")
else:
click.echo("Summary: all checks passed.")
src/openjarvis/core/config.py
+5
View File
@@ -255,6 +255,11 @@ def recommend_model(hw: HardwareInfo, engine: str) -> str:
return ""
def estimated_download_gb(parameter_count_b: float) -> float:
"""Estimate download size in GB for Q4_K_M quantized model."""
return parameter_count_b * 0.5 * 1.1
# ---------------------------------------------------------------------------
# Configuration hierarchy
# ---------------------------------------------------------------------------
src/openjarvis/intelligence/model_catalog.py
+12 -4
View File
@@ -45,11 +45,13 @@ BUILTIN_MODELS: List[ModelSpec] = [
parameter_count_b=3.0,
active_parameter_count_b=0.6,
context_length=131072,
supported_engines=("ollama", "vllm", "llamacpp", "sglang"),
supported_engines=("ollama", "vllm", "llamacpp", "sglang", "mlx"),
provider="alibaba",
metadata={
"architecture": "moe",
"hf_repo": "Qwen/Qwen3.5-3B",
"gguf_file": "qwen3.5-3b-q4_k_m.gguf",
"mlx_repo": "mlx-community/Qwen3.5-3B-4bit",
},
),
ModelSpec(
@@ -58,11 +60,13 @@ BUILTIN_MODELS: List[ModelSpec] = [
parameter_count_b=8.0,
active_parameter_count_b=1.0,
context_length=131072,
supported_engines=("ollama", "vllm", "llamacpp", "sglang"),
supported_engines=("ollama", "vllm", "llamacpp", "sglang", "mlx"),
provider="alibaba",
metadata={
"architecture": "moe",
"hf_repo": "Qwen/Qwen3.5-8B",
"gguf_file": "qwen3.5-8b-q4_k_m.gguf",
"mlx_repo": "mlx-community/Qwen3.5-8B-4bit",
},
),
ModelSpec(
@@ -71,11 +75,13 @@ BUILTIN_MODELS: List[ModelSpec] = [
parameter_count_b=14.0,
active_parameter_count_b=2.0,
context_length=131072,
supported_engines=("ollama", "vllm", "llamacpp", "sglang"),
supported_engines=("ollama", "vllm", "llamacpp", "sglang", "mlx"),
provider="alibaba",
metadata={
"architecture": "moe",
"hf_repo": "Qwen/Qwen3.5-14B",
"gguf_file": "qwen3.5-14b-q4_k_m.gguf",
"mlx_repo": "mlx-community/Qwen3.5-14B-4bit",
},
),
ModelSpec(
@@ -223,11 +229,13 @@ BUILTIN_MODELS: List[ModelSpec] = [
active_parameter_count_b=0.5,
context_length=262144,
min_vram_gb=3.0,
supported_engines=("ollama", "vllm", "sglang", "llamacpp"),
supported_engines=("ollama", "vllm", "sglang", "llamacpp", "mlx"),
provider="alibaba",
metadata={
"architecture": "moe",
"hf_repo": "Qwen/Qwen3.5-4B",
"gguf_file": "qwen3.5-4b-q4_k_m.gguf",
"mlx_repo": "mlx-community/Qwen3.5-4B-4bit",
},
),
ModelSpec(
tests/cli/test_cli.py
+4 -1
View File
@@ -80,8 +80,11 @@ class TestCLI:
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(cli, ["init", "--engine", "ollama"])
result = CliRunner().invoke(
cli, ["init", "--engine", "ollama", "--no-download"]
)
assert result.exit_code == 0
assert config_path.exists()
content = config_path.read_text()
tests/cli/test_init_guidance.py
+141 -4
View File
@@ -10,6 +10,8 @@ from click.testing import CliRunner
from openjarvis.cli import cli
from openjarvis.cli.init_cmd import _next_steps_text
_NO_DL = "--no-download"
class TestInitShowsNextSteps:
def test_init_shows_next_steps(self, tmp_path: Path) -> None:
@@ -19,8 +21,11 @@ class TestInitShowsNextSteps:
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(cli, ["init", "--engine", "llamacpp"])
result = CliRunner().invoke(
cli, ["init", "--engine", "llamacpp", _NO_DL]
)
assert result.exit_code == 0
assert "Getting Started" in result.output
assert "jarvis ask" in result.output
@@ -33,8 +38,11 @@ class TestInitShowsNextSteps:
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(cli, ["init", "--engine", "llamacpp"])
result = CliRunner().invoke(
cli, ["init", "--engine", "llamacpp", _NO_DL]
)
assert result.exit_code == 0
assert "[engine]" in result.output
assert "[intelligence]" in result.output
@@ -92,8 +100,11 @@ class TestMinimalConfig:
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(cli, ["init", "--engine", "ollama"])
result = CliRunner().invoke(
cli, ["init", "--engine", "ollama", _NO_DL]
)
assert result.exit_code == 0
content = config_path.read_text()
# Minimal config should be short
@@ -109,11 +120,137 @@ class TestMinimalConfig:
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(cli, ["init", "--full", "--engine", "ollama"])
result = CliRunner().invoke(
cli,
["init", "--full", "--engine", "ollama", _NO_DL],
)
assert result.exit_code == 0
content = config_path.read_text()
# Full config should have many sections
assert "[engine.ollama]" in content
assert "[server]" in content
assert "[security]" in content
class TestInitDownloadPrompt:
def test_init_shows_download_prompt(self, tmp_path: Path) -> None:
config_dir = tmp_path / ".openjarvis"
config_path = config_dir / "config.toml"
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(
cli, ["init", "--engine", "ollama"], input="n\n"
)
assert result.exit_code == 0
assert "Download" in result.output
assert "now?" in result.output
def test_init_no_download_flag_skips_prompt(self, tmp_path: Path) -> None:
config_dir = tmp_path / ".openjarvis"
config_path = config_dir / "config.toml"
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(
cli, ["init", "--engine", "ollama", _NO_DL]
)
assert result.exit_code == 0
assert "Download" not in result.output
class TestInitEmptyModelFallback:
def test_init_no_model_shows_warning(self, tmp_path: Path) -> None:
config_dir = tmp_path / ".openjarvis"
config_path = config_dir / "config.toml"
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.recommend_model", return_value=""),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(
cli, ["init", "--engine", "llamacpp"]
)
assert result.exit_code == 0
assert (
"Not enough memory" in result.output
or "not enough memory" in result.output
)
class TestNextStepsExoNexa:
def test_next_steps_exo(self) -> None:
text = _next_steps_text("exo")
assert "exo" in text.lower()
assert "jarvis ask" in text
assert "ollama" not in text.lower()
def test_next_steps_nexa(self) -> None:
text = _next_steps_text("nexa")
assert "nexa" in text.lower()
assert "jarvis ask" in text
assert "ollama" not in text.lower()
class TestInitDownloadDispatch:
def test_init_ollama_download_calls_ollama_pull(
self, tmp_path: Path
) -> None:
config_dir = tmp_path / ".openjarvis"
config_path = config_dir / "config.toml"
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch(
"openjarvis.cli.init_cmd.ollama_pull",
return_value=True,
) as mock_pull,
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(
cli, ["init", "--engine", "ollama"], input="y\n"
)
assert result.exit_code == 0
mock_pull.assert_called_once()
def test_init_vllm_shows_auto_download_message(self, tmp_path: Path) -> None:
config_dir = tmp_path / ".openjarvis"
config_path = config_dir / "config.toml"
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"),
):
result = CliRunner().invoke(
cli, ["init", "--engine", "vllm"], input="y\n"
)
assert result.exit_code == 0
assert "automatically" in result.output
class TestInitPrivacyHook:
def test_init_shows_privacy_summary(self, tmp_path: Path) -> None:
config_dir = tmp_path / ".openjarvis"
config_path = config_dir / "config.toml"
with (
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_DIR", config_dir),
mock.patch("openjarvis.cli.init_cmd.DEFAULT_CONFIG_PATH", config_path),
mock.patch("openjarvis.cli.init_cmd.PrivacyScanner") as MockScanner,
):
from openjarvis.cli.scan_cmd import ScanResult
instance = MockScanner.return_value
instance.run_quick.return_value = [
ScanResult("FileVault", "ok", "FileVault enabled", "darwin"),
]
result = CliRunner().invoke(
cli, ["init", "--engine", "llamacpp", _NO_DL]
)
assert result.exit_code == 0
assert "jarvis scan" in result.output
tests/cli/test_model_pull.py
+108
View File
@@ -0,0 +1,108 @@
"""Tests for ``jarvis model pull`` multi-engine support."""
from __future__ import annotations
from unittest import mock
from click.testing import CliRunner
from rich.console import Console
from openjarvis.cli.model import ollama_pull
class TestOllamaPull:
"""Test the extracted ollama_pull helper."""
def test_ollama_pull_success(self) -> None:
import io
console = Console(file=io.StringIO())
mock_lines = [
'{"status": "pulling manifest"}',
'{"status": "downloading", "total": 100, "completed": 100}',
'{"status": "success"}',
]
mock_resp = mock.MagicMock()
mock_resp.raise_for_status = mock.MagicMock()
mock_resp.iter_lines.return_value = iter(mock_lines)
mock_resp.__enter__ = mock.MagicMock(return_value=mock_resp)
mock_resp.__exit__ = mock.MagicMock(return_value=False)
with mock.patch("httpx.stream", return_value=mock_resp):
result = ollama_pull("http://localhost:11434", "qwen3.5:3b", console)
assert result is True
def test_ollama_pull_connect_error(self) -> None:
import io
import httpx
console = Console(file=io.StringIO())
with mock.patch("httpx.stream", side_effect=httpx.ConnectError("refused")):
result = ollama_pull("http://localhost:11434", "qwen3.5:3b", console)
assert result is False
class TestPullCliMultiEngine:
"""Test the pull CLI command dispatches to correct engine."""
def test_pull_llamacpp_uses_huggingface_cli(self) -> None:
from openjarvis.cli import cli
runner = CliRunner()
with (
mock.patch("openjarvis.cli.model.load_config") as mock_cfg,
mock.patch("subprocess.run") as mock_run,
):
mock_cfg.return_value.engine.default = "llamacpp"
mock_cfg.return_value.engine.ollama_host = None
mock_run.return_value = mock.MagicMock(returncode=0)
result = runner.invoke(
cli, ["model", "pull", "qwen3.5:8b", "--engine", "llamacpp"]
)
assert result.exit_code == 0
mock_run.assert_called_once()
call_args = mock_run.call_args[0][0]
assert "huggingface-cli" in call_args
assert "qwen3.5-8b-q4_k_m.gguf" in call_args
def test_pull_mlx_uses_huggingface_cli(self) -> None:
from openjarvis.cli import cli
runner = CliRunner()
with (
mock.patch("openjarvis.cli.model.load_config") as mock_cfg,
mock.patch("subprocess.run") as mock_run,
):
mock_cfg.return_value.engine.default = "mlx"
mock_cfg.return_value.engine.ollama_host = None
mock_run.return_value = mock.MagicMock(returncode=0)
result = runner.invoke(
cli, ["model", "pull", "qwen3.5:8b", "--engine", "mlx"]
)
assert result.exit_code == 0
mock_run.assert_called_once()
call_args = mock_run.call_args[0][0]
assert "huggingface-cli" in call_args
assert "mlx-community/Qwen3.5-8B-4bit" in call_args
def test_pull_llamacpp_huggingface_cli_not_found(self) -> None:
from openjarvis.cli import cli
runner = CliRunner()
with (
mock.patch("openjarvis.cli.model.load_config") as mock_cfg,
mock.patch("subprocess.run", side_effect=FileNotFoundError),
):
mock_cfg.return_value.engine.default = "llamacpp"
mock_cfg.return_value.engine.ollama_host = None
result = runner.invoke(
cli, ["model", "pull", "qwen3.5:8b", "--engine", "llamacpp"]
)
assert result.exit_code != 0
assert "huggingface_hub" in result.output or "pip install" in result.output
tests/cli/test_scan.py
+350
View File
@@ -0,0 +1,350 @@
"""Tests for ``jarvis scan`` privacy scanner CLI command."""
from __future__ import annotations
import json
import sys
from subprocess import CompletedProcess
from unittest.mock import MagicMock, patch
from openjarvis.cli.scan_cmd import PrivacyScanner, ScanResult
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_proc(
stdout: str = "",
stderr: str = "",
returncode: int = 0,
) -> CompletedProcess:
return CompletedProcess(
args=[], returncode=returncode, stdout=stdout, stderr=stderr
)
# ---------------------------------------------------------------------------
# TestScanResultDataclass
# ---------------------------------------------------------------------------
class TestScanResultDataclass:
def test_fields_exist(self) -> None:
r = ScanResult(
name="test", status="ok", message="all good", platform="all"
)
assert r.name == "test"
assert r.status == "ok"
assert r.message == "all good"
assert r.platform == "all"
def test_status_values(self) -> None:
for status in ("ok", "warn", "fail", "skip"):
r = ScanResult(name="x", status=status, message="", platform="all")
assert r.status == status
def test_platform_values(self) -> None:
for plat in ("darwin", "linux", "all"):
r = ScanResult(name="x", status="ok", message="", platform=plat)
assert r.platform == plat
# ---------------------------------------------------------------------------
# TestFileVault
# ---------------------------------------------------------------------------
class TestFileVault:
def test_filevault_enabled(self) -> None:
scanner = PrivacyScanner()
proc = _make_proc(stdout="FileVault is On.")
with patch("subprocess.run", return_value=proc):
result = scanner.check_filevault()
assert result.status == "ok"
def test_filevault_disabled(self) -> None:
scanner = PrivacyScanner()
proc = _make_proc(stdout="FileVault is Off.")
with patch("subprocess.run", return_value=proc):
result = scanner.check_filevault()
assert result.status == "fail"
def test_command_not_found(self) -> None:
scanner = PrivacyScanner()
with patch("subprocess.run", side_effect=FileNotFoundError):
result = scanner.check_filevault()
assert result.status == "skip"
# ---------------------------------------------------------------------------
# TestMDM
# ---------------------------------------------------------------------------
class TestMDM:
def test_not_enrolled(self) -> None:
scanner = PrivacyScanner()
stdout = "Enrolled via DEP: No\nMDM enrollment: not enrolled"
with patch(
"subprocess.run",
return_value=_make_proc(stdout=stdout),
):
result = scanner.check_mdm()
assert result.status == "ok"
def test_enrolled(self) -> None:
scanner = PrivacyScanner()
stdout = "MDM enrollment: Yes\nEnrolled via DEP: Yes"
with patch(
"subprocess.run",
return_value=_make_proc(stdout=stdout),
):
result = scanner.check_mdm()
assert result.status == "warn"
# ---------------------------------------------------------------------------
# TestCloudSync
# ---------------------------------------------------------------------------
class TestCloudSync:
def test_no_agents(self) -> None:
"""pgrep returns non-zero → no sync agents running."""
scanner = PrivacyScanner()
with patch(
"subprocess.run",
return_value=_make_proc(returncode=1, stdout=""),
):
result = scanner.check_cloud_sync_agents()
assert result.status == "ok"
def test_dropbox_running(self) -> None:
"""pgrep returns 0 when Dropbox is in the command."""
scanner = PrivacyScanner()
def _pgrep_side_effect(cmd, **kwargs):
if "Dropbox" in cmd:
return _make_proc(returncode=0, stdout="1234")
return _make_proc(returncode=1, stdout="")
with patch("subprocess.run", side_effect=_pgrep_side_effect):
result = scanner.check_cloud_sync_agents()
assert result.status == "warn"
# ---------------------------------------------------------------------------
# TestNetworkExposure
# ---------------------------------------------------------------------------
class TestNetworkExposure:
def test_no_exposed_ports(self) -> None:
"""Only localhost bindings -> ok."""
scanner = PrivacyScanner()
lsof_out = "ollama 1234 user IPv4 TCP 127.0.0.1:11434 (LISTEN)\n"
with patch("subprocess.run", return_value=_make_proc(stdout=lsof_out)):
result = scanner.check_network_exposure()
assert result.status == "ok"
def test_exposed_port(self) -> None:
"""A port bound to * -> warn with port in message."""
scanner = PrivacyScanner()
lsof_out = "ollama 1234 user IPv4 TCP *:11434 (LISTEN)\n"
with patch("subprocess.run", return_value=_make_proc(stdout=lsof_out)):
result = scanner.check_network_exposure()
assert result.status == "warn"
assert "11434" in result.message
# ---------------------------------------------------------------------------
# TestLUKS
# ---------------------------------------------------------------------------
class TestLUKS:
def test_encrypted(self) -> None:
"""lsblk JSON contains crypto_LUKS -> ok."""
scanner = PrivacyScanner()
lsblk_data = {
"blockdevices": [
{
"name": "sda",
"type": "disk",
"fstype": None,
"children": [
{"name": "sda1", "type": "part", "fstype": "crypto_LUKS"}
],
}
]
}
with patch(
"subprocess.run",
return_value=_make_proc(stdout=json.dumps(lsblk_data)),
):
result = scanner.check_luks()
assert result.status == "ok"
def test_not_encrypted(self) -> None:
"""lsblk JSON has only ext4 -> fail."""
scanner = PrivacyScanner()
lsblk_data = {
"blockdevices": [
{"name": "sda", "type": "disk", "fstype": None},
{"name": "sda1", "type": "part", "fstype": "ext4"},
]
}
with patch(
"subprocess.run",
return_value=_make_proc(stdout=json.dumps(lsblk_data)),
):
result = scanner.check_luks()
assert result.status == "fail"
# ---------------------------------------------------------------------------
# TestScreenRecording
# ---------------------------------------------------------------------------
class TestScreenRecording:
def test_none_running(self) -> None:
"""No screen-recording processes -> ok."""
scanner = PrivacyScanner()
with patch(
"subprocess.run",
return_value=_make_proc(returncode=1, stdout=""),
):
result = scanner.check_screen_recording()
assert result.status == "ok"
def test_teamviewer_running(self) -> None:
"""TeamViewer found -> warn."""
scanner = PrivacyScanner()
def _pgrep_side_effect(cmd, **kwargs):
if "TeamViewer" in cmd:
return _make_proc(returncode=0, stdout="5678")
return _make_proc(returncode=1, stdout="")
with patch("subprocess.run", side_effect=_pgrep_side_effect):
result = scanner.check_screen_recording()
assert result.status == "warn"
# ---------------------------------------------------------------------------
# TestPlatformFiltering
# ---------------------------------------------------------------------------
class TestPlatformFiltering:
def test_run_all_filters_to_current_platform(self) -> None:
"""run_all() returns only current-platform + 'all' checks."""
scanner = PrivacyScanner()
current_plat = "darwin" if sys.platform == "darwin" else "linux"
other_plat = "linux" if current_plat == "darwin" else "darwin"
with patch.object(scanner, "_get_all_checks") as mock_get:
darwin_ck = MagicMock(
return_value=ScanResult("d", "ok", "msg", "darwin")
)
linux_ck = MagicMock(
return_value=ScanResult("l", "ok", "msg", "linux")
)
all_ck = MagicMock(
return_value=ScanResult("a", "ok", "msg", "all")
)
mock_get.return_value = [darwin_ck, linux_ck, all_ck]
results = scanner.run_all()
result_platforms = {r.platform for r in results}
assert other_plat not in result_platforms
assert "all" in result_platforms or current_plat in result_platforms
# ---------------------------------------------------------------------------
# TestRemoteAccess
# ---------------------------------------------------------------------------
class TestRemoteAccess:
"""Tests for check_remote_access (Linux)."""
def test_no_remote_access(self) -> None:
scanner = PrivacyScanner()
with patch.object(scanner, "_run") as mock_run:
mock_run.return_value = CompletedProcess(
[], 1, stdout="", stderr=""
)
result = scanner.check_remote_access()
assert result.status == "ok"
def test_xrdp_running(self) -> None:
scanner = PrivacyScanner()
with patch.object(scanner, "_run") as mock_run:
def side_effect(cmd, **kw):
if any("xrdp" in str(c) for c in cmd):
return CompletedProcess(
cmd, 0, stdout="12345", stderr=""
)
return CompletedProcess(
cmd, 1, stdout="", stderr=""
)
mock_run.side_effect = side_effect
result = scanner.check_remote_access()
assert result.status == "warn"
# ---------------------------------------------------------------------------
# TestICloudSync
# ---------------------------------------------------------------------------
class TestICloudSync:
"""Tests for check_icloud_sync (macOS)."""
def test_no_icloud_sync(self) -> None:
scanner = PrivacyScanner()
with patch.object(scanner, "_run") as mock_run:
mock_run.return_value = CompletedProcess(
[], 0, stdout="no relevant output", stderr=""
)
result = scanner.check_icloud_sync()
assert result.status in ("ok", "skip")
def test_icloud_defaults_error_falls_through_to_ok(self) -> None:
scanner = PrivacyScanner()
with patch.object(scanner, "_run", side_effect=FileNotFoundError):
result = scanner.check_icloud_sync()
assert result.status == "ok"
# ---------------------------------------------------------------------------
# TestRunQuick
# ---------------------------------------------------------------------------
class TestRunQuick:
"""Tests for run_quick subset."""
def test_run_quick_returns_subset(self) -> None:
scanner = PrivacyScanner()
plat = sys.platform
with (
patch.object(scanner, "check_filevault") as fv,
patch.object(scanner, "check_luks") as luks,
patch.object(scanner, "check_icloud_sync") as ic,
patch.object(scanner, "check_cloud_sync_agents") as cs,
):
fv.return_value = ScanResult("FV", "ok", "ok", "darwin")
luks.return_value = ScanResult("LUKS", "ok", "ok", "linux")
ic.return_value = ScanResult("iCloud", "ok", "ok", "darwin")
cs.return_value = ScanResult("Cloud", "ok", "ok", plat)
results = scanner.run_quick()
for r in results:
assert r.platform in (plat, "all")
names = {r.name for r in results}
assert "Network Exposure" not in names
assert "Screen Recording" not in names
tests/core/test_recommend_model.py
+40
View File
@@ -115,3 +115,43 @@ class TestRecommendModelEdgeCases:
# With ollama, 397b is excluded (only vllm, sglang)
result = recommend_model(hw, "ollama")
assert result == "qwen3.5:122b"
class TestRecommendModelMlx:
"""Apple Silicon (MLX) model recommendation."""
def test_apple_silicon_8gb_mlx(self) -> None:
hw = HardwareInfo(
platform="darwin",
ram_gb=8.0,
gpu=GpuInfo(vendor="apple", name="Apple M1", vram_gb=8.0, count=1),
)
result = recommend_model(hw, "mlx")
assert result == "qwen3.5:8b"
def test_apple_silicon_16gb_mlx(self) -> None:
hw = HardwareInfo(
platform="darwin",
ram_gb=16.0,
gpu=GpuInfo(vendor="apple", name="Apple M2", vram_gb=16.0, count=1),
)
result = recommend_model(hw, "mlx")
assert result == "qwen3.5:14b"
def test_apple_silicon_32gb_mlx(self) -> None:
hw = HardwareInfo(
platform="darwin",
ram_gb=32.0,
gpu=GpuInfo(vendor="apple", name="Apple M2 Pro", vram_gb=32.0, count=1),
)
result = recommend_model(hw, "mlx")
assert result == "qwen3.5:14b"
def test_apple_silicon_64gb_mlx(self) -> None:
hw = HardwareInfo(
platform="darwin",
ram_gb=64.0,
gpu=GpuInfo(vendor="apple", name="Apple M2 Max", vram_gb=64.0, count=1),
)
result = recommend_model(hw, "mlx")
assert result == "qwen3.5:14b"