Files
kernalix7--winpodx/tests/test_sync_password_agent_first.py
wehub-resource-sync 3e779be6f3
CI / lint (push) Failing after 13m4s
CI / test (3.11, ubuntu-latest) (push) Failing after 2m4s
CI / test (3.13, ubuntu-latest) (push) Successful in 13m30s
CI / test (3.14, ubuntu-latest) (push) Successful in 17m21s
CI / test (3.12, ubuntu-latest) (push) Successful in 17m55s
CI / discover-apps-ps (push) Successful in 1m56s
CI / test (3.9, ubuntu-latest) (push) Successful in 13m17s
CI / test (3.10, ubuntu-latest) (push) Successful in 26m21s
CI / audit (push) Successful in 13m38s
Deploy site / deploy (push) Has been cancelled
CI / test (3.14, ubuntu-24.04-arm) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:32:37 +08:00

199 lines
6.6 KiB
Python

# SPDX-License-Identifier: MIT
"""Agent-first behaviour for ``winpodx pod sync-password``.
Same drive-redirect bug that affected ``rotate-password`` blocked the
recovery path; ``_sync_password`` now prefers AgentTransport so a user
whose rotation broke isn't stuck without a recovery path.
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
def _ok_result():
from winpodx.core.transport.base import ExecResult
return ExecResult(rc=0, stdout="password reset\n", stderr="")
def _fail_result(rc: int = 1, stderr: str = "boom"):
from winpodx.core.transport.base import ExecResult
return ExecResult(rc=rc, stdout="", stderr=stderr)
@pytest.fixture()
def _cfg(tmp_path, monkeypatch):
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
from winpodx.core.config import Config
cfg = Config()
cfg.rdp.user = "User"
cfg.rdp.password = "target-pw"
cfg.pod.backend = "podman"
cfg.save()
return cfg
def test_agent_ok_skips_freerdp_and_recovery_prompt(_cfg, monkeypatch):
"""Happy path: agent reachable, no recovery password prompt, no FreeRDP."""
from winpodx.cli import pod as cli_pod
transport = MagicMock()
transport.exec.return_value = _ok_result()
dispatch = MagicMock(return_value=transport)
monkeypatch.setattr("winpodx.core.transport.dispatch", dispatch)
rin = MagicMock()
monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", rin)
# Hard fail if either prompt path runs.
def _no_prompt(*_a, **_kw):
raise AssertionError("getpass should not be called when agent is healthy")
monkeypatch.setattr("getpass.getpass", _no_prompt)
cli_pod._sync_password(non_interactive=False)
transport.exec.assert_called_once()
_, kwargs = transport.exec.call_args
assert kwargs.get("description") == "sync-password"
assert kwargs.get("timeout") == 90
rin.assert_not_called()
def test_agent_unavailable_falls_back_to_freerdp(_cfg, monkeypatch):
"""Agent down → prompt for recovery pw → FreeRDP path with rescue cfg."""
from winpodx.cli import pod as cli_pod
from winpodx.core.transport.base import TransportUnavailable
monkeypatch.setattr(
"winpodx.core.transport.dispatch",
MagicMock(side_effect=TransportUnavailable("agent down")),
)
rin = MagicMock(return_value=_ok_result())
monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", rin)
monkeypatch.setattr("getpass.getpass", lambda *_a, **_kw: "recovery-pw")
cli_pod._sync_password(non_interactive=False)
rin.assert_called_once()
args, kwargs = rin.call_args
rescue_cfg = args[0]
assert rescue_cfg.rdp.password == "recovery-pw"
assert "net user" in args[1]
assert "target-pw" in args[1] # the value from cfg, not the recovery
assert kwargs.get("description") == "sync-password"
assert kwargs.get("timeout") == 120
def test_agent_auth_error_does_not_fall_back(_cfg, monkeypatch, capsys):
"""TransportAuthError on the agent path is fatal, no FreeRDP retry."""
from winpodx.cli import pod as cli_pod
from winpodx.core.transport.base import TransportAuthError
transport = MagicMock()
transport.exec.side_effect = TransportAuthError("401")
monkeypatch.setattr("winpodx.core.transport.dispatch", MagicMock(return_value=transport))
rin = MagicMock()
monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", rin)
def _no_prompt(*_a, **_kw):
raise AssertionError("recovery prompt should not run on agent auth failure")
monkeypatch.setattr("getpass.getpass", _no_prompt)
with pytest.raises(SystemExit) as exc:
cli_pod._sync_password(non_interactive=False)
assert exc.value.code == 3
rin.assert_not_called()
captured = capsys.readouterr()
assert "auth" in captured.out.lower()
def test_agent_nonzero_rc_returns_failure_without_fallback(_cfg, monkeypatch):
"""Script-level failure on the agent path is reported, not retried via FreeRDP."""
from winpodx.cli import pod as cli_pod
transport = MagicMock()
transport.exec.return_value = _fail_result(rc=2, stderr="net user failed")
monkeypatch.setattr("winpodx.core.transport.dispatch", MagicMock(return_value=transport))
rin = MagicMock()
monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", rin)
with pytest.raises(SystemExit) as exc:
cli_pod._sync_password(non_interactive=False)
assert exc.value.code == 3
rin.assert_not_called()
def test_agent_unavailable_non_interactive_uses_env_recovery_pw(_cfg, monkeypatch):
"""Fallback path in non-interactive mode reads WINPODX_RECOVERY_PASSWORD."""
from winpodx.cli import pod as cli_pod
from winpodx.core.transport.base import TransportUnavailable
monkeypatch.setattr(
"winpodx.core.transport.dispatch",
MagicMock(side_effect=TransportUnavailable("agent down")),
)
monkeypatch.setenv("WINPODX_RECOVERY_PASSWORD", "env-recovery")
rin = MagicMock(return_value=_ok_result())
monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", rin)
cli_pod._sync_password(non_interactive=True)
rin.assert_called_once()
rescue_cfg = rin.call_args.args[0]
assert rescue_cfg.rdp.password == "env-recovery"
def test_agent_unavailable_non_interactive_without_env_exits(_cfg, monkeypatch):
from winpodx.cli import pod as cli_pod
from winpodx.core.transport.base import TransportUnavailable
monkeypatch.setattr(
"winpodx.core.transport.dispatch",
MagicMock(side_effect=TransportUnavailable("agent down")),
)
monkeypatch.delenv("WINPODX_RECOVERY_PASSWORD", raising=False)
rin = MagicMock()
monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", rin)
with pytest.raises(SystemExit) as exc:
cli_pod._sync_password(non_interactive=True)
assert exc.value.code == 2
rin.assert_not_called()
def test_agent_unavailable_freerdp_channel_failure_exits_3(_cfg, monkeypatch):
"""If both transports fail, surface an exit-3 error with the recovery hint."""
from winpodx.cli import pod as cli_pod
from winpodx.core.transport.base import TransportUnavailable
from winpodx.core.windows_exec import WindowsExecError
monkeypatch.setattr(
"winpodx.core.transport.dispatch",
MagicMock(side_effect=TransportUnavailable("agent down")),
)
monkeypatch.setattr("getpass.getpass", lambda *_a, **_kw: "recovery-pw")
rin = MagicMock(side_effect=WindowsExecError("freerdp died"))
monkeypatch.setattr("winpodx.core.windows_exec.run_in_windows", rin)
with pytest.raises(SystemExit) as exc:
cli_pod._sync_password(non_interactive=False)
assert exc.value.code == 3