feat: matrix-hermes project structure and Matrix authentication

tests/conftest.py

@@ -0,0 +1,42 @@
+"""Test configuration for matrix-hermes."""
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# Add src to path for imports
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
+
+
+@pytest.fixture
+def clean_env(monkeypatch):
+    """Remove Matrix env vars for clean test state."""
+    for key in list(os.environ):
+        if key.startswith("MATRIX_"):
+            monkeypatch.delenv(key, raising=False)
+
+
+@pytest.fixture
+def mock_env(monkeypatch):
+    """Set up a minimal valid Matrix environment."""
+    monkeypatch.setenv("MATRIX_HOMESERVER", "https://matrix.example.org")
+    monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_test_token_abc123")
+    monkeypatch.setenv("MATRIX_USER_ID", "@bot:example.org")
+
+
+@pytest.fixture
+def mock_urlopen():
+    """Mock urllib.request.urlopen for HTTP testing."""
+    with patch("urllib.request.urlopen") as mock:
+        yield mock
+
+
+@pytest.fixture
+def mock_request():
+    """Mock urllib.request.Request for HTTP testing."""
+    with patch("urllib.request.Request") as mock:
+        yield mock

tests/test_auth.py

@@ -0,0 +1,377 @@
+"""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

tests/test_cli.py

@@ -0,0 +1,205 @@
+"""Tests for matrix_hermes.cli module."""
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from matrix_hermes.cli import (
+    _mask_token,
+    build_parser,
+    cmd_check,
+    cmd_login,
+    cmd_logout,
+    cmd_show_config,
+    cmd_whoami,
+    main,
+)
+
+
+class TestMaskToken:
+    """Tests for _mask_token helper."""
+
+    def test_short_token(self):
+        assert _mask_token("abc") == "***"
+
+    def test_long_token(self):
+        result = _mask_token("syt_abcdefghijklmnopqrstuvwxyz")
+        assert result.startswith("syt_ab")
+        assert result.endswith("uvwxyz")
+        assert "..." in result
+
+    def test_exact_length(self):
+        """12-char token with visible=6: len<=12 means fully masked."""
+        result = _mask_token("123456789012")
+        assert result == "************"
+
+    def test_just_over_threshold(self):
+        """13-char token with visible=6: len>12 means partial mask."""
+        result = _mask_token("1234567890123")
+        assert result == "123456...890123"
+
+
+class TestParser:
+    """Tests for CLI argument parser."""
+
+    def test_build_parser(self):
+        parser = build_parser()
+        assert "login" in parser._subparsers._group_actions[0].choices
+        assert "whoami" in parser._subparsers._group_actions[0].choices
+        assert "check" in parser._subparsers._group_actions[0].choices
+        assert "show-config" in parser._subparsers._group_actions[0].choices
+
+    def test_no_args_shows_help(self, capsys):
+        """Running with no args shows help."""
+        result = main([])
+        # Should not crash
+        captured = capsys.readouterr()
+        assert "usage" in captured.out.lower() or result == 0
+
+
+class TestCmdCheck:
+    """Tests for 'check' command."""
+
+    def test_not_configured(self, clean_env, capsys):
+        """When no Matrix vars are set, check should fail."""
+        args = build_parser().parse_args(["check"])
+        result = cmd_check(args)
+        assert result == 1
+        captured = capsys.readouterr()
+        assert "not configured" in captured.out.lower()
+
+    def test_configured(self, mock_env, capsys):
+        """When Matrix vars are set, check should succeed."""
+        args = build_parser().parse_args(["check"])
+        result = cmd_check(args)
+        assert result == 0
+        captured = capsys.readouterr()
+        assert "valid" in captured.out.lower()
+
+
+class TestCmdShowConfig:
+    """Tests for 'show-config' command."""
+
+    def test_output_format(self, mock_env, capsys):
+        """show-config outputs env var format."""
+        args = build_parser().parse_args(["show-config"])
+        result = cmd_show_config(args)
+        assert result == 0
+        captured = capsys.readouterr()
+        assert "MATRIX_HOMESERVER=" in captured.out
+        assert "MATRIX_ACCESS_TOKEN=" in captured.out
+
+    def test_redacted(self, mock_env, capsys):
+        """show-config --redact masks sensitive values."""
+        args = build_parser().parse_args(["show-config", "--redact"])
+        result = cmd_show_config(args)
+        assert result == 0
+        captured = capsys.readouterr()
+        # Token should be masked
+        assert "syt_test_token" not in captured.out
+        assert "syt_test" in captured.out or "..." in captured.out
+
+
+class TestCmdLogin:
+    """Tests for 'login' command."""
+
+    def test_no_homeserver(self, clean_env, capsys):
+        """Login fails without homeserver."""
+        args = build_parser().parse_args(["login"])
+        result = cmd_login(args)
+        assert result == 1
+
+    def test_with_valid_token(self, mock_env, capsys):
+        """Login with valid token prints success."""
+        with patch("urllib.request.urlopen") as mock_urlopen:
+            mock_resp = MagicMock()
+            mock_resp.read.return_value = json.dumps(
+                {"user_id": "@bot:example.org", "device_id": "DEV1"}
+            ).encode()
+            mock_resp.__enter__.return_value = mock_resp
+            mock_urlopen.return_value = mock_resp
+
+            args = build_parser().parse_args(["login"])
+            result = cmd_login(args)
+            assert result == 0
+
+    def test_with_bad_credentials(self, mock_env, capsys):
+        """Login with bad credentials prints error."""
+        with patch("urllib.request.urlopen") as mock_urlopen:
+            mock_resp = MagicMock()
+            mock_resp.read.return_value = json.dumps(
+                {"errcode": "M_FORBIDDEN", "error": "Bad password"}
+            ).encode()
+            mock_resp.__enter__.return_value = mock_resp
+            mock_urlopen.return_value = mock_resp
+
+            # Override with password-based login (no valid token)
+            import os
+
+            os.environ["MATRIX_ACCESS_TOKEN"] = ""
+            os.environ["MATRIX_PASSWORD"] = "wrong"
+
+            args = build_parser().parse_args(["login"])
+            result = cmd_login(args)
+            assert result == 1
+
+
+class TestCmdWhoami:
+    """Tests for 'whoami' command."""
+
+    def test_no_homeserver(self, clean_env, capsys):
+        """whoami fails without homeserver."""
+        args = build_parser().parse_args(["whoami"])
+        result = cmd_whoami(args)
+        assert result == 1
+
+    def test_valid(self, mock_env, capsys):
+        """whoami with valid token."""
+        with patch("urllib.request.urlopen") as mock_urlopen:
+            mock_resp = MagicMock()
+            mock_resp.read.return_value = json.dumps(
+                {"user_id": "@bot:example.org"}
+            ).encode()
+            mock_resp.__enter__.return_value = mock_resp
+            mock_urlopen.return_value = mock_resp
+
+            args = build_parser().parse_args(["whoami"])
+            result = cmd_whoami(args)
+            assert result == 0
+
+    def test_invalid(self, mock_env, capsys):
+        """whoami with invalid token."""
+        with patch("urllib.request.urlopen") as mock_urlopen:
+            mock_resp = MagicMock()
+            mock_resp.read.return_value = json.dumps(
+                {"errcode": "M_UNKNOWN_TOKEN", "error": "Unrecognised"}
+            ).encode()
+            mock_resp.__enter__.return_value = mock_resp
+            mock_urlopen.return_value = mock_resp
+
+            args = build_parser().parse_args(["whoami"])
+            result = cmd_whoami(args)
+            assert result == 1
+
+
+class TestCmdLogout:
+    """Tests for 'logout' command."""
+
+    def test_no_homeserver(self, clean_env, capsys):
+        """logout fails without homeserver."""
+        args = build_parser().parse_args(["logout"])
+        result = cmd_logout(args)
+        assert result == 1
+
+    def test_success(self, mock_env, capsys):
+        """logout with valid token."""
+        with patch("urllib.request.urlopen") as mock_urlopen:
+            mock_resp = MagicMock()
+            mock_resp.read.return_value = json.dumps({}).encode()
+            mock_resp.__enter__.return_value = mock_resp
+            mock_urlopen.return_value = mock_resp
+
+            args = build_parser().parse_args(["logout"])
+            result = cmd_logout(args)
+            assert result == 0

tests/test_config.py

@@ -0,0 +1,107 @@
+"""Test Matrix configuration loading from env vars and files."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from matrix_hermes.config import MatrixConfig, _parse_csv
+
+
+class TestParseCSV:
+    """Tests for the CSV parsing helper."""
+
+    def test_empty_string(self) -> None:
+        assert _parse_csv("") == []
+
+    def test_single_value(self) -> None:
+        assert _parse_csv("alice") == ["alice"]
+
+    def test_multiple_values(self) -> None:
+        assert _parse_csv("alice, bob , charlie") == ["alice", "bob", "charlie"]
+
+    def test_whitespace_only(self) -> None:
+        assert _parse_csv("  ,  , ") == []
+
+
+class TestMatrixConfigDefaults:
+    """Tests for default config values."""
+
+    def test_empty_config(self) -> None:
+        cfg = MatrixConfig()
+        assert cfg.homeserver == ""
+        assert cfg.access_token == ""
+        assert cfg.password == ""
+
+    def test_is_configured_false(self) -> None:
+        cfg = MatrixConfig()
+        assert cfg.is_configured() is False
+
+    def test_is_configured_with_homeserver_and_token(self) -> None:
+        cfg = MatrixConfig(homeserver="https://matrix.org", access_token="fake-token")
+        assert cfg.is_configured() is True
+
+    def test_is_configured_with_homeserver_and_password(self) -> None:
+        cfg = MatrixConfig(
+            homeserver="https://matrix.org",
+            user_id="@test:matrix.org",
+            password="secret",
+        )
+        assert cfg.is_configured() is True
+
+    def test_validate_empty_config(self) -> None:
+        cfg = MatrixConfig()
+        errors = cfg.validate()
+        assert len(errors) > 0
+        assert any("MATRIX_HOMESERVER" in e for e in errors)
+
+    def test_validate_missing_auth(self) -> None:
+        cfg = MatrixConfig(homeserver="https://matrix.org")
+        errors = cfg.validate()
+        assert any("ACCESS_TOKEN" in e or "PASSWORD" in e for e in errors)
+
+    def test_resolved_session_dir_default(self) -> None:
+        cfg = MatrixConfig()
+        assert cfg.resolved_session_dir == str(
+            Path.home() / ".hermes" / "matrix_sessions"
+        )
+
+    def test_resolved_session_dir_custom(self) -> None:
+        cfg = MatrixConfig(session_dir="/tmp/matrix_test")
+        assert cfg.resolved_session_dir == "/tmp/matrix_test"
+
+
+class TestFromEnv:
+    """Tests for env var loading."""
+
+    def test_loads_from_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("MATRIX_HOMESERVER", "https://example.com")
+        monkeypatch.setenv("MATRIX_ACCESS_TOKEN", "syt_abc123")
+        monkeypatch.setenv("MATRIX_USER_ID", "@bot:example.com")
+        monkeypatch.setenv("MATRIX_REQUIRE_MENTION", "false")
+
+        cfg = MatrixConfig.from_env()
+        assert cfg.homeserver == "https://example.com"
+        assert cfg.access_token == "syt_abc123"
+        assert cfg.user_id == "@bot:example.com"
+        assert cfg.require_mention is False
+
+    def test_encryption_flag(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("MATRIX_ENCRYPTION", "true")
+        cfg = MatrixConfig.from_env()
+        assert cfg.encryption is True
+
+    def test_allowed_users(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("MATRIX_ALLOWED_USERS", "@a:example.com, @b:example.com")
+        cfg = MatrixConfig.from_env()
+        assert cfg.allowed_users == ["@a:example.com", "@b:example.com"]
+
+    def test_auto_thread_default(self) -> None:
+        cfg = MatrixConfig.from_env()
+        assert cfg.auto_thread is True
+
+    def test_require_mention_default(self) -> None:
+        cfg = MatrixConfig.from_env()
+        assert cfg.require_mention is True