378 lines
12 KiB
Python
378 lines
12 KiB
Python
"""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
|