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
74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
# SPDX-License-Identifier: MIT
|
|
"""Tests for the RemoteApp session-interactive gate (#332)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from winpodx.core import rdp
|
|
from winpodx.core.config import Config
|
|
|
|
|
|
class _Exec:
|
|
def __init__(self, stdout: str) -> None:
|
|
self.rc = 0
|
|
self.stdout = stdout
|
|
self.stderr = ""
|
|
|
|
|
|
def _cfg() -> Config:
|
|
cfg = Config()
|
|
cfg.pod.backend = "podman"
|
|
return cfg
|
|
|
|
|
|
def test_gate_ready_returns_true_fast(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
class FakeClient:
|
|
def __init__(self, cfg): # noqa: ANN001
|
|
pass
|
|
|
|
def health(self):
|
|
return {"ok": True}
|
|
|
|
def exec(self, script, timeout=60): # noqa: ANN001
|
|
return _Exec("READY")
|
|
|
|
import winpodx.core.agent as agent_mod
|
|
|
|
monkeypatch.setattr(agent_mod, "AgentClient", FakeClient)
|
|
assert rdp._wait_session_interactive(_cfg(), timeout=5) is True
|
|
|
|
|
|
def test_gate_agent_down_returns_false_no_block(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import winpodx.core.agent as agent_mod
|
|
|
|
class FakeClient:
|
|
def __init__(self, cfg): # noqa: ANN001
|
|
pass
|
|
|
|
def health(self):
|
|
raise agent_mod.AgentUnavailableError("down")
|
|
|
|
monkeypatch.setattr(agent_mod, "AgentClient", FakeClient)
|
|
# Must return quickly (no polling) when the agent is unreachable.
|
|
assert rdp._wait_session_interactive(_cfg(), timeout=5) is False
|
|
|
|
|
|
def test_gate_times_out_when_locked(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
import winpodx.core.agent as agent_mod
|
|
|
|
class FakeClient:
|
|
def __init__(self, cfg): # noqa: ANN001
|
|
pass
|
|
|
|
def health(self):
|
|
return {"ok": True}
|
|
|
|
def exec(self, script, timeout=60): # noqa: ANN001
|
|
return _Exec("LOCKED")
|
|
|
|
monkeypatch.setattr(agent_mod, "AgentClient", FakeClient)
|
|
monkeypatch.setattr(rdp, "log", rdp.log) # keep
|
|
# timeout=1 so the loop exits fast; LOCKED never becomes READY.
|
|
assert rdp._wait_session_interactive(_cfg(), timeout=1) is False
|