#!/usr/bin/env python3
"""lib/shim_path.py — does a fresh login shell reach the shim?

Everything runs against a temp HOME whose rc files this test writes, with a fake shim
(`repo/bin/claude`, carrying the multiacc-shim header) and a fake real binary in
`~/.local/bin`. Real shells are started (zsh/bash/sh, whichever are installed), the
way a terminal or a `bash -lc` launcher starts them; nothing touches the operator's
rc files, and nothing here needs a Claude Code install.
"""

from __future__ import annotations

import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest

REPO = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(REPO / 'lib'))

import shim_path  # noqa: E402

MARK_BEGIN = '# >>> claude-multiacc >>>'
MARK_END = '# <<< claude-multiacc <<<'


def installed_block(repo_dir):
    """The exact block lib/install_actions.sh writes for repo_dir."""
    env = {**os.environ, 'REPO_DIR': str(repo_dir), 'MARK_BEGIN': MARK_BEGIN, 'MARK_END': MARK_END}
    return subprocess.check_output(
        ['bash', '-c', f'. "{REPO}/lib/install_actions.sh"; rc_block'], env=env, text=True)


def legacy_block(repo_dir):
    """The block as every install before the prompt hook wrote it."""
    lines = installed_block(repo_dir).splitlines()
    return '\n'.join(lines[:3] + [MARK_END]) + '\n'


