TimeCapsuleSMB/src/timecapsulesmb/cli/doctor.py at main · jamesyc/TimeCapsuleSMB · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from __future__ import annotations

import argparse
from typing import Optional

from timecapsulesmb.checks.doctor import run_doctor_checks
from timecapsulesmb.checks.models import CheckResult
from timecapsulesmb.cli.context import CommandContext
from timecapsulesmb.cli.runtime import add_config_argument, print_json
from timecapsulesmb.cli.util import color_green, color_red
from timecapsulesmb.identity import ensure_install_id
from timecapsulesmb.services.doctor import build_doctor_error, doctor_status_counts
from timecapsulesmb.services.runtime import load_env_config
from timecapsulesmb.telemetry import TelemetryClient
from timecapsulesmb.core.paths import resolve_app_paths


def print_result(result: CheckResult) -> None:
    status = result.status
    if status == "PASS":
        status = color_green(status)
    elif status == "FAIL":
        status = color_red(status)
    print(f"{status} {result.message}")


def print_followup_help() -> None:
    print("")
    print("Some troubleshooting tips:")
    print("- (To remove old Apple devices entries from mDNS cache) try running:")
    print("    sudo dscacheutil -flushcache && sudo killall -HUP mDNSResponder")
    print("- (If you have disk corruption issues, or error 22) then try running:")
    print("    .venv/bin/tcapsule fsck")
    print("- (If you have xattr issues, or macOS Error -50) then try running:")
    print("    .venv/bin/tcapsule repair-xattrs")


def main(argv: Optional[list[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="Run local diagnostics for the current TimeCapsuleSMB setup.")
    add_config_argument(parser)
    parser.add_argument("--skip-ssh", action="store_true", help="Skip SSH reachability checks")
    parser.add_argument("--skip-bonjour", action="store_true", help="Skip Bonjour browse/resolve checks")
    parser.add_argument("--skip-smb", action="store_true", help="Skip authenticated SMB listing check")
    parser.add_argument("--json", action="store_true", help="Output doctor results as JSON")
    args = parser.parse_args(argv)

    ensure_install_id()
    app_paths = resolve_app_paths(config_path=args.config)
    config = load_env_config(env_path=args.config)
    telemetry = TelemetryClient.from_config(config)
    with CommandContext(telemetry, "doctor", "doctor_started", "doctor_finished", config=config, args=args) as command_context:
        command_context.update_fields(
            skip_ssh=args.skip_ssh,
            skip_bonjour=args.skip_bonjour,
            skip_smb=args.skip_smb,
            json_output=args.json,
        )
        if not args.skip_ssh and config.has_value("TC_HOST"):
            command_context.set_stage("resolve_connection")
            connection = command_context.resolve_env_connection(allow_empty_password=True)
            if connection.password:
                command_context.start_optional_airport_identity_probe(connection)
        command_context.set_stage("run_checks")
        doctor_debug: dict[str, object] = {}
        results, fatal = run_doctor_checks(
            config,
            repo_root=app_paths.distribution_root,
            connection=command_context.connection,
            precomputed_interface_probe=command_context.interface_probe,
            precomputed_probe_state=command_context.probe_state,
            skip_ssh=args.skip_ssh,
            skip_bonjour=args.skip_bonjour,
            skip_smb=args.skip_smb,
            on_result=None if args.json else print_result,
            debug_fields=doctor_debug,
        )
        command_context.add_debug_fields(**doctor_debug)
        status_counts = doctor_status_counts(results)
        command_context.update_fields(
            fatal=fatal,
            check_count=len(results),
            pass_count=status_counts["PASS"],
            warn_count=status_counts["WARN"],
            fail_count=status_counts["FAIL"],
            info_count=status_counts["INFO"],
        )

        if args.json:
            command_context.set_stage("render_json")
            print_json({
                "fatal": fatal,
                "results": [{"status": result.status, "message": result.message} for result in results],
                "summary": "doctor found one or more fatal problems." if fatal else "doctor checks passed.",
            })
            if fatal:
                error = build_doctor_error(results, command_context.debug_fields)
                if error:
                    command_context.set_error(error)
                command_context.fail()
            else:
                command_context.succeed()
            return 1 if fatal else 0

        command_context.set_stage("render_results")
        if fatal:
            print("\nSummary: doctor found one or more fatal problems.")
            print_followup_help()
            error = build_doctor_error(results, command_context.debug_fields)
            if error:
                command_context.set_error(error)
            command_context.fail()
            return 1

        print("\nSummary: doctor checks passed.")
        print_followup_help()
        command_context.succeed()
        return 0
    return 1