From 128dea2fa64906652e751cd682f6d0f9f1f28ce6 Mon Sep 17 00:00:00 2001 From: trisdoan Date: Tue, 3 Mar 2026 15:02:02 +0700 Subject: [PATCH 1/5] feat: add edit-config command for interactive config file editing --- trobz_local/main.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/trobz_local/main.py b/trobz_local/main.py index bcd2c5b..15290af 100644 --- a/trobz_local/main.py +++ b/trobz_local/main.py @@ -1,3 +1,4 @@ +import os import subprocess from pathlib import Path from typing import Annotated @@ -32,6 +33,7 @@ get_config, get_os_info, get_uv_path, + show_config_instructions, ) app = typer.Typer() @@ -71,6 +73,26 @@ def init(ctx: typer.Context): _run_init(ctx) +@app.command() +def edit_config(): + config_path = get_code_root() / "config.toml" + + if not config_path.exists(): + show_config_instructions() + raise typer.Exit(code=1) + + editor = os.environ.get("VISUAL") or os.environ.get("EDITOR", "vi") + typer.echo(f"Opening {config_path} with {editor}...") + try: + subprocess.run([editor, str(config_path)], check=True) # noqa: S603 + except FileNotFoundError: + typer.secho(f"Editor '{editor}' not found. Set $EDITOR or $VISUAL.", fg=typer.colors.RED) + raise typer.Exit(code=1) from None + except subprocess.CalledProcessError as e: + typer.secho(f"Editor exited with error: {e.returncode}", fg=typer.colors.RED) + raise typer.Exit(code=1) from None + + def _run_init(ctx: typer.Context): code_root = get_code_root() confirm_step( From 13d5816d73250a0ed511f9a51866f920ee5da5a5 Mon Sep 17 00:00:00 2001 From: trisdoan Date: Tue, 3 Mar 2026 15:21:15 +0700 Subject: [PATCH 2/5] refactor(utils): extract repo enumeration helpers to utils.py Move ODOO_URLS, iter_org_entries(), get_repo_tasks() from main.py to utils.py. Eliminates duplication and allows reuse by other modules (e.g., doctor.py). --- tests/test_pull_repos.py | 10 ++++----- trobz_local/main.py | 48 ++-------------------------------------- trobz_local/utils.py | 46 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 53 insertions(+), 51 deletions(-) diff --git a/tests/test_pull_repos.py b/tests/test_pull_repos.py index 92325e0..c55f97a 100644 --- a/tests/test_pull_repos.py +++ b/tests/test_pull_repos.py @@ -5,8 +5,8 @@ from rich.progress import Progress, TaskID from typer.testing import CliRunner -from trobz_local.main import ODOO_URLS, _get_tasks, _pull_repo -from trobz_local.utils import get_config +from trobz_local.main import _pull_repo +from trobz_local.utils import ODOO_URLS, get_config, get_repo_tasks runner = CliRunner() @@ -149,7 +149,7 @@ def test_get_tasks_generates_correct_list(mock_config, tmp_path): repos_config = {"odoo": ["odoo"], "oca": ["server-tools"]} code_root = tmp_path / "code" - tasks = _get_tasks(odoo_versions, repos_config, code_root, None) + tasks = get_repo_tasks(odoo_versions, repos_config, code_root, None) expected_tasks = [ { @@ -193,7 +193,7 @@ def test_get_tasks_with_filter(mock_config, tmp_path): repos_config = {"odoo": ["odoo"], "oca": ["server-tools"]} code_root = tmp_path / "code" - tasks = _get_tasks(odoo_versions, repos_config, code_root, repo_filter=["odoo"]) + tasks = get_repo_tasks(odoo_versions, repos_config, code_root, repo_filter=["odoo"]) expected_tasks = [ { @@ -222,7 +222,7 @@ def test_get_tasks_inline_branch_override(mock_config, tmp_path): } code_root = tmp_path / "code" - tasks = _get_tasks(odoo_versions, repos_config, code_root, None) + tasks = get_repo_tasks(odoo_versions, repos_config, code_root, None) paths = {(t["repo_name"], t["version"]): t["repo_path"] for t in tasks} diff --git a/trobz_local/main.py b/trobz_local/main.py index 15290af..b0f068c 100644 --- a/trobz_local/main.py +++ b/trobz_local/main.py @@ -32,17 +32,13 @@ get_code_root, get_config, get_os_info, + get_repo_tasks, get_uv_path, show_config_instructions, ) app = typer.Typer() -ODOO_URLS = { - "odoo": "git@github.com:odoo/odoo.git", - "enterprise": "git@github.com:odoo/enterprise.git", -} - @app.callback(invoke_without_command=True) def main( @@ -165,7 +161,7 @@ def pull_repos( # noqa: C901 repos_config = config.get("repos", {}) code_root = get_code_root() - repo_infos_for_tasks = _get_tasks(odoo_versions, repos_config, code_root, repo_filter) + repo_infos_for_tasks = get_repo_tasks(odoo_versions, repos_config, code_root, repo_filter) if not repo_infos_for_tasks: return @@ -218,46 +214,6 @@ def pull_repos( # noqa: C901 typer.secho("\nAll repositories updated successfully.", fg=typer.colors.GREEN) -def _iter_org_entries(org_repos, odoo_versions): - """Yield (repo_name, branch) pairs for an org's repo list. - - Plain strings use all configured versions; [name, [branch, ...]] entries - use their explicit branch list. - """ - for entry in org_repos: - if isinstance(entry, str): - for version in odoo_versions: - yield entry, str(version) - else: - for branch in entry[1]: - yield entry[0], str(branch) - - -def _get_tasks(odoo_versions, repos_config, code_root, repo_filter): - tasks = [] - for version in odoo_versions: - for repo_name in repos_config.get("odoo", []): - if repo_name in ODOO_URLS and (not repo_filter or repo_name in repo_filter): - tasks.append({ - "repo_name": repo_name, - "repo_path": code_root / "odoo" / repo_name / version, - "repo_url": ODOO_URLS[repo_name], - "version": str(version), - }) - for org, org_repos in repos_config.items(): - if org == "odoo": - continue - for repo_name, branch in _iter_org_entries(org_repos, odoo_versions): - if not repo_filter or repo_name in repo_filter: - tasks.append({ - "repo_name": repo_name, - "repo_path": code_root / org / branch / repo_name, - "repo_url": f"git@github.com:{org}/{repo_name}.git", - "version": branch, - }) - return tasks - - def _pull_repo(progress: Progress, task_id: TaskID, repo_info: dict): repo_name = repo_info["repo_name"] repo_path = repo_info["repo_path"] diff --git a/trobz_local/utils.py b/trobz_local/utils.py index 4ef6bec..642d434 100644 --- a/trobz_local/utils.py +++ b/trobz_local/utils.py @@ -216,6 +216,52 @@ def validate_versions(cls, v: list[str]): return v +ODOO_URLS = { + "odoo": "git@github.com:odoo/odoo.git", + "enterprise": "git@github.com:odoo/enterprise.git", +} + + +def iter_org_entries(org_repos, odoo_versions): + """Yield (repo_name, branch) pairs for an org's repo list. + + Plain strings use all configured versions; [name, [branch, ...]] entries + use their explicit branch list. + """ + for entry in org_repos: + if isinstance(entry, str): + for version in odoo_versions: + yield entry, str(version) + else: + for branch in entry[1]: + yield entry[0], str(branch) + + +def get_repo_tasks(odoo_versions, repos_config, code_root, repo_filter): + tasks = [] + for version in odoo_versions: + for repo_name in repos_config.get("odoo", []): + if repo_name in ODOO_URLS and (not repo_filter or repo_name in repo_filter): + tasks.append({ + "repo_name": repo_name, + "repo_path": code_root / "odoo" / repo_name / version, + "repo_url": ODOO_URLS[repo_name], + "version": str(version), + }) + for org, org_repos in repos_config.items(): + if org == "odoo": + continue + for repo_name, branch in iter_org_entries(org_repos, odoo_versions): + if not repo_filter or repo_name in repo_filter: + tasks.append({ + "repo_name": repo_name, + "repo_path": code_root / org / branch / repo_name, + "repo_url": f"git@github.com:{org}/{repo_name}.git", + "version": branch, + }) + return tasks + + def get_code_root() -> Path: """Get the code root directory from TLC_CODE_DIR env var or default to ~/code.""" env_code_dir = os.environ.get("TLC_CODE_DIR") From 444655eb4b27c8c41423027d752103672938f4fa Mon Sep 17 00:00:00 2001 From: trisdoan Date: Tue, 3 Mar 2026 15:21:19 +0700 Subject: [PATCH 3/5] feat(doctor): add repositories health check group Add check_repos() function to diagnose repo state (exists, clone-able, git status). Wire into run_doctor() as "Repositories" check group. Fix check_config message (remove version count). Add 7 new test cases covering repo validation scenarios. --- tests/test_doctor.py | 135 +++++++++++++++++++++++++++++++++++++++++- trobz_local/doctor.py | 61 +++++++++++++++++-- 2 files changed, 190 insertions(+), 6 deletions(-) diff --git a/tests/test_doctor.py b/tests/test_doctor.py index b1d182c..9d94aad 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -8,6 +8,7 @@ CheckStatus, check_config, check_github_ssh, + check_repos, check_system_tools, check_tool_versions, list_venvs, @@ -27,7 +28,7 @@ def test_check_config_valid(tmp_path): (tmp_path / "config.toml").write_text('versions = ["18.0"]\n\n[repos]\nodoo = ["odoo"]\n') result = check_config(tmp_path) assert result.status == CheckStatus.OK - assert "1 version" in result.message + assert "Valid" in result.message def test_check_config_missing(tmp_path): @@ -247,16 +248,18 @@ def test_list_venvs_empty_versions(tmp_path): # --------------------------------------------------------------------------- +@patch("trobz_local.doctor.check_repos") @patch("trobz_local.doctor.check_github_ssh") @patch("trobz_local.doctor.check_system_tools") @patch("trobz_local.doctor.check_tool_versions") @patch("trobz_local.doctor.list_venvs") -def test_run_doctor_with_config(mock_venvs, mock_tools, mock_sys_tools, mock_ssh, tmp_path): +def test_run_doctor_with_config(mock_venvs, mock_tools, mock_sys_tools, mock_ssh, mock_repos, tmp_path): (tmp_path / "config.toml").write_text('versions = ["18.0"]\n') mock_ssh.return_value = CheckResult("GitHub SSH", CheckStatus.OK, "Authenticated") mock_sys_tools.return_value = [CheckResult("git", CheckStatus.OK, "Found (git version 2.45.0)")] mock_tools.return_value = [] mock_venvs.return_value = [] + mock_repos.return_value = [CheckResult("odoo (18.0)", CheckStatus.OK, "up to date")] groups = run_doctor(tmp_path) @@ -264,6 +267,7 @@ def test_run_doctor_with_config(mock_venvs, mock_tools, mock_sys_tools, mock_ssh assert "Connectivity" in groups assert "Tools" in groups assert "Virtual Environments" in groups + assert "Repositories" in groups assert groups["Configuration"][0].status == CheckStatus.OK # System tools always included assert groups["Tools"][0].name == "git" @@ -324,3 +328,130 @@ def test_doctor_command_warnings_exit_zero(mock_root, mock_doctor, tmp_path): result = runner.invoke(app, ["doctor"]) assert result.exit_code == 0 assert "!!" in result.output + + +# --------------------------------------------------------------------------- +# check_repos +# --------------------------------------------------------------------------- + + +@patch("trobz_local.doctor.get_repo_tasks") +def test_check_repos_not_cloned(mock_tasks, tmp_path): + mock_tasks.return_value = [ + {"repo_name": "odoo", "repo_path": tmp_path / "nonexistent", "version": "18.0"}, + ] + results = check_repos(tmp_path, {}, ["18.0"]) + assert len(results) == 1 + assert results[0].status == CheckStatus.FAIL + assert "not cloned" in results[0].message + + +@patch("trobz_local.doctor.git.Repo") +@patch("trobz_local.doctor.get_repo_tasks") +def test_check_repos_up_to_date(mock_tasks, mock_repo_cls, tmp_path): + repo_path = tmp_path / "odoo" + repo_path.mkdir() + mock_tasks.return_value = [ + {"repo_name": "odoo", "repo_path": repo_path, "version": "18.0"}, + ] + mock_repo = MagicMock() + mock_repo.head.commit.hexsha = "abc123" + mock_repo.remotes.origin.refs.__getitem__.return_value.commit.hexsha = "abc123" + mock_repo.is_dirty.return_value = False + mock_repo_cls.return_value = mock_repo + + results = check_repos(tmp_path, {}, ["18.0"]) + assert results[0].status == CheckStatus.OK + assert "up to date" in results[0].message + + +@patch("trobz_local.doctor.git.Repo") +@patch("trobz_local.doctor.get_repo_tasks") +def test_check_repos_behind(mock_tasks, mock_repo_cls, tmp_path): + repo_path = tmp_path / "odoo" + repo_path.mkdir() + mock_tasks.return_value = [ + {"repo_name": "odoo", "repo_path": repo_path, "version": "18.0"}, + ] + mock_repo = MagicMock() + mock_repo.head.commit.hexsha = "abc123" + mock_repo.remotes.origin.refs.__getitem__.return_value.commit.hexsha = "def456" + mock_repo.is_dirty.return_value = False + mock_repo_cls.return_value = mock_repo + + results = check_repos(tmp_path, {}, ["18.0"]) + assert results[0].status == CheckStatus.WARN + assert "behind upstream" in results[0].message + + +@patch("trobz_local.doctor.git.Repo") +@patch("trobz_local.doctor.get_repo_tasks") +def test_check_repos_dirty(mock_tasks, mock_repo_cls, tmp_path): + repo_path = tmp_path / "odoo" + repo_path.mkdir() + mock_tasks.return_value = [ + {"repo_name": "odoo", "repo_path": repo_path, "version": "18.0"}, + ] + mock_repo = MagicMock() + mock_repo.head.commit.hexsha = "abc123" + mock_repo.remotes.origin.refs.__getitem__.return_value.commit.hexsha = "abc123" + mock_repo.is_dirty.return_value = True + mock_repo_cls.return_value = mock_repo + + results = check_repos(tmp_path, {}, ["18.0"]) + assert results[0].status == CheckStatus.WARN + assert "dirty" in results[0].message + + +@patch("trobz_local.doctor.git.Repo") +@patch("trobz_local.doctor.get_repo_tasks") +def test_check_repos_behind_and_dirty(mock_tasks, mock_repo_cls, tmp_path): + repo_path = tmp_path / "odoo" + repo_path.mkdir() + mock_tasks.return_value = [ + {"repo_name": "odoo", "repo_path": repo_path, "version": "18.0"}, + ] + mock_repo = MagicMock() + mock_repo.head.commit.hexsha = "abc123" + mock_repo.remotes.origin.refs.__getitem__.return_value.commit.hexsha = "def456" + mock_repo.is_dirty.return_value = True + mock_repo_cls.return_value = mock_repo + + results = check_repos(tmp_path, {}, ["18.0"]) + assert results[0].status == CheckStatus.WARN + assert "behind upstream" in results[0].message + assert "dirty" in results[0].message + + +@patch("trobz_local.doctor.git.Repo") +@patch("trobz_local.doctor.get_repo_tasks") +def test_check_repos_fetch_error(mock_tasks, mock_repo_cls, tmp_path): + repo_path = tmp_path / "odoo" + repo_path.mkdir() + mock_tasks.return_value = [ + {"repo_name": "odoo", "repo_path": repo_path, "version": "18.0"}, + ] + mock_repo = MagicMock() + mock_repo.remotes.origin.fetch.side_effect = Exception("network error") + mock_repo_cls.return_value = mock_repo + + results = check_repos(tmp_path, {}, ["18.0"]) + assert results[0].status == CheckStatus.WARN + assert "fetch failed" in results[0].message + + +@patch("trobz_local.doctor.git.Repo") +@patch("trobz_local.doctor.get_repo_tasks") +def test_check_repos_invalid_repo(mock_tasks, mock_repo_cls, tmp_path): + import git as git_mod + + repo_path = tmp_path / "odoo" + repo_path.mkdir() + mock_tasks.return_value = [ + {"repo_name": "odoo", "repo_path": repo_path, "version": "18.0"}, + ] + mock_repo_cls.side_effect = git_mod.exc.InvalidGitRepositoryError("bad") + + results = check_repos(tmp_path, {}, ["18.0"]) + assert results[0].status == CheckStatus.FAIL + assert "invalid git repository" in results[0].message diff --git a/trobz_local/doctor.py b/trobz_local/doctor.py index b7e76e7..9ffc200 100644 --- a/trobz_local/doctor.py +++ b/trobz_local/doctor.py @@ -5,10 +5,11 @@ from enum import Enum from pathlib import Path +import git import tomli from pydantic import ValidationError -from .utils import ConfigModel +from .utils import ConfigModel, get_repo_tasks class CheckStatus(Enum): @@ -48,7 +49,7 @@ def check_config(code_root: Path) -> CheckResult: ) try: - validated = ConfigModel(**raw) + ConfigModel(**raw) except ValidationError as e: errors = "; ".join(err["msg"] for err in e.errors()) return CheckResult( @@ -58,11 +59,10 @@ def check_config(code_root: Path) -> CheckResult: detail=errors, ) - version_count = len(validated.versions) return CheckResult( name="Config file", status=CheckStatus.OK, - message=f"Valid — {version_count} version(s) defined", + message=f"Valid — Config file at {config_path}", ) @@ -335,6 +335,52 @@ def list_venvs(code_root: Path, versions: list[str]) -> list[CheckResult]: return results +def check_repos(code_root: Path, repos_config: dict, versions: list[str]) -> list[CheckResult]: + """Check each configured repo's upstream status and dirty state.""" + tasks = get_repo_tasks(versions, repos_config, code_root, repo_filter=None) + results = [] + for task in tasks: + name = f"{task['repo_name']} ({task['version']})" + repo_path = task["repo_path"] + version = task["version"] + + if not repo_path.exists(): + results.append(CheckResult(name=name, status=CheckStatus.FAIL, message="not cloned")) + continue + + try: + repo = git.Repo(repo_path) + except git.exc.InvalidGitRepositoryError: + results.append(CheckResult(name=name, status=CheckStatus.FAIL, message="invalid git repository")) + continue + + # Fetch and compare SHAs + try: + repo.remotes.origin.fetch(version) + local_sha = repo.head.commit.hexsha + remote_sha = repo.remotes.origin.refs[version].commit.hexsha + behind = local_sha != remote_sha + except Exception as e: + results.append(CheckResult(name=name, status=CheckStatus.WARN, message=f"fetch failed: {e}")) + continue + + dirty = repo.is_dirty(untracked_files=True) + + if behind and dirty: + msg = "behind upstream, dirty" + elif behind: + msg = "behind upstream" + elif dirty: + msg = "dirty (uncommitted changes)" + else: + msg = "up to date" + + status = CheckStatus.WARN if (behind or dirty) else CheckStatus.OK + results.append(CheckResult(name=name, status=status, message=msg)) + + return results + + def run_doctor(code_root: Path) -> dict[str, list[CheckResult]]: """Run all health checks and return grouped results.""" groups: dict[str, list[CheckResult]] = {} @@ -374,4 +420,11 @@ def run_doctor(code_root: Path) -> dict[str, list[CheckResult]]: versions = config.versions if config else [] groups["Virtual Environments"] = list_venvs(code_root, versions) + # --- Repositories --- + if config: + repos_config = config.repos.model_dump() + groups["Repositories"] = check_repos(code_root, repos_config, versions) + else: + groups["Repositories"] = [] + return groups From 4d4f4f93ee3a0bbb466884bb1d8ad1535301e4fb Mon Sep 17 00:00:00 2001 From: trisdoan Date: Wed, 4 Mar 2026 10:51:24 +0700 Subject: [PATCH 4/5] feat(cli): add version, debug, and quiet options with improved error handling - Add --version/-V flag with version callback - Add --debug flag to show full stack traces on error - Add --quiet/-q flag to suppress informational output - Implement cli() wrapper with top-level exception handling - Update entry point from 'app' to 'cli' in pyproject.toml - Add epilog to CLI help with docs link - Use Annotated types for improved CLI option definitions --- pyproject.toml | 2 +- trobz_local/main.py | 118 ++++++++++++++++++++++++++++++++++---------- 2 files changed, 94 insertions(+), 26 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 1a4b9b3..33471dd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,7 +19,7 @@ dependencies = [ Repository = "https://github.com/trobz/trobz_local" [project.scripts] -tlc = "trobz_local.main:app" +tlc = "trobz_local.main:cli" [dependency-groups] dev = [ diff --git a/trobz_local/main.py b/trobz_local/main.py index b0f068c..cd93e0f 100644 --- a/trobz_local/main.py +++ b/trobz_local/main.py @@ -1,8 +1,10 @@ import os import subprocess +import traceback from pathlib import Path from typing import Annotated +import click import git import typer from rich import print as rprint @@ -37,29 +39,71 @@ show_config_instructions, ) -app = typer.Typer() +app = typer.Typer( + rich_markup_mode="rich", + epilog="Docs & issues: https://github.com/trobz/trobz_local", +) + + +def _version_callback(value: bool) -> None: + if value: + from importlib.metadata import version + + typer.echo(f"tlc {version('trobz_local')}") + raise typer.Exit() @app.callback(invoke_without_command=True) def main( ctx: typer.Context, - newcomer: bool = typer.Option( - True, - help="Enable newcomer mode with confirmations and help.", - envvar="NEWCOMER_MODE", - ), - yes: bool = typer.Option( - False, - "--yes", - "-y", - help="Skip all confirmations (non-interactive mode).", - ), + version: Annotated[ + bool, + typer.Option( + "--version", + "-V", + callback=_version_callback, + is_eager=True, + help="Show version and exit.", + ), + ] = False, + newcomer: Annotated[ + bool, + typer.Option( + help="Enable newcomer mode with confirmations and help.", + envvar="NEWCOMER_MODE", + ), + ] = True, + yes: Annotated[ + bool, + typer.Option( + "--yes", + "-y", + help="Skip all confirmations (non-interactive mode).", + ), + ] = False, + debug: Annotated[ + bool, + typer.Option( + "--debug", + help="Show full stack traces on error.", + ), + ] = False, + quiet: Annotated[ + bool, + typer.Option( + "--quiet", + "-q", + help="Suppress informational output.", + ), + ] = False, ): """ Hi, I'm a CLI to help you setup and manage your local environment for Odoo development. """ ctx.ensure_object(dict) ctx.obj["newcomer"] = newcomer and not yes + ctx.obj["debug"] = debug + ctx.obj["quiet"] = quiet if ctx.invoked_subcommand is None: _run_init(ctx) @@ -112,7 +156,7 @@ def _run_init(ctx: typer.Context): odoo_versions = config.get("versions") if not odoo_versions: - typer.echo("versions not found in config file.") + typer.echo("versions not found in config file.", err=True) raise typer.Exit(code=1) for version in odoo_versions: @@ -206,9 +250,9 @@ def pull_repos( # noqa: C901 results = run_tasks(concurrency_tasks) failed_tasks = [res for res in results if not res.success] if failed_tasks: - typer.secho("\n--- Some repository operations failed ---", fg=typer.colors.RED) + typer.secho("\n--- Some repository operations failed ---", fg=typer.colors.RED, err=True) for res in failed_tasks: - typer.secho(f"✗ {res.name}: {res.message}", fg=typer.colors.RED) + typer.secho(f"✗ {res.name}: {res.message}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) else: typer.secho("\nAll repositories updated successfully.", fg=typer.colors.GREEN) @@ -334,7 +378,9 @@ def _run_installers( def install_tools( ctx: typer.Context, dry_run: bool = typer.Option(False, "--dry-run", help="Show what would be installed without executing."), - install_default_system_packages: bool = typer.Option(True, help="Install default OS system packages."), + no_default_packages: Annotated[ + bool, typer.Option("--no-default-packages", help="Skip default OS system packages.") + ] = False, ): """ Install tools using uv tool based on config. @@ -357,14 +403,16 @@ def install_tools( msg = _build_install_message(tools_config) confirm_step(ctx, msg, "install-tools") - all_results, any_failed = _run_installers(tools_config, dry_run, install_default_system_packages) + all_results, any_failed = _run_installers( + tools_config, dry_run, install_default_system_packages=not no_default_packages + ) if not dry_run: if any_failed: failed = [r for r in all_results if not r.success] - typer.secho("\n--- Some installations failed ---", fg=typer.colors.RED) + typer.secho("\n--- Some installations failed ---", fg=typer.colors.RED, err=True) for r in failed: - typer.secho(f"✗ {r.name}: {r.message}", fg=typer.colors.RED) + typer.secho(f"✗ {r.name}: {r.message}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) else: typer.secho("\n✓ All tools installed successfully.", fg=typer.colors.GREEN) @@ -426,10 +474,10 @@ def create_venvs(ctx: typer.Context): results = run_tasks(concurrency_tasks) failed_tasks = [res for res in results if not res.success] if failed_tasks: - typer.secho("\n--- Some virtual environment operations failed ---", fg=typer.colors.RED) + typer.secho("\n--- Some virtual environment operations failed ---", fg=typer.colors.RED, err=True) for res in failed_tasks: error_message = f"✗ {res.name}: {res.message}" - typer.secho(error_message, fg=typer.colors.RED) + typer.secho(error_message, fg=typer.colors.RED, err=True) raise typer.Exit(code=1) else: typer.secho("\nAll virtual environments created successfully.", fg=typer.colors.GREEN) @@ -502,7 +550,7 @@ def ensure_db_user(ctx: typer.Context): # Check PostgreSQL is running typer.echo("Checking PostgreSQL status...") if not check_postgres_running(): - typer.secho("✗ PostgreSQL is not running on localhost", fg=typer.colors.RED) + typer.secho("✗ PostgreSQL is not running on localhost", fg=typer.colors.RED, err=True) if system == "Darwin": typer.echo("Try: brew services start postgresql") elif system == "Linux": @@ -518,7 +566,7 @@ def ensure_db_user(ctx: typer.Context): typer.echo(f"User '{username}' not found, creating...") success, error_msg = create_user(username, password, system) if not success: - typer.secho(f"✗ Failed to create user '{username}'", fg=typer.colors.RED) + typer.secho(f"✗ Failed to create user '{username}'", fg=typer.colors.RED, err=True) if system == "Linux" and "sudo" in error_msg.lower(): typer.echo("Manual instructions:") typer.echo(" sudo -u postgres createuser -s odoo") @@ -531,14 +579,16 @@ def ensure_db_user(ctx: typer.Context): # Test connection typer.echo("Testing connection...") if not verify_connection(host, username, password): - typer.secho("✗ Connection test failed", fg=typer.colors.RED) + typer.secho("✗ Connection test failed", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) typer.secho("✓ Connection successful", fg=typer.colors.GREEN) typer.echo() typer.secho(f"✓ PostgreSQL user '{username}' is ready for Odoo development", fg=typer.colors.GREEN) typer.echo() - typer.secho("⚠️ WARNING: Using dev-only credentials (odoo:odoo). Never use in production!", fg=typer.colors.YELLOW) + typer.secho( + "⚠️ WARNING: Using dev-only credentials (odoo:odoo). Never use in production!", fg=typer.colors.YELLOW, err=True + ) _STATUS_ICONS = { @@ -550,6 +600,7 @@ def ensure_db_user(ctx: typer.Context): @app.command() def doctor(): + """Check local environment health and report issues.""" code_root = get_code_root() groups = run_doctor(code_root) @@ -581,3 +632,20 @@ def doctor(): if has_fail: raise typer.Exit(code=1) + + +def cli() -> None: + """Entry point wrapper with top-level exception handling.""" + try: + app() + except Exception as e: + # typer.Exit and SystemExit are not caught by bare `except Exception` + # so this only catches unexpected errors + ctx = click.get_current_context(silent=True) + debug = ctx.obj.get("debug", False) if ctx and ctx.obj else False + if debug: + typer.secho(traceback.format_exc(), fg=typer.colors.RED, err=True) + else: + typer.secho(f"Error: {e}", fg=typer.colors.RED, err=True) + typer.secho("Run with --debug for full stack trace.", err=True) + raise SystemExit(1) from e From 6f753057b0adbcd31f092c68ef1dc671d52b29c0 Mon Sep 17 00:00:00 2001 From: trisdoan Date: Wed, 4 Mar 2026 10:51:27 +0700 Subject: [PATCH 5/5] fix(output): redirect error and warning messages to stderr - Add err=True to all typer.secho() calls for error/warning output - Route progress bar output to stderr via Console(stderr=True) - Add err=True to doctor command docstring - Ensures proper output stream separation between stdout and stderr --- trobz_local/concurrency.py | 2 ++ trobz_local/installers.py | 22 +++++++++++++--------- trobz_local/utils.py | 10 +++++----- 3 files changed, 20 insertions(+), 14 deletions(-) diff --git a/trobz_local/concurrency.py b/trobz_local/concurrency.py index f98c237..1b7bdf6 100644 --- a/trobz_local/concurrency.py +++ b/trobz_local/concurrency.py @@ -1,6 +1,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass +from rich.console import Console from rich.progress import BarColumn, Progress, TextColumn @@ -21,6 +22,7 @@ def run_tasks(tasks, max_workers: int = 4): TextColumn("[progress.description]{task.description}"), BarColumn(), TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), + console=Console(stderr=True), ) as progress: overall = progress.add_task(f"[cyan]0/{total} done", total=total) diff --git a/trobz_local/installers.py b/trobz_local/installers.py index cce5efe..866dd02 100644 --- a/trobz_local/installers.py +++ b/trobz_local/installers.py @@ -151,7 +151,7 @@ def _run_package_install(cmd: list[str], packages: list[str]) -> bool: try: subprocess.run(full_cmd, check=True, text=True) # noqa: S603 except subprocess.CalledProcessError as e: - typer.secho(f"Error installing packages: {e}", fg=typer.colors.RED) + typer.secho(f"Error installing packages: {e}", fg=typer.colors.RED, err=True) return False return True @@ -165,11 +165,11 @@ def install_system_packages(packages: list[str], dry_run: bool = False, install_ if config is None: if system == "Darwin": - typer.secho("Error: Homebrew is not installed. Please install it first.", fg=typer.colors.RED) + typer.secho("Error: Homebrew is not installed. Please install it first.", fg=typer.colors.RED, err=True) elif system == "Linux": - typer.secho(f"Error: Unsupported Linux distribution: {distro}", fg=typer.colors.RED) + typer.secho(f"Error: Unsupported Linux distribution: {distro}", fg=typer.colors.RED, err=True) else: - typer.secho(f"Error: Unsupported operating system: {system}", fg=typer.colors.RED) + typer.secho(f"Error: Unsupported operating system: {system}", fg=typer.colors.RED, err=True) return False cmd, default_packages = config @@ -232,7 +232,9 @@ def setup_postgresql_repo() -> bool: # Get distribution codename (e.g. "jammy", "bookworm") lsb_release_path = shutil.which("lsb_release") if not lsb_release_path: - typer.secho("Warning: lsb_release not found, skipping PostgreSQL repo setup", fg=typer.colors.YELLOW) + typer.secho( + "Warning: lsb_release not found, skipping PostgreSQL repo setup", fg=typer.colors.YELLOW, err=True + ) return True result = subprocess.run([lsb_release_path, "-cs"], check=True, capture_output=True, text=True) # noqa: S603 @@ -240,12 +242,12 @@ def setup_postgresql_repo() -> bool: curl_path = shutil.which("curl") if not curl_path: - typer.secho("Warning: curl not found, skipping PostgreSQL repo setup", fg=typer.colors.YELLOW) + typer.secho("Warning: curl not found, skipping PostgreSQL repo setup", fg=typer.colors.YELLOW, err=True) return True gpg_path = shutil.which("gpg") if not gpg_path: - typer.secho("Warning: gpg not found, skipping PostgreSQL repo setup", fg=typer.colors.YELLOW) + typer.secho("Warning: gpg not found, skipping PostgreSQL repo setup", fg=typer.colors.YELLOW, err=True) return True # Download and import the official PGDG GPG key @@ -272,7 +274,7 @@ def setup_postgresql_repo() -> bool: tee_path = shutil.which("tee") if not tee_path: - typer.secho("Warning: tee not found, skipping PostgreSQL repo setup", fg=typer.colors.YELLOW) + typer.secho("Warning: tee not found, skipping PostgreSQL repo setup", fg=typer.colors.YELLOW, err=True) return True subprocess.run( # noqa: S603 @@ -294,9 +296,10 @@ def setup_postgresql_repo() -> bool: f"Warning: Failed to setup PostgreSQL repository: {e}\n" "You may need to configure it manually if you need PostgreSQL.", fg=typer.colors.YELLOW, + err=True, ) except Exception as e: - typer.secho(f"Warning: Unexpected error during PostgreSQL repo setup: {e}", fg=typer.colors.YELLOW) + typer.secho(f"Warning: Unexpected error during PostgreSQL repo setup: {e}", fg=typer.colors.YELLOW, err=True) return True # Always succeeds — never fails the install-tools pipeline @@ -327,6 +330,7 @@ def install_npm_packages(packages: list[str], dry_run: bool = False) -> list: typer.secho( "Error: npm is not installed. Please install Node.js first.", fg=typer.colors.RED, + err=True, ) return [TaskResult(name="npm-check", success=False, message="npm is not installed")] diff --git a/trobz_local/utils.py b/trobz_local/utils.py index 642d434..6d0b553 100644 --- a/trobz_local/utils.py +++ b/trobz_local/utils.py @@ -273,14 +273,14 @@ def get_code_root() -> Path: def get_uv_path(): uv_path = shutil.which("uv") if not uv_path: - typer.secho("Error: uv is not installed. Please install uv first.", fg=typer.colors.RED) + typer.secho("Error: uv is not installed. Please install uv first.", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) return uv_path def show_config_instructions(): content = files("trobz_local").joinpath("assets/odoo_dev.toml").read_text() - typer.secho("Config file not found.", fg=typer.colors.YELLOW) + typer.secho("Config file not found.", fg=typer.colors.YELLOW, err=True) code_root = get_code_root() typer.echo(f"Please create {code_root}/config.toml with content like this:") typer.echo(content) @@ -302,7 +302,7 @@ def get_config(): with open(config_path, "rb") as f: raw_config = tomli.load(f) except tomli.TOMLDecodeError as e: - typer.secho(f"Error: Invalid TOML in {config_path}", fg=typer.colors.RED) + typer.secho(f"Error: Invalid TOML in {config_path}", fg=typer.colors.RED, err=True) typer.echo(str(e)) raise typer.Exit(code=1) from e @@ -310,7 +310,7 @@ def get_config(): validated_config = ConfigModel(**raw_config) except ValidationError as e: for error in e.errors(): - typer.secho(f"{error['msg']}", fg=typer.colors.RED) + typer.secho(f"{error['msg']}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) from None return validated_config.model_dump() @@ -339,7 +339,7 @@ def confirm_step(ctx: typer.Context, message: str, command: str): If in newcomer mode, prints a help message and asks for confirmation to proceed. """ if ctx.obj.get("newcomer", False): - typer.secho(f"About to run: {command}", fg=typer.colors.BLUE) + typer.secho(f"About to run: {command}", fg=typer.colors.BLUE, err=True) rprint(message) if not typer.confirm("Do you want to proceed?"): raise typer.Abort()