class ShimPathTests(unittest.TestCase):
    def setUp(self):
        self.tmp = Path(tempfile.mkdtemp(prefix='multiacc-shim-path-'))
        self.addCleanup(shutil.rmtree, self.tmp, True)
        self.home = self.tmp / 'home'
        self.repo = self.tmp / 'repo'
        (self.home / '.local' / 'bin').mkdir(parents=True)
        (self.repo / 'bin').mkdir(parents=True)
        for name in shim_path.NAMES:
            shim = self.repo / 'bin' / name
            shim.write_text('#!/bin/sh\n# claude-multiacc-shim (fake for tests)\necho SHIM\n')
            shim.chmod(0o755)
            real = self.home / '.local' / 'bin' / name
            real.write_text('#!/bin/sh\necho REAL\n')
            real.chmod(0o755)
        self.shells = [s for s in ('zsh', 'bash', 'sh') if shim_path.shell_path(s)]
        self.assertTrue(self.shells, 'no shell to test with')
        os.environ.pop('CLAUDE_MULTIACC_PATH_PROBE', None)

    def write(self, name, *parts):
        (self.home / name).write_text(''.join(parts))

    def probe(self):
        return shim_path.probe(str(self.repo), home=str(self.home), user='tester', timeout=40)

    def by_mode(self, rows):
        return {row['mode']: row for row in rows}

    def all_modes(self, rows, name, expect_ok):
        for row in rows:
            if row['error'] == 'shell not installed':
                continue
            with self.subTest(mode=row['mode'], name=name):
                self.assertIsNone(row['error'], row)
                self.assertIs(row['results'][name]['ok'], expect_ok, row)

    def test_no_rc_files_means_this_repo_is_never_reached(self):
        # The SYSTEM rc files still run (macOS path_helper adds /opt/homebrew/bin; a Linux
        # root install's /etc/profile.d puts ITS shim first), so a real binary — or another
        # install's shim — may be found. What must never be found is this repo's shim.
        rows = self.probe()
        for row in rows:
            if row['error'] == 'shell not installed':
                continue
            self.assertIsNone(row['error'], row)
            for name in shim_path.NAMES:
                self.assertNotEqual(row['results'][name]['resolved'], str(self.repo / 'bin' / name))
        text = '\n'.join(shim_path.render(rows))
        if shim_path.bypassed(rows):
            self.assertRegex(text, r'NOT ON PATH|BYPASSED')
            self.assertIn('claude-multiacc install', text)

    def test_installed_block_wins_in_every_shell(self):
        block = installed_block(self.repo)
        prepend = 'export PATH="$HOME/.local/bin:$PATH"\n'
        self.write('.zshenv', block)
        self.write('.zprofile', prepend, block)
        self.write('.zshrc', prepend, block)
        self.write('.profile', prepend, block)
        self.write('.bashrc', prepend, block)
        rows = self.probe()
        for name in shim_path.NAMES:
            self.all_modes(rows, name, True)
        self.assertEqual(shim_path.bypassed(rows), [])
        text = '\n'.join(shim_path.render(rows))
        self.assertNotIn('BYPASSED', text)
        self.assertIn('claude: OK', text)

    def test_real_binary_prepended_after_the_block_is_reported(self):
        block = installed_block(self.repo)
        late = 'export PATH="$HOME/.local/bin:$PATH"\n'
        self.write('.zshenv', block, late)
        self.write('.zprofile', block, late)
        self.write('.zshrc', block, late)
        self.write('.profile', block, late)
        rows = self.probe()
        real = str(self.home / '.local' / 'bin' / 'claude')
        # Non-interactive login shells see rc order alone: bypassed.
        for mode, row in self.by_mode(rows).items():
            if row['error'] == 'shell not installed':
                continue
            if mode.endswith('-lc'):
                self.assertEqual(row['results']['claude']['resolved'], real, row)
                self.assertFalse(row['results']['claude']['ok'])
            else:
                # Interactive shells reach a prompt: the hook puts the shim back first.
                self.assertTrue(row['results']['claude']['ok'], row)
        bad = shim_path.bypassed(rows)
        self.assertTrue(bad)
        self.assertTrue(all(mode.endswith('-lc') for mode, _n, _r in bad), bad)
        text = '\n'.join(shim_path.render(rows))
        self.assertIn(f'BYPASSED -> {real}', text)
        self.assertIn('OUTSIDE the pool', text)

    @unittest.skipUnless(shim_path.shell_path('zsh'), 'zsh required')
    def test_interrupted_zshrc_is_the_incident_and_the_hook_fixes_it(self):
        """my-mini 2026-09-17: ~/.zshrc stopped (Ctrl-C at the conda hook) before the
        multiacc block, ~/.local/bin stayed in front, `claude` at that prompt ran the
        real binary. With the legacy block the probe must SEE that; with the hooked
        block the prompt must hand `claude` to the shim again."""
        prepend = 'export PATH="$HOME/.local/bin:$PATH"\n'
        for block, expect_ok in ((legacy_block(self.repo), False), (installed_block(self.repo), True)):
            self.write('.zshenv', block)
            self.write('.zprofile', prepend, block)
            self.write('.zshrc', prepend, 'return 0  # interrupted before the block\n', block)
            rows = self.by_mode(self.probe())
            with self.subTest(hooked=expect_ok):
                self.assertIs(rows['zsh -li']['results']['claude']['ok'], expect_ok, rows['zsh -li'])
                # `zsh -lc` never reads ~/.zshrc, so it is fine either way.
                self.assertTrue(rows['zsh -lc']['results']['claude']['ok'], rows['zsh -lc'])

    def test_a_symlink_to_the_shim_counts_as_the_shim(self):
        link_dir = self.tmp / 'usr-local-bin'
        link_dir.mkdir()
        os.symlink(self.repo / 'bin' / 'claude', link_dir / 'claude')
        self.assertTrue(shim_path.is_shim(str(link_dir / 'claude'), str(self.repo), 'claude'))
        copy = self.tmp / 'copy'
        copy.write_text('#!/bin/sh\n# claude-multiacc-shim copied elsewhere\n')
        self.assertTrue(shim_path.is_shim(str(copy), str(self.repo), 'claude'))
        self.assertFalse(shim_path.is_shim(str(self.home / '.local' / 'bin' / 'claude'),
                                           str(self.repo), 'claude'))
        self.assertFalse(shim_path.is_shim(None, str(self.repo), 'claude'))
        self.assertFalse(shim_path.is_shim(str(self.tmp / 'missing'), str(self.repo), 'claude'))

    def test_rc_file_that_hangs_or_exits_is_an_error_not_a_verdict(self):
        self.write('.profile', 'exit 3\n')
        self.write('.zprofile', 'exit 3\n')
        self.write('.zshrc', 'exit 3\n')
        rows = self.by_mode(self.probe())
        for mode in ('bash -lc', 'sh -lc', 'zsh -lc'):
            if mode in rows and rows[mode]['error'] != 'shell not installed':
                self.assertIn('without answering', rows[mode]['error'], rows[mode])
                self.assertEqual(rows[mode]['results'], {})
        self.write('.profile', 'sleep 30\n')
        row = shim_path._run_mode('bash', '-lc', False, shim_path.NAMES, str(self.repo),
                                  str(self.home), 'tester', timeout=1)
        if row['error'] != 'shell not installed':
            self.assertIn('no answer within 1s', row['error'])
        text = '\n'.join(shim_path.render([row]))
        self.assertIn('could not be probed', text)

    def test_kill_switch_and_cli_exit_codes(self):
        os.environ['CLAUDE_MULTIACC_PATH_PROBE'] = '0'
        try:
            self.assertEqual(self.probe(), [])
            self.assertIn('probe disabled', '\n'.join(shim_path.render([])))
        finally:
            del os.environ['CLAUDE_MULTIACC_PATH_PROBE']
        env = {**os.environ, 'HOME': str(self.home), 'USER': 'tester'}
        env.pop('CLAUDE_MULTIACC_PATH_PROBE', None)
        cli = [sys.executable, str(REPO / 'lib' / 'shim_path.py')]
        self.assertEqual(subprocess.run(cli, env=env, capture_output=True).returncode, 2)
        self.assertEqual(subprocess.run(cli + [str(self.repo), '--timeout=x'], env=env,
                                        capture_output=True).returncode, 2)
        # nothing in HOME -> this repo's shim is not reached -> 1 (unless a system-wide
        # rc file on this host puts some other install's shim first, see above)
        r = subprocess.run(cli + [str(self.repo), '--json'], env=env, capture_output=True, text=True)
        import json
        doc = json.loads(r.stdout)
        self.assertEqual(r.returncode, 1 if doc['bypassed'] else 0, r.stderr)
        self.assertEqual(doc['repo_dir'], str(self.repo))
        block = installed_block(self.repo)
        for rc in ('.zshenv', '.zprofile', '.zshrc', '.profile'):
            self.write(rc, block)
        r = subprocess.run(cli + [str(self.repo)], env=env, capture_output=True, text=True)
        self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
        self.assertIn('claude: OK', r.stdout)

    def test_probe_leaves_shell_history_alone(self):
        block = installed_block(self.repo)
        for rc in ('.zshenv', '.zprofile', '.zshrc', '.profile'):
            self.write(rc, block)
        before = sorted(p.name for p in self.home.iterdir())
        self.probe()
        after = sorted(p.name for p in self.home.iterdir())
        self.assertEqual(before, after)
        for hist in ('.zsh_history', '.bash_history', '.sh_history'):
            self.assertFalse((self.home / hist).exists(), hist)

    def test_rc_that_reads_stdin_costs_an_answer_not_a_verdict(self):
        """A prompt or `read` in an rc file eats one line of the fed script. The lost
        answer must show as a row error, never as NOT ON PATH — that would page."""
        block = installed_block(self.repo)
        eater = 'read -r _eaten\n'
        self.write('.zshenv', block)
        self.write('.zprofile', block)
        self.write('.zshrc', block, eater)
        self.write('.profile', block)
        self.write('.bashrc', block, eater)
        rows = self.by_mode(self.probe())
        for mode in ('zsh -li', 'bash -li'):
            row = rows.get(mode)
            if not row or row['error'] == 'shell not installed':
                continue
            with self.subTest(mode=mode):
                oks = [r['ok'] for r in row['results'].values()]
                self.assertNotIn(False, oks, row)
                # the eater swallowed the history line, the sentinel answers survived
                self.assertTrue(row['results'], row)
        self.assertEqual(shim_path.bypassed(list(rows.values())), [])
        # ...and an rc that tries to eat EVERY line: depending on how the shell buffers a
        # piped stdin the probe either still answers or reads as unanswered — either way
        # every name is OK or unknown, never a bypass.
        self.write('.zshrc', block, eater * 6)
        self.write('.bashrc', block, eater * 6)
        rows = self.by_mode(self.probe())
        for mode in ('zsh -li', 'bash -li'):
            row = rows.get(mode)
            if not row or row['error'] == 'shell not installed':
                continue
            if row['results']:
                self.assertNotIn(False, [r['ok'] for r in row['results'].values()], row)
            else:
                self.assertIn('without answering', row['error'], row)
        self.assertEqual(shim_path.bypassed(list(rows.values())), [])

    def test_rc_noise_does_not_break_the_report(self):
        block = installed_block(self.repo)
        noise = ("printf 'claude=/nowhere\\n'\n"                 # a bare line that looks like an answer
                 "printf '\\xff\\xfe not utf-8 \\xc3\\n'\n"   # a latin-1 banner
                 "sleep 3 &\n")                                     # a child holding stdout open
        for rc in ('.zshenv', '.zprofile', '.zshrc', '.profile'):
            self.write(rc, noise, block)
        started = __import__('time').monotonic()
        rows = shim_path.probe(str(self.repo), home=str(self.home), user='tester', timeout=20)
        elapsed = __import__('time').monotonic() - started
        for name in shim_path.NAMES:
            self.all_modes(rows, name, True)
        self.assertEqual(shim_path.bypassed(rows), [])
        # the backgrounded child must not have cost the timeout
        self.assertLess(elapsed, 15, elapsed)

    def test_sh_alias_in_profile_is_not_a_bypass(self):
        block = installed_block(self.repo)
        self.write('.profile', block, "alias claude='claude --dangerously-skip-permissions'\n")
        rows = self.by_mode(self.probe())
        row = rows.get('sh -lc')
        if row and row['error'] != 'shell not installed':
            self.assertTrue(row['results']['claude']['ok'], row)


if __name__ == '__main__':
    unittest.main()
