77 lines
2.5 KiB
Python
77 lines
2.5 KiB
Python
"""Tests for host session list/parse helpers."""
|
|
|
|
from app.services.host_sessions import (
|
|
event_supports_session_terminate,
|
|
filter_logind_session_rows,
|
|
parse_loginctl_sessions,
|
|
parse_loginctl_sessions_json,
|
|
parse_qwinsta_sessions,
|
|
)
|
|
|
|
|
|
def test_parse_loginctl_sessions():
|
|
stdout = "c1 1000 alice seat0 pts/0 active -\n 2 1001 bob - tty2 active -\n"
|
|
rows = parse_loginctl_sessions(stdout)
|
|
assert len(rows) == 2
|
|
assert rows[0].session_id == "c1"
|
|
assert rows[0].user == "alice"
|
|
assert rows[0].tty == "pts/0"
|
|
|
|
|
|
def test_parse_loginctl_sessions_prefers_pts_over_ephemeral_duplicate():
|
|
stdout = """21166 1000 papatramp - pts/0 active no -
|
|
21169 1000 papatramp - - active no -
|
|
"""
|
|
rows = parse_loginctl_sessions(stdout)
|
|
assert len(rows) == 1
|
|
assert rows[0].session_id == "21166"
|
|
assert rows[0].tty == "pts/0"
|
|
|
|
|
|
def test_parse_loginctl_sessions_json():
|
|
stdout = """[
|
|
{"session":"21166","uid":1000,"user":"papatramp","seat":null,"tty":"pts/0","state":"active","idle":false,"since":null},
|
|
{"session":"21169","uid":1000,"user":"papatramp","seat":null,"tty":null,"state":"active","idle":false,"since":null}
|
|
]"""
|
|
rows = parse_loginctl_sessions_json(stdout)
|
|
assert len(rows) == 1
|
|
assert rows[0].session_id == "21166"
|
|
assert rows[0].tty == "pts/0"
|
|
|
|
|
|
def test_filter_logind_session_rows_keeps_distinct_users():
|
|
from app.services.host_sessions import HostSessionRow
|
|
|
|
rows = filter_logind_session_rows(
|
|
[
|
|
HostSessionRow(session_id="1", user="alice", tty="pts/0", state="active"),
|
|
HostSessionRow(session_id="2", user="bob", tty=None, state="active"),
|
|
]
|
|
)
|
|
assert len(rows) == 2
|
|
|
|
|
|
def test_parse_qwinsta_sessions_filters_user():
|
|
stdout = """SESSIONNAME USERNAME ID STATE
|
|
console Administrator 1 Active
|
|
rdp-tcp#0 B26\\alice 2 Active
|
|
"""
|
|
all_rows = parse_qwinsta_sessions(stdout)
|
|
assert len(all_rows) == 2
|
|
filtered = parse_qwinsta_sessions(stdout, filter_user="alice")
|
|
assert len(filtered) == 1
|
|
assert filtered[0].session_id == "2"
|
|
|
|
|
|
def test_event_supports_session_terminate_types():
|
|
class HostStub:
|
|
os_family = "linux"
|
|
product = "ssh-monitor"
|
|
|
|
class EventStub:
|
|
def __init__(self, event_type: str):
|
|
self.type = event_type
|
|
self.host = HostStub()
|
|
|
|
assert event_supports_session_terminate(EventStub("ssh.login.success")) is True
|