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
235 lines
11 KiB
Python
235 lines
11 KiB
Python
# SPDX-License-Identifier: MIT
|
|
"""Tests for RDP process tracking."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import signal
|
|
|
|
from winpodx.core.process import (
|
|
TrackedProcess,
|
|
_cmdline_is_freerdp,
|
|
kill_session,
|
|
list_active_sessions,
|
|
)
|
|
|
|
|
|
class TestCmdlineIsFreerdp:
|
|
def test_argv0_xfreerdp(self):
|
|
assert _cmdline_is_freerdp(b"/usr/bin/xfreerdp3\0/v:host\0")
|
|
|
|
def test_argv0_flatpak_freerdp(self):
|
|
assert _cmdline_is_freerdp(b"flatpak\0run\0com.freerdp.FreeRDP\0/v:host\0")
|
|
|
|
def test_argv0_bwrap_wrapped_xfreerdp(self):
|
|
# `flatpak run com.freerdp.FreeRDP` re-execs (same PID) into
|
|
# `bwrap ... -- xfreerdp ...`; the wrapped client must still count or
|
|
# list_active_sessions() unlinks a live session's .cproc.
|
|
assert _cmdline_is_freerdp(
|
|
b"/usr/bin/bwrap\0--args\0072\0--\0xfreerdp\0/v:127.0.0.1:3390\0"
|
|
)
|
|
|
|
def test_argv0_bwrap_wrapped_flatpak_appid(self):
|
|
assert _cmdline_is_freerdp(b"bwrap\0--\0com.freerdp.FreeRDP\0/v:host\0")
|
|
|
|
def test_bwrap_without_freerdp_rejected(self):
|
|
# A bwrap sandbox around something else must not be adopted.
|
|
assert not _cmdline_is_freerdp(b"/usr/bin/bwrap\0--\0sleep\0900\0")
|
|
|
|
def test_freerdp_only_in_later_argv_rejected(self):
|
|
# Regression: "freerdp" in a later arg must not match.
|
|
assert not _cmdline_is_freerdp(
|
|
b"/usr/bin/python3\0-m\0pytest\0--deselect=test_freerdp_pid\0"
|
|
)
|
|
|
|
def test_freerdp_only_in_argv0_path_component_rejected(self):
|
|
assert not _cmdline_is_freerdp(b"/home/user/freerdp-notes/run.sh\0")
|
|
|
|
|
|
class TestTrackedProcess:
|
|
def test_dead_pid_not_alive(self):
|
|
proc = TrackedProcess(app_name="test", pid=99999999)
|
|
assert not proc.is_alive
|
|
|
|
def test_current_process_not_freerdp(self):
|
|
proc = TrackedProcess(app_name="test", pid=os.getpid())
|
|
assert not proc.is_alive
|
|
|
|
|
|
class TestListActiveSessions:
|
|
def test_empty_when_no_runtime_dir(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"winpodx.core.process.runtime_dir",
|
|
lambda: tmp_path / "nonexistent",
|
|
)
|
|
assert list_active_sessions() == []
|
|
|
|
def test_cleans_stale_cproc(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
cproc = tmp_path / "stale-app.cproc"
|
|
cproc.write_text("99999999")
|
|
sessions = list_active_sessions()
|
|
assert len(sessions) == 0
|
|
assert not cproc.exists()
|
|
|
|
def test_cleans_invalid_cproc(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
cproc = tmp_path / "bad.cproc"
|
|
cproc.write_text("not-a-number")
|
|
sessions = list_active_sessions()
|
|
assert len(sessions) == 0
|
|
assert not cproc.exists()
|
|
|
|
def test_lists_live_freerdp(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
monkeypatch.setattr("winpodx.core.process._pid_alive", lambda pid: True)
|
|
monkeypatch.setattr("winpodx.core.process.is_freerdp_pid", lambda pid: True)
|
|
cproc = tmp_path / "Notepad.cproc"
|
|
cproc.write_text("4242")
|
|
sessions = list_active_sessions()
|
|
assert [(s.app_name, s.pid) for s in sessions] == [("Notepad", 4242)]
|
|
assert cproc.exists()
|
|
|
|
def test_keeps_live_but_unrecognized_cproc(self, tmp_path, monkeypatch):
|
|
# Regression: a live PID we fail to recognise as FreeRDP (a new sandbox
|
|
# wrapper, or an old reader meeting a newly-wrapped client) must NOT be
|
|
# listed -- but its .cproc must NOT be deleted either. A reader deleting
|
|
# a live session's tracking file was the root cause of sessions
|
|
# vanishing / Terminate finding nothing.
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
monkeypatch.setattr("winpodx.core.process._pid_alive", lambda pid: True)
|
|
monkeypatch.setattr("winpodx.core.process.is_freerdp_pid", lambda pid: False)
|
|
cproc = tmp_path / "app.cproc"
|
|
cproc.write_text("4242")
|
|
sessions = list_active_sessions()
|
|
assert sessions == []
|
|
assert cproc.exists() # NOT deleted -- live PID, just unrecognised
|
|
|
|
|
|
class TestKillSession:
|
|
def test_returns_false_no_pidfile(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
assert kill_session("nonexistent") is False
|
|
|
|
def test_returns_false_dead_pid(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
cproc = tmp_path / "dead-app.cproc"
|
|
cproc.write_text("99999999")
|
|
assert kill_session("dead-app") is False
|
|
assert not cproc.exists()
|
|
|
|
def test_signals_whole_group_when_leader(self, tmp_path, monkeypatch):
|
|
# Launched with start_new_session=True -> PGID == PID -> kill the whole
|
|
# process group so the nested flatpak/bwrap/xfreerdp tree all dies.
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
monkeypatch.setattr("winpodx.core.process.is_freerdp_pid", lambda pid: True)
|
|
monkeypatch.setattr("winpodx.core.process.os.getpgid", lambda pid: pid) # leader
|
|
monkeypatch.setattr("winpodx.core.process.time.sleep", lambda s: None)
|
|
# alive for the guard, then dead -> polite SIGTERM only, no SIGKILL
|
|
alive = iter([True, False, False])
|
|
monkeypatch.setattr("winpodx.core.process._pid_alive", lambda pid: next(alive, False))
|
|
grp = []
|
|
monkeypatch.setattr(
|
|
"winpodx.core.process.os.killpg", lambda pgid, sig: grp.append((pgid, sig))
|
|
)
|
|
called_kill = []
|
|
monkeypatch.setattr(
|
|
"winpodx.core.process.os.kill", lambda pid, sig: called_kill.append(pid)
|
|
)
|
|
cproc = tmp_path / "app.cproc"
|
|
cproc.write_text("4242")
|
|
assert kill_session("app") is True
|
|
assert grp == [(4242, signal.SIGTERM)] # whole group, polite signal only
|
|
assert called_kill == [] # not the single-PID path
|
|
assert not cproc.exists()
|
|
|
|
def test_falls_back_to_single_pid_when_not_leader(self, tmp_path, monkeypatch):
|
|
# Session from an older winpodx (no start_new_session) -> PGID != PID ->
|
|
# signal only the PID, never an unrelated group.
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
monkeypatch.setattr("winpodx.core.process.is_freerdp_pid", lambda pid: True)
|
|
monkeypatch.setattr("winpodx.core.process.os.getpgid", lambda pid: 999) # not leader
|
|
monkeypatch.setattr("winpodx.core.process.time.sleep", lambda s: None)
|
|
alive = iter([True, False, False])
|
|
monkeypatch.setattr("winpodx.core.process._pid_alive", lambda pid: next(alive, False))
|
|
killpg_calls = []
|
|
monkeypatch.setattr(
|
|
"winpodx.core.process.os.killpg", lambda pgid, sig: killpg_calls.append(pgid)
|
|
)
|
|
kill_calls = []
|
|
monkeypatch.setattr(
|
|
"winpodx.core.process.os.kill", lambda pid, sig: kill_calls.append((pid, sig))
|
|
)
|
|
cproc = tmp_path / "app.cproc"
|
|
cproc.write_text("4242")
|
|
assert kill_session("app") is True
|
|
assert kill_calls == [(4242, signal.SIGTERM)]
|
|
assert killpg_calls == [] # never group-kill a group we don't lead
|
|
assert not cproc.exists()
|
|
|
|
def test_escalates_to_sigkill_when_term_ignored(self, tmp_path, monkeypatch):
|
|
# xfreerdp ignores/hangs on SIGTERM -> escalate to SIGKILL so the window
|
|
# is forced down ("Terminate does nothing" / window lingers on exit).
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
monkeypatch.setattr("winpodx.core.process.is_freerdp_pid", lambda pid: True)
|
|
monkeypatch.setattr("winpodx.core.process.os.getpgid", lambda pid: pid)
|
|
monkeypatch.setattr("winpodx.core.process.time.sleep", lambda s: None)
|
|
monkeypatch.setattr("winpodx.core.process._pid_alive", lambda pid: True) # never dies
|
|
sigs = []
|
|
monkeypatch.setattr("winpodx.core.process.os.killpg", lambda pgid, sig: sigs.append(sig))
|
|
cproc = tmp_path / "app.cproc"
|
|
cproc.write_text("4242")
|
|
assert kill_session("app") is True
|
|
assert signal.SIGTERM in sigs
|
|
assert signal.SIGKILL in sigs # escalated
|
|
assert not cproc.exists()
|
|
|
|
def test_expected_pid_mismatch_not_signalled(self, tmp_path, monkeypatch):
|
|
# Relaunch race: a reaper armed against PID 4242, but the .cproc now
|
|
# holds 5555 (a fresh session of the same app overwrote it). With
|
|
# expected_pid=4242 we must NOT reap the newcomer, and must leave its
|
|
# marker intact.
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
monkeypatch.setattr("winpodx.core.process._pid_alive", lambda pid: True)
|
|
monkeypatch.setattr("winpodx.core.process.is_freerdp_pid", lambda pid: True)
|
|
calls = []
|
|
monkeypatch.setattr("winpodx.core.process.os.kill", lambda pid, sig: calls.append(pid))
|
|
monkeypatch.setattr("winpodx.core.process.os.killpg", lambda pgid, sig: calls.append(pgid))
|
|
cproc = tmp_path / "app.cproc"
|
|
cproc.write_text("5555")
|
|
assert kill_session("app", expected_pid=4242) is False
|
|
assert calls == []
|
|
assert cproc.exists() # newcomer's marker left untouched
|
|
assert cproc.read_text() == "5555"
|
|
|
|
def test_expected_pid_match_signals(self, tmp_path, monkeypatch):
|
|
# Same PID the reaper armed against is still in the marker -> reap it.
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
monkeypatch.setattr("winpodx.core.process.is_freerdp_pid", lambda pid: True)
|
|
monkeypatch.setattr("winpodx.core.process.os.getpgid", lambda pid: pid)
|
|
monkeypatch.setattr("winpodx.core.process.time.sleep", lambda s: None)
|
|
alive = iter([True, False, False])
|
|
monkeypatch.setattr("winpodx.core.process._pid_alive", lambda pid: next(alive, False))
|
|
sigs = []
|
|
monkeypatch.setattr("winpodx.core.process.os.killpg", lambda pgid, sig: sigs.append(pgid))
|
|
cproc = tmp_path / "app.cproc"
|
|
cproc.write_text("4242")
|
|
assert kill_session("app", expected_pid=4242) is True
|
|
assert sigs == [4242]
|
|
assert not cproc.exists()
|
|
|
|
def test_reused_pid_not_signalled(self, tmp_path, monkeypatch):
|
|
# PID-reuse guard: a live PID that is NOT our FreeRDP client must never
|
|
# be signalled (we'd kill an innocent process). Just drop the stale file.
|
|
monkeypatch.setattr("winpodx.core.process.runtime_dir", lambda: tmp_path)
|
|
monkeypatch.setattr("winpodx.core.process._pid_alive", lambda pid: True)
|
|
monkeypatch.setattr("winpodx.core.process.is_freerdp_pid", lambda pid: False)
|
|
calls = []
|
|
monkeypatch.setattr("winpodx.core.process.os.kill", lambda pid, sig: calls.append(pid))
|
|
monkeypatch.setattr("winpodx.core.process.os.killpg", lambda pgid, sig: calls.append(pgid))
|
|
cproc = tmp_path / "app.cproc"
|
|
cproc.write_text("4242")
|
|
assert kill_session("app") is False
|
|
assert calls == []
|
|
assert not cproc.exists()
|