Mnemosyne MCP

"""Tests for token storage functionality.""" import json import time from pathlib import Path from unittest.mock import patch, MagicMock import pytest import jwt from neem.utils.token_storage import ( get_config_path, ensure_config_dir, save_token, load_token, load_config, delete_token, is_token_expired, get_token_info, get_dev_user_id, validate_token_and_load, TokenConfig, TokenStorageError, ) @pytest.fixture def temp_config_dir(tmp_path, monkeypatch): """Fixture to use a temporary directory for config storage.""" config_dir = tmp_path / ".mnemosyne" monkeypatch.setenv("MNEMOSYNE_CONFIG_DIR", str(config_dir)) return config_dir @pytest.fixture def sample_jwt_token(): """Create a sample JWT token with expiration.""" payload = { "sub": "user123", "email": "test@example.com", "exp": int(time.time()) + 3600, # Expires in 1 hour } # Create token without signature (we don't validate it) token = jwt.encode(payload, "secret", algorithm="HS256") return token @pytest.fixture def expired_jwt_token(): """Create an expired JWT token.""" payload = { "sub": "user123", "email": "test@example.com", "exp": int(time.time()) - 3600, # Expired 1 hour ago } token = jwt.encode(payload, "secret", algorithm="HS256") return token class TestConfigPath: """Tests for config path functions.""" def test_get_config_path_default(self, monkeypatch): """Test getting default config path.""" monkeypatch.delenv("MNEMOSYNE_CONFIG_DIR", raising=False) path = get_config_path() assert path == TokenConfig.DEFAULT_CONFIG_DIR / TokenConfig.DEFAULT_CONFIG_FILE def test_get_config_path_custom(self, tmp_path, monkeypatch): """Test getting custom config path from environment.""" custom_dir = tmp_path / "custom" monkeypatch.setenv("MNEMOSYNE_CONFIG_DIR", str(custom_dir)) path = get_config_path() assert path == custom_dir / TokenConfig.DEFAULT_CONFIG_FILE def test_ensure_config_dir_creates_directory(self, temp_config_dir): """Test that ensure_config_dir creates the directory.""" assert not temp_config_dir.exists() result = ensure_config_dir() assert result.exists() assert result.is_dir() assert result == temp_config_dir def test_ensure_config_dir_idempotent(self, temp_config_dir): """Test that ensure_config_dir can be called multiple times.""" ensure_config_dir() ensure_config_dir() assert temp_config_dir.exists() def test_ensure_config_dir_permissions(self, temp_config_dir): """Test that config directory has secure permissions.""" result = ensure_config_dir() # Check that directory is readable and writable by owner assert result.stat().st_mode & 0o700 class TestTokenSave: """Tests for saving tokens.""" def test_save_token_basic(self, temp_config_dir, sample_jwt_token): """Test saving a token without user info.""" path = save_token(sample_jwt_token) assert path.exists() assert path == get_config_path() # Verify file contents data = json.loads(path.read_text()) assert data["token"] == sample_jwt_token assert data["version"] == "1.0" assert "user_info" not in data def test_save_token_with_user_info(self, temp_config_dir, sample_jwt_token): """Test saving a token with user info.""" user_info = {"email": "test@example.com", "name": "Test User"} path = save_token(sample_jwt_token, user_info) data = json.loads(path.read_text()) assert data["token"] == sample_jwt_token assert data["user_info"] == user_info def test_save_token_overwrites_existing(self, temp_config_dir, sample_jwt_token): """Test that saving a token overwrites existing one.""" # Save first token save_token("token1") # Save second token path = save_token(sample_jwt_token) # Verify only second token exists data = json.loads(path.read_text()) assert data["token"] == sample_jwt_token def test_save_token_secure_permissions(self, temp_config_dir, sample_jwt_token): """Test that saved token has secure file permissions.""" path = save_token(sample_jwt_token) # Check file is only readable/writable by owner (0o600) mode = path.stat().st_mode & 0o777 assert mode == 0o600 class TestTokenLoad: """Tests for loading tokens.""" def test_load_token_success(self, temp_config_dir, sample_jwt_token): """Test loading a saved token.""" save_token(sample_jwt_token) loaded_token = load_token() assert loaded_token == sample_jwt_token def test_load_token_not_found(self, temp_config_dir): """Test loading when no token exists.""" token = load_token() assert token is None def test_load_token_invalid_json(self, temp_config_dir): """Test loading when config file has invalid JSON.""" config_path = get_config_path() ensure_config_dir() config_path.write_text("invalid json {") token = load_token() assert token is None def test_load_token_missing_token_field(self, temp_config_dir): """Test loading when config file exists but has no token field.""" config_path = get_config_path() ensure_config_dir() config_path.write_text(json.dumps({"version": "1.0"})) token = load_token() assert token is None def test_load_config_success(self, temp_config_dir, sample_jwt_token): """Test loading complete config.""" user_info = {"email": "test@example.com"} save_token(sample_jwt_token, user_info) config = load_config() assert config is not None assert config["token"] == sample_jwt_token assert config["user_info"] == user_info assert config["version"] == "1.0" def test_load_config_not_found(self, temp_config_dir): """Test loading config when file doesn't exist.""" config = load_config() assert config is None class TestTokenDelete: """Tests for deleting tokens.""" def test_delete_token_success(self, temp_config_dir, sample_jwt_token): """Test deleting an existing token.""" save_token(sample_jwt_token) assert get_config_path().exists() result = delete_token() assert result is True assert not get_config_path().exists() def test_delete_token_not_exists(self, temp_config_dir): """Test deleting when no token exists.""" result = delete_token() assert result is False class TestTokenValidation: """Tests for token validation functions.""" def test_is_token_expired_valid(self, sample_jwt_token): """Test checking expiration of a valid token.""" assert is_token_expired(sample_jwt_token) is False def test_is_token_expired_expired(self, expired_jwt_token): """Test checking expiration of an expired token.""" assert is_token_expired(expired_jwt_token) is True def test_is_token_expired_invalid_token(self): """Test checking expiration of an invalid token.""" assert is_token_expired("not.a.valid.token") is True def test_is_token_expired_no_exp_claim(self): """Test checking token without exp claim.""" payload = {"sub": "user123"} token = jwt.encode(payload, "secret", algorithm="HS256") assert is_token_expired(token) is True def test_get_token_info_success(self, sample_jwt_token): """Test extracting info from valid token.""" info = get_token_info(sample_jwt_token) assert info is not None assert info["sub"] == "user123" assert info["email"] == "test@example.com" assert "exp" in info def test_get_token_info_invalid(self): """Test extracting info from invalid token.""" info = get_token_info("invalid.token") assert info is None class TestValidateTokenAndLoad: """Tests for combined validation and loading.""" def test_validate_token_and_load_success(self, temp_config_dir, sample_jwt_token): """Test validating and loading a valid token.""" save_token(sample_jwt_token) token = validate_token_and_load() assert token == sample_jwt_token def test_validate_token_and_load_no_token(self, temp_config_dir): """Test validating when no token exists.""" token = validate_token_and_load() assert token is None def test_validate_token_and_load_expired(self, temp_config_dir, expired_jwt_token, capsys): """Test validating an expired token.""" save_token(expired_jwt_token) token = validate_token_and_load() assert token is None # Check that user-friendly message was printed captured = capsys.readouterr() combined = f"{captured.out} {captured.err}".lower() assert "expired" in combined def test_validate_token_and_load_dev_override(self, temp_config_dir, monkeypatch): """Allow bypassing auth flow via MNEMOSYNE_DEV_TOKEN.""" monkeypatch.setenv("MNEMOSYNE_DEV_TOKEN", "dev-token") token = validate_token_and_load() assert token == "dev-token" class TestDevModeHelpers: """Tests for helper functions that support dev mode.""" def test_get_dev_user_id_explicit_env(self, monkeypatch): monkeypatch.setenv("MNEMOSYNE_DEV_USER_ID", "carol") monkeypatch.delenv("MNEMOSYNE_DEV_TOKEN", raising=False) assert get_dev_user_id() == "carol" def test_get_dev_user_id_falls_back_to_dev_token(self, monkeypatch): monkeypatch.delenv("MNEMOSYNE_DEV_USER_ID", raising=False) monkeypatch.setenv("MNEMOSYNE_DEV_TOKEN", "alice") assert get_dev_user_id() == "alice" def test_get_dev_user_id_none(self, monkeypatch): monkeypatch.delenv("MNEMOSYNE_DEV_USER_ID", raising=False) monkeypatch.delenv("MNEMOSYNE_DEV_TOKEN", raising=False) assert get_dev_user_id() is None class TestEdgeCases: """Tests for edge cases and error handling.""" def test_save_token_creates_parent_directories(self, tmp_path, monkeypatch): """Test that save_token creates parent directories if they don't exist.""" nested_dir = tmp_path / "a" / "b" / "c" monkeypatch.setenv("MNEMOSYNE_CONFIG_DIR", str(nested_dir)) token = "test.token.here" path = save_token(token) assert path.exists() assert path.parent == nested_dir def test_config_path_with_spaces(self, tmp_path, monkeypatch): """Test config path with spaces in directory name.""" config_dir = tmp_path / "my config dir" monkeypatch.setenv("MNEMOSYNE_CONFIG_DIR", str(config_dir)) path = save_token("test.token") assert path.exists() assert "my config dir" in str(path)