"""Test Matrix authentication: login, token validation, session persistence.""" from __future__ import annotations import json import tempfile from pathlib import Path from unittest import mock import pytest from matrix_hermes.auth import ( AuthResult, WhoAmIResult, _password_login, _validate_token, load_session, login, login_from_config, logout, remove_session, save_session, validate_token, ) from matrix_hermes.config import MatrixConfig class TestAuthResult: """Tests for the AuthResult dataclass.""" def test_success_result(self) -> None: result = AuthResult( success=True, user_id="@test:matrix.org", access_token="syt_abc", device_id="DEV123", homeserver="https://matrix.org", ) assert result.success is True assert result.user_id == "@test:matrix.org" assert result.error == "" def test_failure_result(self) -> None: result = AuthResult(success=False, error="Invalid password") assert result.success is False assert result.error == "Invalid password" assert result.access_token == "" def test_default_values(self) -> None: result = AuthResult(success=True) assert result.user_id == "" assert result.access_token == "" assert result.device_id == "" assert result.homeserver == "" class TestWhoAmIResult: """Tests for the WhoAmIResult dataclass.""" def test_valid(self) -> None: result = WhoAmIResult( valid=True, user_id="@test:matrix.org", device_id="DEV1" ) assert result.valid is True def test_invalid(self) -> None: result = WhoAmIResult(valid=False, error="M_UNKNOWN_TOKEN: Unrecognised") assert result.valid is False class TestPasswordLoginHTTP: """Tests for _password_login — HTTP interactions are mocked.""" def test_successful_login(self) -> None: mock_response = { "user_id": "@bot:matrix.org", "access_token": "syt_test123", "device_id": "HERMES01", "home_server": "matrix.org", } with mock.patch( "matrix_hermes.auth._post_json", return_value=mock_response ) as mock_post: result = _password_login( "https://matrix.org", "@bot:matrix.org", "secret" ) assert result.success is True assert result.user_id == "@bot:matrix.org" assert result.access_token == "syt_test123" assert result.device_id == "HERMES01" mock_post.assert_called_once() def test_login_failure(self) -> None: mock_response = { "errcode": "M_FORBIDDEN", "error": "Invalid password", } with mock.patch( "matrix_hermes.auth._post_json", return_value=mock_response ): result = _password_login( "https://matrix.org", "@bot:matrix.org", "wrong" ) assert result.success is False assert "M_FORBIDDEN" in result.error def test_login_network_error(self) -> None: mock_response = { "errcode": "M_CONNECTION_ERROR", "error": "Connection refused", } with mock.patch( "matrix_hermes.auth._post_json", return_value=mock_response ): result = _password_login( "https://down.example.com", "@bot:down.example.com", "pass" ) assert result.success is False assert "M_CONNECTION_ERROR" in result.error class TestValidateTokenHTTP: """Tests for _validate_token — HTTP interactions are mocked.""" def test_valid_token(self) -> None: mock_response = { "user_id": "@bot:matrix.org", "device_id": "HERMES01", } with mock.patch( "matrix_hermes.auth._get_json", return_value=mock_response ): result = _validate_token("https://matrix.org", "syt_valid") assert result.valid is True assert result.user_id == "@bot:matrix.org" def test_invalid_token(self) -> None: mock_response = { "errcode": "M_UNKNOWN_TOKEN", "error": "Unrecognised access token", } with mock.patch( "matrix_hermes.auth._get_json", return_value=mock_response ): result = _validate_token("https://matrix.org", "syt_bad") assert result.valid is False assert "M_UNKNOWN_TOKEN" in result.error class TestLoginFunction: """Tests for the top-level login() function.""" def test_login_with_valid_token(self) -> None: with mock.patch( "matrix_hermes.auth._validate_token", return_value=WhoAmIResult(valid=True, user_id="@bot:matrix.org"), ): result = login( homeserver="https://matrix.org", access_token="syt_good" ) assert result.success is True assert result.access_token == "syt_good" def test_login_with_invalid_token_no_password(self) -> None: with mock.patch( "matrix_hermes.auth._validate_token", return_value=WhoAmIResult(valid=False, error="Bad token"), ): result = login( homeserver="https://matrix.org", access_token="syt_bad" ) # Token invalid and no password → falls through to error assert result.success is False def test_login_with_password_no_user_id(self) -> None: result = login(homeserver="https://matrix.org", password="secret") assert result.success is False assert "MATRIX_USER_ID" in result.error def test_login_no_credentials(self) -> None: result = login(homeserver="https://matrix.org") assert result.success is False assert "No valid authentication method" in result.error class TestLoginFromConfig: """Tests for login_from_config with session persistence.""" def test_with_access_token_valid(self) -> None: cfg = MatrixConfig( homeserver="https://matrix.org", access_token="syt_valid", ) with mock.patch( "matrix_hermes.auth._validate_token", return_value=WhoAmIResult(valid=True, user_id="@bot:matrix.org"), ): result = login_from_config(cfg) assert result.success is True assert result.access_token == "syt_valid" def test_password_login_saves_session(self, tmp_path: Path) -> None: cfg = MatrixConfig( homeserver="https://matrix.org", user_id="@bot:matrix.org", password="secret", session_dir=str(tmp_path), ) login_mock = { "user_id": "@bot:matrix.org", "access_token": "syt_new", "device_id": "DEV1", } with mock.patch( "matrix_hermes.auth._post_json", return_value=login_mock ): result = login_from_config(cfg) assert result.success is True # Session should have been persisted session_file = tmp_path / "@bot_matrix.org.json" assert session_file.exists() data = json.loads(session_file.read_text()) assert data["access_token"] == "syt_new" def test_loads_session_when_no_token(self, tmp_path: Path) -> None: # Pre-create a session file session_dir = tmp_path / "matrix_sessions" session_dir.mkdir(parents=True) session_data = { "user_id": "@bot:matrix.org", "access_token": "syt_saved", "device_id": "DEV1", "homeserver": "https://matrix.org", } (session_dir / "@bot_matrix.org.json").write_text( json.dumps(session_data) ) cfg = MatrixConfig( homeserver="https://matrix.org", user_id="@bot:matrix.org", session_dir=str(session_dir), ) whoami_mock = WhoAmIResult(valid=True, user_id="@bot:matrix.org") with mock.patch( "matrix_hermes.auth._validate_token", return_value=whoami_mock ): result = login_from_config(cfg) assert result.success is True assert result.access_token == "syt_saved" def test_session_invalid_falls_through_to_password( self, tmp_path: Path ) -> None: # Pre-create a stale session file session_dir = tmp_path / "matrix_sessions" session_dir.mkdir(parents=True) session_data = { "user_id": "@bot:matrix.org", "access_token": "syt_stale", } (session_dir / "@bot_matrix.org.json").write_text( json.dumps(session_data) ) cfg = MatrixConfig( homeserver="https://matrix.org", user_id="@bot:matrix.org", password="secret", session_dir=str(session_dir), ) # Session token is invalid → removed → password login used whoami_stale = WhoAmIResult(valid=False, error="Expired") login_mock = { "user_id": "@bot:matrix.org", "access_token": "syt_fresh", } with mock.patch( "matrix_hermes.auth._validate_token", side_effect=[whoami_stale] ): with mock.patch( "matrix_hermes.auth._post_json", return_value=login_mock ): result = login_from_config(cfg) assert result.success is True assert result.access_token == "syt_fresh" class TestSessionPersistence: """Tests for save_session, load_session, remove_session.""" def test_save_and_load_session(self, tmp_path: Path) -> None: cfg = MatrixConfig( user_id="@test:matrix.org", session_dir=str(tmp_path) ) auth = AuthResult( success=True, user_id="@test:matrix.org", access_token="syt_test", device_id="DEV99", homeserver="https://matrix.org", ) path = save_session(auth, cfg) assert path is not None assert path.exists() loaded = load_session(cfg) assert loaded is not None assert loaded.success is True assert loaded.access_token == "syt_test" assert loaded.device_id == "DEV99" def test_load_nonexistent_session(self) -> None: cfg = MatrixConfig( user_id="@nobody:matrix.org", session_dir="/tmp/nonexistent_dir_xyz", ) result = load_session(cfg) assert result is None def test_remove_session(self, tmp_path: Path) -> None: cfg = MatrixConfig( user_id="@test:matrix.org", session_dir=str(tmp_path) ) auth = AuthResult( success=True, user_id="@test:matrix.org", access_token="syt_test", ) save_session(auth, cfg) # Verify it exists assert load_session(cfg) is not None # Remove assert remove_session(cfg) is True assert load_session(cfg) is None def test_remove_nonexistent_session(self) -> None: cfg = MatrixConfig( user_id="@nobody:matrix.org", session_dir="/tmp/nonexistent_dir", ) assert remove_session(cfg) is False def test_save_failure_result_is_none(self) -> None: cfg = MatrixConfig() auth = AuthResult(success=False, error="No credentials") result = save_session(auth, cfg) assert result is None class TestLogout: """Tests for the logout function.""" def test_logout_success(self) -> None: with mock.patch( "matrix_hermes.auth._post_json", return_value={} ): result = logout("https://matrix.org", "syt_test") assert result is True def test_logout_failure(self) -> None: with mock.patch( "matrix_hermes.auth._post_json", return_value={"errcode": "M_UNKNOWN", "error": "Bad"}, ): result = logout("https://matrix.org", "syt_test") assert result is False