greetd-moongreet/tests/test_integration.py

# 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.greeter import faillock_warning, FAILLOCK_MAX_ATTEMPTS
from moongreet.i18n import load_strings
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_multi_stage_auth_sends_cancel(self, tmp_path: Path) -> None:
        """When greetd sends a second auth_message after password, cancel the session."""
        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": "auth_message", "auth_message_type": "secret", "auth_message": "TOTP:"})
        mock.expect({"type": "success"})  # Response to cancel_session
        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 — greetd responds with another auth_message
            response = post_auth_response(sock, "geheim")
            assert response["type"] == "auth_message"

            # Step 3: Cancel because multi-stage auth is not supported
            response = cancel_session(sock)
            assert response["type"] == "success"

            sock.close()
        finally:
            mock.stop()

        # Verify cancel was sent
        assert mock.received[2] == {"type": "cancel_session"}

    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 TestFaillockWarning:
    """Tests for the faillock warning message logic."""

    def test_no_warning_on_first_attempt(self) -> None:
        strings = load_strings("de")
        assert faillock_warning(1, strings) is None

    def test_warning_on_second_attempt(self) -> None:
        strings = load_strings("de")
        warning = faillock_warning(2, strings)
        assert warning is not None
        assert "1" in warning  # 1 Versuch übrig

    def test_warning_on_third_attempt(self) -> None:
        strings = load_strings("de")
        warning = faillock_warning(3, strings)
        assert warning is not None
        assert warning == strings.faillock_locked

    def test_warning_beyond_max_attempts(self) -> None:
        strings = load_strings("de")
        warning = faillock_warning(4, strings)
        assert warning is not None
        assert warning == strings.faillock_locked

    def test_max_attempts_constant_is_three(self) -> None:
        assert FAILLOCK_MAX_ATTEMPTS == 3


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