diff --git a/tests/cli/test_init_guidance.py b/tests/cli/test_init_guidance.py index 74924c53..acca3c0c 100644 --- a/tests/cli/test_init_guidance.py +++ b/tests/cli/test_init_guidance.py @@ -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: @@ -21,7 +23,9 @@ class TestInitShowsNextSteps: 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", "--no-download"]) + 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 @@ -36,7 +40,9 @@ class TestInitShowsNextSteps: 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", "--no-download"]) + result = CliRunner().invoke( + cli, ["init", "--engine", "llamacpp", _NO_DL] + ) assert result.exit_code == 0 assert "[engine]" in result.output assert "[intelligence]" in result.output @@ -96,7 +102,9 @@ class TestMinimalConfig: 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-download"]) + result = CliRunner().invoke( + cli, ["init", "--engine", "ollama", _NO_DL] + ) assert result.exit_code == 0 content = config_path.read_text() # Minimal config should be short @@ -114,7 +122,10 @@ class TestMinimalConfig: 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", "--no-download"]) + 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 @@ -132,9 +143,12 @@ class TestInitDownloadPrompt: 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") + result = CliRunner().invoke( + cli, ["init", "--engine", "ollama"], input="n\n" + ) assert result.exit_code == 0 - assert "Download" in result.output and "now?" in result.output + 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" @@ -144,9 +158,11 @@ class TestInitDownloadPrompt: 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-download"]) + result = CliRunner().invoke( + cli, ["init", "--engine", "ollama", _NO_DL] + ) assert result.exit_code == 0 - assert "Download" not in result.output or "now?" not in result.output + assert "Download" not in result.output class TestInitEmptyModelFallback: @@ -159,9 +175,14 @@ class TestInitEmptyModelFallback: mock.patch("openjarvis.cli.init_cmd.recommend_model", return_value=""), mock.patch("openjarvis.cli.init_cmd.PrivacyScanner"), ): - result = CliRunner().invoke(cli, ["init", "--engine", "llamacpp"]) + 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 + assert ( + "Not enough memory" in result.output + or "not enough memory" in result.output + ) class TestNextStepsExoNexa: @@ -179,16 +200,23 @@ class TestNextStepsExoNexa: class TestInitDownloadDispatch: - def test_init_ollama_download_calls_ollama_pull(self, tmp_path: Path) -> None: + 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.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") + result = CliRunner().invoke( + cli, ["init", "--engine", "ollama"], input="y\n" + ) assert result.exit_code == 0 mock_pull.assert_called_once() @@ -200,7 +228,9 @@ class TestInitDownloadDispatch: 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") + result = CliRunner().invoke( + cli, ["init", "--engine", "vllm"], input="y\n" + ) assert result.exit_code == 0 assert "automatically" in result.output @@ -220,7 +250,7 @@ class TestInitPrivacyHook: ScanResult("FileVault", "ok", "FileVault enabled", "darwin"), ] result = CliRunner().invoke( - cli, ["init", "--engine", "llamacpp", "--no-download"] + cli, ["init", "--engine", "llamacpp", _NO_DL] ) assert result.exit_code == 0 assert "jarvis scan" in result.output diff --git a/tests/cli/test_model_pull.py b/tests/cli/test_model_pull.py index a87a55ae..3a724950 100644 --- a/tests/cli/test_model_pull.py +++ b/tests/cli/test_model_pull.py @@ -4,7 +4,6 @@ from __future__ import annotations from unittest import mock -import pytest from click.testing import CliRunner from rich.console import Console @@ -34,6 +33,7 @@ class TestOllamaPull: def test_ollama_pull_connect_error(self) -> None: import io + import httpx console = Console(file=io.StringIO()) diff --git a/tests/cli/test_scan.py b/tests/cli/test_scan.py index c89760f3..536439d8 100644 --- a/tests/cli/test_scan.py +++ b/tests/cli/test_scan.py @@ -3,21 +3,25 @@ from __future__ import annotations import json +import sys from subprocess import CompletedProcess from unittest.mock import MagicMock, patch -import pytest - -from openjarvis.cli.scan_cmd import ScanResult, PrivacyScanner - +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) +def _make_proc( + stdout: str = "", + stderr: str = "", + returncode: int = 0, +) -> CompletedProcess: + return CompletedProcess( + args=[], returncode=returncode, stdout=stdout, stderr=stderr + ) # --------------------------------------------------------------------------- @@ -27,7 +31,9 @@ def _make_proc(stdout: str = "", stderr: str = "", returncode: int = 0) -> Compl class TestScanResultDataclass: def test_fields_exist(self) -> None: - r = ScanResult(name="test", status="ok", message="all good", platform="all") + r = ScanResult( + name="test", status="ok", message="all good", platform="all" + ) assert r.name == "test" assert r.status == "ok" assert r.message == "all good" @@ -52,13 +58,15 @@ class TestScanResultDataclass: class TestFileVault: def test_filevault_enabled(self) -> None: scanner = PrivacyScanner() - with patch("subprocess.run", return_value=_make_proc(stdout="FileVault is On.")): + 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() - with patch("subprocess.run", return_value=_make_proc(stdout="FileVault is Off.")): + proc = _make_proc(stdout="FileVault is Off.") + with patch("subprocess.run", return_value=proc): result = scanner.check_filevault() assert result.status == "fail" @@ -77,18 +85,20 @@ class TestFileVault: 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="Enrolled via DEP: No\nMDM enrollment: not enrolled"), + 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="MDM enrollment: Yes\nEnrolled via DEP: Yes"), + return_value=_make_proc(stdout=stdout), ): result = scanner.check_mdm() assert result.status == "warn" @@ -101,9 +111,12 @@ class TestMDM: class TestCloudSync: def test_no_agents(self) -> None: - """pgrep returns non-zero (process not found) → no sync agents running.""" + """pgrep returns non-zero → no sync agents running.""" scanner = PrivacyScanner() - with patch("subprocess.run", return_value=_make_proc(returncode=1, stdout="")): + with patch( + "subprocess.run", + return_value=_make_proc(returncode=1, stdout=""), + ): result = scanner.check_cloud_sync_agents() assert result.status == "ok" @@ -128,19 +141,18 @@ class TestCloudSync: class TestNetworkExposure: def test_no_exposed_ports(self) -> None: - """Only localhost bindings → ok.""" + """Only localhost bindings -> ok.""" scanner = PrivacyScanner() - # lsof / ss output showing only 127.0.0.1 - lsof_output = "ollama 1234 user IPv4 TCP 127.0.0.1:11434 (LISTEN)\n" - with patch("subprocess.run", return_value=_make_proc(stdout=lsof_output)): + 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 0.0.0.0 or * → warn with port in message.""" + """A port bound to * -> warn with port in message.""" scanner = PrivacyScanner() - lsof_output = "ollama 1234 user IPv4 TCP *:11434 (LISTEN)\n" - with patch("subprocess.run", return_value=_make_proc(stdout=lsof_output)): + 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 @@ -153,7 +165,7 @@ class TestNetworkExposure: class TestLUKS: def test_encrypted(self) -> None: - """lsblk JSON contains crypto_LUKS → ok.""" + """lsblk JSON contains crypto_LUKS -> ok.""" scanner = PrivacyScanner() lsblk_data = { "blockdevices": [ @@ -175,7 +187,7 @@ class TestLUKS: assert result.status == "ok" def test_not_encrypted(self) -> None: - """lsblk JSON has only ext4 → fail.""" + """lsblk JSON has only ext4 -> fail.""" scanner = PrivacyScanner() lsblk_data = { "blockdevices": [ @@ -198,14 +210,17 @@ class TestLUKS: class TestScreenRecording: def test_none_running(self) -> None: - """No screen-recording processes → ok.""" + """No screen-recording processes -> ok.""" scanner = PrivacyScanner() - with patch("subprocess.run", return_value=_make_proc(returncode=1, stdout="")): + 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.""" + """TeamViewer found -> warn.""" scanner = PrivacyScanner() def _pgrep_side_effect(cmd, **kwargs): @@ -225,44 +240,44 @@ class TestScreenRecording: class TestPlatformFiltering: def test_run_all_filters_to_current_platform(self) -> None: - """run_all() should only call checks tagged 'all' or current platform.""" + """run_all() returns only current-platform + 'all' checks.""" scanner = PrivacyScanner() - called_platforms: list[str] = [] - - # Patch every check method to record its platform tag instead of running - checks = scanner._get_all_checks() - for check_fn in checks: - # Run the real method but capture which ones get included - pass - - import sys - current_plat = "darwin" if sys.platform == "darwin" else "linux" other_plat = "linux" if current_plat == "darwin" else "darwin" - # Make every check return immediately with a known platform with patch.object(scanner, "_get_all_checks") as mock_get: - darwin_check = MagicMock(return_value=ScanResult("d", "ok", "msg", "darwin")) - linux_check = MagicMock(return_value=ScanResult("l", "ok", "msg", "linux")) - all_check = MagicMock(return_value=ScanResult("a", "ok", "msg", "all")) + 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_check, linux_check, all_check] + mock_get.return_value = [darwin_ck, linux_ck, all_ck] results = scanner.run_all() - # The check for the other platform should not appear in results - other_result_platform = "linux" if current_plat == "darwin" else "darwin" result_platforms = {r.platform for r in results} - assert other_result_platform not in result_platforms + 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="") + mock_run.return_value = CompletedProcess( + [], 1, stdout="", stderr="" + ) result = scanner.check_remote_access() assert result.status == "ok" @@ -271,13 +286,22 @@ class TestRemoteAccess: 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="") + 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).""" @@ -292,32 +316,35 @@ class TestICloudSync: def test_icloud_defaults_error_falls_through_to_ok(self) -> None: scanner = PrivacyScanner() - # When _run raises, the nested try/except catches it and falls to ok 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() - 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: - import sys - plat = sys.platform + 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() - # Should only contain current platform results for r in results: assert r.platform in (plat, "all") - # Should not contain network or screen recording names = {r.name for r in results} assert "Network Exposure" not in names assert "Screen Recording" not in names