feat: initial Moongreet greeter implementation

greetd-Greeter für Wayland mit Python + GTK4 + gtk4-layer-shell.
Enthält IPC-Protokoll, User/Session-Erkennung, Power-Actions,
komplettes UI-Layout und 36 Tests (Unit + Integration).
This commit is contained in:
2026-03-26 09:47:19 +01:00
commit 87c2e7d9c8
21 changed files with 1610 additions and 0 deletions
View File
+170
View File
@@ -0,0 +1,170 @@
# ABOUTME: Integration tests — verifies the login flow end-to-end via a mock greetd socket.
# ABOUTME: Tests the IPC sequence: create_session → post_auth → start_session.
import json
import os
import socket
import struct
import threading
from pathlib import Path
import pytest
from moongreet.ipc import create_session, post_auth_response, start_session, cancel_session
class MockGreetd:
"""A mock greetd server that listens on a Unix socket and responds to IPC messages."""
def __init__(self, sock_path: Path) -> None:
self.sock_path = sock_path
self._responses: list[dict] = []
self._received: list[dict] = []
self._server: socket.socket | None = None
def expect(self, response: dict) -> None:
"""Queue a response to send for the next received message."""
self._responses.append(response)
@property
def received(self) -> list[dict]:
return self._received
def start(self) -> None:
"""Start the mock server in a background thread."""
self._server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self._server.bind(str(self.sock_path))
self._server.listen(1)
self._thread = threading.Thread(target=self._serve, daemon=True)
self._thread.start()
def _serve(self) -> None:
conn, _ = self._server.accept()
try:
for response in self._responses:
# Receive a message
header = conn.recv(4)
if len(header) < 4:
break
length = struct.unpack("!I", header)[0]
payload = conn.recv(length)
msg = json.loads(payload.decode("utf-8"))
self._received.append(msg)
# Send response
resp_payload = json.dumps(response).encode("utf-8")
conn.sendall(struct.pack("!I", len(resp_payload)) + resp_payload)
finally:
conn.close()
def stop(self) -> None:
if self._server:
self._server.close()
class TestLoginFlow:
"""Integration tests for the complete login flow via mock greetd."""
def test_successful_login(self, tmp_path: Path) -> None:
"""Simulate a complete successful login: create → auth → start."""
sock_path = tmp_path / "greetd.sock"
mock = MockGreetd(sock_path)
mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
mock.expect({"type": "success"})
mock.expect({"type": "success"})
mock.start()
try:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect(str(sock_path))
# Step 1: Create session
response = create_session(sock, "dominik")
assert response["type"] == "auth_message"
# Step 2: Send password
response = post_auth_response(sock, "geheim")
assert response["type"] == "success"
# Step 3: Start session
response = start_session(sock, ["Hyprland"])
assert response["type"] == "success"
sock.close()
finally:
mock.stop()
# Verify what the mock received
assert mock.received[0] == {"type": "create_session", "username": "dominik"}
assert mock.received[1] == {"type": "post_auth_message_response", "response": "geheim"}
assert mock.received[2] == {"type": "start_session", "cmd": ["Hyprland"]}
def test_wrong_password(self, tmp_path: Path) -> None:
"""Simulate a failed login due to wrong password."""
sock_path = tmp_path / "greetd.sock"
mock = MockGreetd(sock_path)
mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
mock.expect({"type": "error", "error_type": "auth_error", "description": "Authentication failed"})
mock.start()
try:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect(str(sock_path))
response = create_session(sock, "dominik")
assert response["type"] == "auth_message"
response = post_auth_response(sock, "falsch")
assert response["type"] == "error"
assert response["description"] == "Authentication failed"
sock.close()
finally:
mock.stop()
def test_cancel_session(self, tmp_path: Path) -> None:
"""Simulate cancelling a session after create."""
sock_path = tmp_path / "greetd.sock"
mock = MockGreetd(sock_path)
mock.expect({"type": "auth_message", "auth_message_type": "secret", "auth_message": "Password:"})
mock.expect({"type": "success"})
mock.start()
try:
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.connect(str(sock_path))
create_session(sock, "dominik")
response = cancel_session(sock)
assert response["type"] == "success"
sock.close()
finally:
mock.stop()
assert mock.received[1] == {"type": "cancel_session"}
class TestLastUser:
"""Tests for saving and loading the last logged-in user."""
def test_save_and_load_last_user(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
cache_path = tmp_path / "cache" / "moongreet" / "last-user"
monkeypatch.setattr("moongreet.greeter.LAST_USER_PATH", cache_path)
from moongreet.greeter import GreeterWindow
GreeterWindow._save_last_user("dominik")
assert cache_path.exists()
assert cache_path.read_text() == "dominik"
result = GreeterWindow._load_last_user()
assert result == "dominik"
def test_load_last_user_missing_file(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
cache_path = tmp_path / "nonexistent" / "last-user"
monkeypatch.setattr("moongreet.greeter.LAST_USER_PATH", cache_path)
from moongreet.greeter import GreeterWindow
result = GreeterWindow._load_last_user()
assert result is None
+203
View File
@@ -0,0 +1,203 @@
# ABOUTME: Tests for greetd IPC protocol — socket communication with length-prefixed JSON.
# ABOUTME: Uses mock sockets to verify message encoding/decoding and greetd request types.
import json
import struct
import socket
from unittest.mock import MagicMock, patch
import pytest
from moongreet.ipc import (
send_message,
recv_message,
create_session,
post_auth_response,
start_session,
cancel_session,
)
class FakeSocket:
"""A fake socket that records sent data and provides canned receive data."""
def __init__(self, recv_data: bytes = b""):
self.sent = bytearray()
self._recv_data = recv_data
self._recv_offset = 0
def sendall(self, data: bytes) -> None:
self.sent.extend(data)
def recv(self, n: int, flags: int = 0) -> bytes:
chunk = self._recv_data[self._recv_offset : self._recv_offset + n]
self._recv_offset += n
return chunk
@classmethod
def with_response(cls, response: dict) -> "FakeSocket":
"""Create a FakeSocket pre-loaded with a length-prefixed JSON response."""
payload = json.dumps(response).encode("utf-8")
data = struct.pack("!I", len(payload)) + payload
return cls(recv_data=data)
class TestSendMessage:
"""Tests for encoding and sending length-prefixed JSON messages."""
def test_sends_length_prefixed_json(self) -> None:
sock = FakeSocket()
msg = {"type": "create_session", "username": "testuser"}
send_message(sock, msg)
payload = json.dumps(msg).encode("utf-8")
expected = struct.pack("!I", len(payload)) + payload
assert bytes(sock.sent) == expected
def test_sends_empty_dict(self) -> None:
sock = FakeSocket()
send_message(sock, {})
payload = json.dumps({}).encode("utf-8")
expected = struct.pack("!I", len(payload)) + payload
assert bytes(sock.sent) == expected
def test_sends_nested_message(self) -> None:
sock = FakeSocket()
msg = {"type": "post_auth_message_response", "response": "secret123"}
send_message(sock, msg)
# Verify the payload is correctly length-prefixed
length_bytes = bytes(sock.sent[:4])
length = struct.unpack("!I", length_bytes)[0]
decoded = json.loads(sock.sent[4:])
assert length == len(json.dumps(msg).encode("utf-8"))
assert decoded == msg
class TestRecvMessage:
"""Tests for receiving and decoding length-prefixed JSON messages."""
def test_receives_valid_message(self) -> None:
response = {"type": "success"}
sock = FakeSocket.with_response(response)
result = recv_message(sock)
assert result == response
def test_receives_complex_message(self) -> None:
response = {
"type": "auth_message",
"auth_message_type": "secret",
"auth_message": "Password:",
}
sock = FakeSocket.with_response(response)
result = recv_message(sock)
assert result == response
def test_raises_on_empty_recv(self) -> None:
sock = FakeSocket(recv_data=b"")
with pytest.raises(ConnectionError):
recv_message(sock)
class TestCreateSession:
"""Tests for the create_session greetd request."""
def test_sends_create_session_request(self) -> None:
response = {
"type": "auth_message",
"auth_message_type": "secret",
"auth_message": "Password:",
}
sock = FakeSocket.with_response(response)
result = create_session(sock, "dominik")
# Verify sent message
length = struct.unpack("!I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {"type": "create_session", "username": "dominik"}
assert result == response
class TestPostAuthResponse:
"""Tests for posting authentication responses (passwords)."""
def test_sends_password_response(self) -> None:
response = {"type": "success"}
sock = FakeSocket.with_response(response)
result = post_auth_response(sock, "mypassword")
length = struct.unpack("!I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {
"type": "post_auth_message_response",
"response": "mypassword",
}
assert result == response
def test_sends_none_response(self) -> None:
"""For auth types that don't require a response."""
response = {"type": "success"}
sock = FakeSocket.with_response(response)
result = post_auth_response(sock, None)
length = struct.unpack("!I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {
"type": "post_auth_message_response",
"response": None,
}
class TestStartSession:
"""Tests for starting a session after authentication."""
def test_sends_start_session_request(self) -> None:
response = {"type": "success"}
sock = FakeSocket.with_response(response)
result = start_session(sock, ["Hyprland"])
length = struct.unpack("!I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {"type": "start_session", "cmd": ["Hyprland"]}
assert result == response
def test_sends_multi_arg_command(self) -> None:
response = {"type": "success"}
sock = FakeSocket.with_response(response)
result = start_session(sock, ["sway", "--config", "/etc/sway/config"])
length = struct.unpack("!I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {
"type": "start_session",
"cmd": ["sway", "--config", "/etc/sway/config"],
}
class TestCancelSession:
"""Tests for cancelling an in-progress session."""
def test_sends_cancel_session_request(self) -> None:
response = {"type": "success"}
sock = FakeSocket.with_response(response)
result = cancel_session(sock)
length = struct.unpack("!I", bytes(sock.sent[:4]))[0]
sent_msg = json.loads(sock.sent[4 : 4 + length])
assert sent_msg == {"type": "cancel_session"}
assert result == response
+32
View File
@@ -0,0 +1,32 @@
# ABOUTME: Tests for power actions — reboot and shutdown via loginctl.
# ABOUTME: Uses mocking to avoid actually calling system commands.
from unittest.mock import patch, call
import pytest
from moongreet.power import reboot, shutdown
class TestReboot:
"""Tests for the reboot power action."""
@patch("moongreet.power.subprocess.run")
def test_calls_loginctl_reboot(self, mock_run) -> None:
reboot()
mock_run.assert_called_once_with(
["loginctl", "reboot"], check=True
)
class TestShutdown:
"""Tests for the shutdown power action."""
@patch("moongreet.power.subprocess.run")
def test_calls_loginctl_poweroff(self, mock_run) -> None:
shutdown()
mock_run.assert_called_once_with(
["loginctl", "poweroff"], check=True
)
+104
View File
@@ -0,0 +1,104 @@
# ABOUTME: Tests for session detection — parsing .desktop files from wayland/xsessions dirs.
# ABOUTME: Uses temporary directories to simulate session file locations.
from pathlib import Path
import pytest
from moongreet.sessions import Session, get_sessions
class TestGetSessions:
"""Tests for discovering available sessions from .desktop files."""
def test_finds_wayland_session(self, tmp_path: Path) -> None:
wayland_dir = tmp_path / "wayland-sessions"
wayland_dir.mkdir()
desktop = wayland_dir / "hyprland.desktop"
desktop.write_text(
"[Desktop Entry]\n"
"Name=Hyprland\n"
"Exec=Hyprland\n"
"Type=Application\n"
)
sessions = get_sessions(wayland_dirs=[wayland_dir], xsession_dirs=[])
assert len(sessions) == 1
assert sessions[0].name == "Hyprland"
assert sessions[0].exec_cmd == "Hyprland"
assert sessions[0].session_type == "wayland"
def test_finds_xsession(self, tmp_path: Path) -> None:
x_dir = tmp_path / "xsessions"
x_dir.mkdir()
desktop = x_dir / "i3.desktop"
desktop.write_text(
"[Desktop Entry]\n"
"Name=i3\n"
"Exec=i3\n"
"Type=Application\n"
)
sessions = get_sessions(wayland_dirs=[], xsession_dirs=[x_dir])
assert len(sessions) == 1
assert sessions[0].session_type == "x11"
def test_finds_sessions_from_multiple_dirs(self, tmp_path: Path) -> None:
wayland_dir = tmp_path / "wayland-sessions"
wayland_dir.mkdir()
(wayland_dir / "sway.desktop").write_text(
"[Desktop Entry]\nName=Sway\nExec=sway\n"
)
x_dir = tmp_path / "xsessions"
x_dir.mkdir()
(x_dir / "openbox.desktop").write_text(
"[Desktop Entry]\nName=Openbox\nExec=openbox-session\n"
)
sessions = get_sessions(wayland_dirs=[wayland_dir], xsession_dirs=[x_dir])
names = {s.name for s in sessions}
assert names == {"Sway", "Openbox"}
def test_returns_empty_for_no_sessions(self, tmp_path: Path) -> None:
empty = tmp_path / "empty"
sessions = get_sessions(wayland_dirs=[empty], xsession_dirs=[empty])
assert sessions == []
def test_skips_files_without_name(self, tmp_path: Path) -> None:
wayland_dir = tmp_path / "wayland-sessions"
wayland_dir.mkdir()
(wayland_dir / "broken.desktop").write_text(
"[Desktop Entry]\nExec=something\n"
)
sessions = get_sessions(wayland_dirs=[wayland_dir], xsession_dirs=[])
assert sessions == []
def test_skips_files_without_exec(self, tmp_path: Path) -> None:
wayland_dir = tmp_path / "wayland-sessions"
wayland_dir.mkdir()
(wayland_dir / "noexec.desktop").write_text(
"[Desktop Entry]\nName=NoExec\n"
)
sessions = get_sessions(wayland_dirs=[wayland_dir], xsession_dirs=[])
assert sessions == []
def test_handles_exec_with_arguments(self, tmp_path: Path) -> None:
wayland_dir = tmp_path / "wayland-sessions"
wayland_dir.mkdir()
(wayland_dir / "sway.desktop").write_text(
"[Desktop Entry]\nName=Sway\nExec=sway --config /etc/sway/config\n"
)
sessions = get_sessions(wayland_dirs=[wayland_dir], xsession_dirs=[])
assert sessions[0].exec_cmd == "sway --config /etc/sway/config"
+134
View File
@@ -0,0 +1,134 @@
# ABOUTME: Tests for user detection — parsing /etc/passwd, avatar lookup, GTK theme reading.
# ABOUTME: Uses temporary files and mocking to avoid system dependencies.
from pathlib import Path
from dataclasses import dataclass
import pytest
from moongreet.users import User, get_users, get_avatar_path, get_user_gtk_theme
class TestGetUsers:
"""Tests for parsing /etc/passwd to find login users."""
def test_returns_users_in_uid_range(self, tmp_path: Path) -> None:
passwd = tmp_path / "passwd"
passwd.write_text(
"root:x:0:0:root:/root:/bin/bash\n"
"nobody:x:65534:65534:Nobody:/:/usr/bin/nologin\n"
"dominik:x:1000:1000:Dominik:/home/dominik:/bin/zsh\n"
"testuser:x:1001:1001:Test User:/home/testuser:/bin/bash\n"
)
users = get_users(passwd_path=passwd)
assert len(users) == 2
assert users[0].username == "dominik"
assert users[0].uid == 1000
assert users[0].gecos == "Dominik"
assert users[0].home == Path("/home/dominik")
assert users[1].username == "testuser"
def test_excludes_nologin_shells(self, tmp_path: Path) -> None:
passwd = tmp_path / "passwd"
passwd.write_text(
"systemuser:x:1000:1000:System:/home/system:/usr/sbin/nologin\n"
"falseuser:x:1001:1001:False:/home/false:/bin/false\n"
"realuser:x:1002:1002:Real:/home/real:/bin/bash\n"
)
users = get_users(passwd_path=passwd)
assert len(users) == 1
assert users[0].username == "realuser"
def test_returns_empty_for_no_matching_users(self, tmp_path: Path) -> None:
passwd = tmp_path / "passwd"
passwd.write_text("root:x:0:0:root:/root:/bin/bash\n")
users = get_users(passwd_path=passwd)
assert users == []
def test_handles_missing_gecos_field(self, tmp_path: Path) -> None:
passwd = tmp_path / "passwd"
passwd.write_text("user:x:1000:1000::/home/user:/bin/bash\n")
users = get_users(passwd_path=passwd)
assert len(users) == 1
assert users[0].gecos == ""
assert users[0].display_name == "user"
class TestGetAvatarPath:
"""Tests for avatar file lookup."""
def test_finds_accountsservice_icon(self, tmp_path: Path) -> None:
icons_dir = tmp_path / "icons"
icons_dir.mkdir()
avatar = icons_dir / "dominik"
avatar.write_bytes(b"PNG")
result = get_avatar_path("dominik", accountsservice_dir=icons_dir)
assert result == avatar
def test_falls_back_to_dot_face(self, tmp_path: Path) -> None:
home = tmp_path / "home" / "dominik"
home.mkdir(parents=True)
face = home / ".face"
face.write_bytes(b"PNG")
empty_icons = tmp_path / "no_icons"
result = get_avatar_path(
"dominik", accountsservice_dir=empty_icons, home_dir=home
)
assert result == face
def test_returns_none_when_no_avatar(self, tmp_path: Path) -> None:
empty_icons = tmp_path / "no_icons"
home = tmp_path / "home" / "nobody"
result = get_avatar_path(
"nobody", accountsservice_dir=empty_icons, home_dir=home
)
assert result is None
class TestGetUserGtkTheme:
"""Tests for reading GTK theme from user's settings.ini."""
def test_reads_theme_from_settings(self, tmp_path: Path) -> None:
gtk_dir = tmp_path / ".config" / "gtk-4.0"
gtk_dir.mkdir(parents=True)
settings = gtk_dir / "settings.ini"
settings.write_text(
"[Settings]\n"
"gtk-theme-name=Adwaita-dark\n"
"gtk-icon-theme-name=Papirus\n"
)
result = get_user_gtk_theme(config_dir=gtk_dir)
assert result == "Adwaita-dark"
def test_returns_none_when_no_settings(self, tmp_path: Path) -> None:
gtk_dir = tmp_path / "nonexistent"
result = get_user_gtk_theme(config_dir=gtk_dir)
assert result is None
def test_returns_none_when_no_theme_key(self, tmp_path: Path) -> None:
gtk_dir = tmp_path / ".config" / "gtk-4.0"
gtk_dir.mkdir(parents=True)
settings = gtk_dir / "settings.ini"
settings.write_text("[Settings]\ngtk-icon-theme-name=Papirus\n")
result = get_user_gtk_theme(config_dir=gtk_dir)
assert result is None