slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(system): security: DPLAN-0155 — telegram dead code removed, api secrets-only cleanup

Browse Source 
Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-04-26 17:59:25 -07:00
co-authored by @devpulse
parent ec406f4aa4
commit 1c063bf236
10 changed files with 91 additions and 456 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/ai_mail/apps/handlers/dispatch/daemon.py
+1 -34
View File
@@ -28,8 +28,6 @@ import subprocess
from pathlib import Path
from datetime import datetime, date
from typing import Dict, Any, Optional
from urllib.request import Request, urlopen
from urllib.error import URLError
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.ai_mail.apps.handlers.json import json_handler
@@ -48,9 +46,6 @@ DAEMON_LOG_FILE = _AI_MAIL_DIR / ".ai_mail.local" / "dispatch_daemon.log"
DAEMON_PID_FILE = _AI_MAIL_DIR / ".ai_mail.local" / "daemon.pid"
BRANCH_REGISTRY = _REPO_ROOT / "AIPASS_REGISTRY.json"
# Telegram notifications (scheduler bot)
SCHEDULER_CONFIG = _REPO_ROOT / ".aipass" / "scheduler_config.json"
# Graceful shutdown
SHUTDOWN = False
@@ -58,30 +53,6 @@ SHUTDOWN = False
from aipass.ai_mail.apps.handlers.dispatch.test_token import scan_and_ack_test_emails
def _notify_telegram(message: str) -> bool:
"""Send a notification to Patrick's Telegram via the scheduler bot."""
try:
with open(SCHEDULER_CONFIG, "r", encoding="utf-8") as f:
config = json.load(f)
bot_token = config["telegram_bot_token"]
chat_id = config["telegram_chat_id"]
except (FileNotFoundError, KeyError, json.JSONDecodeError):
logger.info("Telegram notification skipped (no scheduler config)")
return False
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
payload = json.dumps({"chat_id": chat_id, "text": message}).encode("utf-8")
req = Request(url, data=payload, headers={"Content-Type": "application/json"})
try:
with urlopen(req, timeout=10) as resp:
result = json.loads(resp.read())
return result.get("ok", False)
except (URLError, Exception):
logger.info("Telegram notification failed: %s", message[:60])
return False
def _handle_signal(signum, _frame):
"""Handle shutdown signals for graceful daemon stop."""
global SHUTDOWN
@@ -431,7 +402,6 @@ def spawn_agent(
logger.info(f'SPAWN {branch_email} PID={monitor_pid} (monitor) sender={sender} subject="{subject[:60]}"')
log_dispatch(branch_email, monitor_pid, "spawned")
_notify_telegram(f"[Dispatch] {branch_email} woke\nTask from {sender}: {subject[:80]}")
return True
except Exception as e:
@@ -440,7 +410,6 @@ def spawn_agent(
lock_file.unlink(missing_ok=True)
logger.info(f"SPAWN FAILED {branch_email}: {e}")
log_dispatch(branch_email, None, "failed", error_msg=str(e))
_notify_telegram(f"[Dispatch FAILED] {branch_email}\n{type(e).__name__}: {e}")
return False
@@ -465,7 +434,7 @@ def _read_session_type(pid_str: str) -> str:
# Session types that should NOT block dispatch (idle/background sessions)
_NON_BLOCKING_SESSION_TYPES = {"telegram", "dispatched", "daemon"}
_NON_BLOCKING_SESSION_TYPES = {"dispatched", "daemon"}
def _is_branch_occupied(branch_path: Path) -> bool:
@@ -574,7 +543,6 @@ def run_daemon() -> None:
logger.info("=" * 60)
logger.info(f"DISPATCH DAEMON STARTING (PID {os.getpid()})")
logger.info("=" * 60)
_notify_telegram(f"[Daemon] Started (PID {os.getpid()})")
config = load_config()
poll_interval = config.get("poll_interval_seconds", 300)
@@ -630,7 +598,6 @@ def run_daemon() -> None:
_remove_pid_file()
logger.info("DISPATCH DAEMON STOPPED")
_notify_telegram("[Daemon] Stopped")
if __name__ == "__main__":
src/aipass/ai_mail/apps/handlers/dispatch/wake.py
+1 -1
View File
@@ -226,7 +226,7 @@ def _read_session_type(pid_str: str) -> str:
# Session types that should NOT block dispatch (idle/background sessions)
_NON_BLOCKING_SESSION_TYPES = {"telegram", "dispatched", "daemon"}
_NON_BLOCKING_SESSION_TYPES = {"dispatched", "daemon"}
def _is_branch_occupied(branch_path: Path) -> bool:
src/aipass/ai_mail/tests/test_daemon.py
-95
View File
@@ -770,7 +770,6 @@ import sys
from unittest.mock import MagicMock, mock_open
from aipass.ai_mail.apps.handlers.dispatch.daemon import (
_notify_telegram,
_handle_signal,
_check_lock,
_acquire_lock,
@@ -784,80 +783,6 @@ from aipass.ai_mail.apps.handlers.dispatch.daemon import (
)
# ---- _notify_telegram tests ------------------------------------
def test_notify_telegram_success(tmp_path, monkeypatch):
"""Successful Telegram notification returns True."""
config_file = tmp_path / "scheduler_config.json"
config_file.write_text(
json.dumps({"telegram_bot_token": "fake-token", "telegram_chat_id": "12345"}),
encoding="utf-8",
)
monkeypatch.setattr(daemon_mod, "SCHEDULER_CONFIG", config_file)
mock_resp = MagicMock()
mock_resp.read.return_value = json.dumps({"ok": True}).encode("utf-8")
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
mock_resp.__exit__ = MagicMock(return_value=False)
with patch("aipass.ai_mail.apps.handlers.dispatch.daemon.urlopen", return_value=mock_resp):
result = _notify_telegram("Test message")
assert result is True
def test_notify_telegram_config_missing(tmp_path, monkeypatch):
"""Missing scheduler config returns False."""
monkeypatch.setattr(daemon_mod, "SCHEDULER_CONFIG", tmp_path / "nonexistent.json")
result = _notify_telegram("Test message")
assert result is False
def test_notify_telegram_config_decode_error(tmp_path, monkeypatch):
"""Corrupt scheduler config returns False."""
config_file = tmp_path / "scheduler_config.json"
config_file.write_text("{bad json!", encoding="utf-8")
monkeypatch.setattr(daemon_mod, "SCHEDULER_CONFIG", config_file)
result = _notify_telegram("Test message")
assert result is False
def test_notify_telegram_config_missing_key(tmp_path, monkeypatch):
"""Config missing required keys returns False."""
config_file = tmp_path / "scheduler_config.json"
config_file.write_text(json.dumps({"telegram_bot_token": "tok"}), encoding="utf-8")
monkeypatch.setattr(daemon_mod, "SCHEDULER_CONFIG", config_file)
result = _notify_telegram("Test message")
assert result is False
def test_notify_telegram_url_error(tmp_path, monkeypatch):
"""URLError during sending returns False."""
from urllib.error import URLError
config_file = tmp_path / "scheduler_config.json"
config_file.write_text(
json.dumps({"telegram_bot_token": "fake-token", "telegram_chat_id": "12345"}),
encoding="utf-8",
)
monkeypatch.setattr(daemon_mod, "SCHEDULER_CONFIG", config_file)
with patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.urlopen",
side_effect=URLError("connection refused"),
):
result = _notify_telegram("Test message")
assert result is False
# ---- _handle_signal tests --------------------------------------
@@ -1329,10 +1254,6 @@ def test_spawn_agent_success(tmp_path):
return_value=(True, "Lock acquired"),
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
return_value=True,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.send_notification",
create=True,
@@ -1361,10 +1282,6 @@ def test_spawn_agent_exception(tmp_path):
side_effect=OSError("command not found"),
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
return_value=True,
),
):
result = spawn_agent(branch_path, "@testbranch", message, config, state)
@@ -1403,10 +1320,6 @@ def test_spawn_agent_strips_claude_env_vars(tmp_path, monkeypatch):
return_value=(True, "Lock acquired"),
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
return_value=True,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.send_notification",
create=True,
@@ -1446,10 +1359,6 @@ def test_run_daemon_kill_switch_pauses(tmp_path, monkeypatch):
return_value=True,
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon._remove_pid_file"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
return_value=True,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.load_config",
return_value={
@@ -1487,10 +1396,6 @@ def test_run_daemon_shutdown_exits_loop(tmp_path, monkeypatch):
return_value=True,
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon._remove_pid_file"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
return_value=True,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.load_config",
return_value={
src/aipass/ai_mail/tests/test_misc_handlers.py
-1
View File
@@ -163,7 +163,6 @@ def test_daemon_poll_cycle_is_called(tmp_path, monkeypatch):
with (
patch.object(daemon_mod, "_write_pid_file", return_value=True),
patch.object(daemon_mod, "_remove_pid_file"),
patch.object(daemon_mod, "_notify_telegram", return_value=False),
patch.object(daemon_mod, "poll_cycle", side_effect=mock_poll_cycle),
patch.object(daemon_mod, "save_daemon_state"),
patch.object(daemon_mod, "is_kill_switch_active", return_value=False),
src/aipass/api/.seedgo/bypass.json
+11 -1
View File
@@ -33,7 +33,7 @@
{
"file": "apps/handlers/auth/keys.py",
"standard": "deep_nesting",
"reason": "2 functions: _read_key_from_secrets() depth 5 (file read with validation guards), get_key_from_config() depth 4 (config navigation with nested dict structure)"
"reason": "_read_key_from_secrets() depth 5 — file read with validation guards"
},
{
"file": "apps/handlers/openrouter/client.py",
@@ -90,6 +90,16 @@
"standard": "architecture",
"reason": "Test file — lives in tests/ by convention, not in the 3-layer app structure. Test files are exempt from layer architecture standard."
},
{
"file": "tests/test_api_key.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by convention, not in the 3-layer app structure. Test files are exempt from layer architecture standard."
},
{
"file": "tests/test_critical_paths.py",
"standard": "architecture",
"reason": "Test file — lives in tests/ by convention, not in the 3-layer app structure. Test files are exempt from layer architecture standard."
},
{
"file": "apps/integrations/broken_driver/driver.py",
"standard": "architecture",
src/aipass/api/apps/handlers/auth/keys.py
+10 -81
View File
@@ -10,12 +10,11 @@
API Key Management Handler
Handles API key retrieval and validation for multiple providers.
Keys are read from config JSON or directly from ~/.secrets/aipass/.env.
Keys are read from ~/.secrets/aipass/.env (single source of truth).
Functions:
get_api_key() - Get validated API key
validate_key() - Validate key format for provider
get_key_from_config() - Retrieve key from config JSON
get_validation_rules() - Get provider-specific validation rules
"""
@@ -34,10 +33,6 @@ from aipass.api.apps.handlers.json import json_handler
# CONSTANTS
# ==============================================
# Navigate: keys.py -> auth/ -> handlers/ -> apps/ -> api/
API_ROOT = Path(__file__).resolve().parent.parent.parent.parent
API_JSON_DIR = API_ROOT / "api_json"
# Provider validation rules (embedded - no config dependency for core validation)
VALIDATION_RULES = {
"openrouter": {"prefix": "sk-or-v1-", "min_length": 40},
@@ -55,11 +50,9 @@ VALIDATION_RULES = {
def get_api_key(provider: str = "openrouter") -> Optional[str]:
"""
Get validated API key for provider.
Get validated API key for provider from secrets file.
Sources (in order):
1. Config JSON file (api_json/api_connect_config.json)
2. Secrets file (~/.secrets/aipass/.env)
Source: ~/.secrets/aipass/.env (single source of truth)
Args:
provider: Provider name (default: 'openrouter')
@@ -73,24 +66,11 @@ def get_api_key(provider: str = "openrouter") -> Optional[str]:
... print(f"Got key: {key[:20]}...")
"""
try:
source = ""
# 1. Try config file
key = get_key_from_config(provider)
key = _read_key_from_secrets(provider)
if key and validate_key(key, provider):
source = "config"
# 2. Try secrets file
if not source:
key = _read_key_from_secrets(provider)
if key and validate_key(key, provider):
source = "secrets"
if source:
json_handler.log_operation("key_retrieved", {"provider": provider, "source": source})
json_handler.log_operation("key_retrieved", {"provider": provider, "source": "secrets"})
return key
# No valid key found
return None
except Exception as e:
@@ -129,50 +109,6 @@ def _read_key_from_secrets(provider: str) -> Optional[str]:
return None
def get_key_from_config(provider: str) -> Optional[str]:
"""
Retrieve API key from config JSON file.
Reads from: <api_root>/api_json/api_connect_config.json
Args:
provider: Provider name (e.g., 'openrouter')
Returns:
str: API key from config or None if not found
Example:
>>> key = get_key_from_config('openrouter')
"""
try:
config_path = API_JSON_DIR / "api_connect_config.json"
if not config_path.exists():
# Config file not found
return None
import json
with open(config_path, "r", encoding="utf-8") as f:
config = json.load(f)
# Navigate config structure
if "config" in config:
providers = config["config"].get("providers", {})
if provider in providers:
key = providers[provider].get("api_key", "")
if key:
return key
# No key in config file
return None
except Exception as e:
# Error reading config
logger.error(f"Error reading config for provider '{provider}': {e}")
return None
# ==============================================
# KEY VALIDATION
# ==============================================
@@ -255,7 +191,7 @@ def diagnose_key(provider: str = "openrouter") -> str:
"""
Diagnose why get_api_key() returned None.
Checks all sources for a raw key (skipping validation) and explains
Checks secrets file for a raw key (skipping validation) and explains
exactly why it failed — missing entirely, wrong prefix, too short, etc.
Args:
@@ -268,27 +204,20 @@ def diagnose_key(provider: str = "openrouter") -> str:
>>> if not get_api_key('openrouter'):
... print(diagnose_key('openrouter'))
"""
# Check all sources for raw key (without validation)
key = get_key_from_config(provider)
source = "config"
if not key:
key = _read_key_from_secrets(provider)
source = "secrets"
key = _read_key_from_secrets(provider)
if not key:
secrets_path = Path.home() / ".secrets" / "aipass" / ".env"
return f"API key for {provider} not found. Expected at {secrets_path}. Run drone @api setup to configure."
# Key exists but failed validation — explain why
key = key.strip()
rules = get_validation_rules(provider)
if "prefix" in rules and not key.startswith(rules["prefix"]):
actual_prefix = key[: len(rules["prefix"])] if len(key) >= len(rules["prefix"]) else key[:6]
return f"Key found ({source}) but invalid — expected prefix '{rules['prefix']}', got '{actual_prefix}...'"
return f"Key found (secrets) but invalid — expected prefix '{rules['prefix']}', got '{actual_prefix}...'"
if "min_length" in rules and len(key) < rules["min_length"]:
return f"Key found ({source}) but too short — {len(key)} chars, need {rules['min_length']}+"
return f"Key found (secrets) but too short — {len(key)} chars, need {rules['min_length']}+"
return f"Key found ({source}) but failed validation"
return "Key found (secrets) but failed validation"
src/aipass/api/apps/modules/api_key.py
+15
View File
@@ -168,6 +168,21 @@ def init_env():
error("Failed to create environment template")
def fetch_api_key(provider: str = "openrouter"):
"""Retrieve a validated API key for a provider from secrets."""
return keys.get_api_key(provider)
def fetch_validate_key(key: str, provider: str = "openrouter") -> bool:
"""Validate an API key format for a given provider."""
return keys.validate_key(key, provider)
def get_validation_rules(provider: str) -> dict:
"""Retrieve validation rules for a provider from the auth handler."""
return keys.get_validation_rules(provider)
def print_help():
"""Print help output for API key management"""
import argparse
src/aipass/api/apps/modules/openrouter_client.py
+5
View File
@@ -354,6 +354,11 @@ def get_response(prompt: str, caller: str | None = None, model: str | None = Non
return client.get_response(prompt, caller, model, **kwargs)
def extract_response(response):
"""Public API: Extract content from an OpenRouter API response object."""
return client.extract_response(response)
if __name__ == "__main__":
"""Standalone execution mode"""
args = sys.argv[1:]
src/aipass/api/tests/test_api_key.py
+8 -131
View File
@@ -522,168 +522,45 @@ def test_handle_command_propagates_exception(mock_header, mock_console, mock_jh,
# =============================================
# get_key_from_config tests (auth.keys handler)
# =============================================
from aipass.api.apps.handlers.auth import keys as auth_keys
class TestGetKeyFromConfig:
"""Tests for auth.keys.get_key_from_config()."""
def test_returns_key_from_valid_config(self, tmp_path, monkeypatch):
"""Valid config JSON should return the API key string."""
import json
config_dir = tmp_path / "api_json"
config_dir.mkdir()
config_file = config_dir / "api_connect_config.json"
config_file.write_text(
json.dumps({"config": {"providers": {"openrouter": {"api_key": "FAKE-sk-or-testkey-abc123"}}}}),
encoding="utf-8",
)
monkeypatch.setattr(auth_keys, "API_JSON_DIR", config_dir)
result = auth_keys.get_key_from_config("openrouter")
assert result == "FAKE-sk-or-testkey-abc123"
def test_returns_none_when_config_file_missing(self, tmp_path, monkeypatch):
"""Missing config file should return None."""
config_dir = tmp_path / "api_json"
config_dir.mkdir()
monkeypatch.setattr(auth_keys, "API_JSON_DIR", config_dir)
result = auth_keys.get_key_from_config("openrouter")
assert result is None
def test_returns_none_when_provider_not_in_config(self, tmp_path, monkeypatch):
"""Config exists but provider not listed should return None."""
import json
config_dir = tmp_path / "api_json"
config_dir.mkdir()
config_file = config_dir / "api_connect_config.json"
config_file.write_text(
json.dumps({"config": {"providers": {"openai": {"api_key": "FAKE-sk-openai-key-123"}}}}), encoding="utf-8"
)
monkeypatch.setattr(auth_keys, "API_JSON_DIR", config_dir)
result = auth_keys.get_key_from_config("openrouter")
assert result is None
def test_returns_none_when_api_key_empty(self, tmp_path, monkeypatch):
"""Provider present but api_key is empty string should return None."""
import json
config_dir = tmp_path / "api_json"
config_dir.mkdir()
config_file = config_dir / "api_connect_config.json"
config_file.write_text(json.dumps({"config": {"providers": {"openrouter": {"api_key": ""}}}}), encoding="utf-8")
monkeypatch.setattr(auth_keys, "API_JSON_DIR", config_dir)
result = auth_keys.get_key_from_config("openrouter")
assert result is None
def test_returns_none_when_config_missing_config_key(self, tmp_path, monkeypatch):
"""JSON file without 'config' top-level key should return None."""
import json
config_dir = tmp_path / "api_json"
config_dir.mkdir()
config_file = config_dir / "api_connect_config.json"
config_file.write_text(json.dumps({"other": "data"}), encoding="utf-8")
monkeypatch.setattr(auth_keys, "API_JSON_DIR", config_dir)
result = auth_keys.get_key_from_config("openrouter")
assert result is None
@patch("aipass.api.apps.handlers.auth.keys.logger")
def test_returns_none_on_invalid_json(self, mock_logger, tmp_path, monkeypatch):
"""Malformed JSON should return None and log error."""
config_dir = tmp_path / "api_json"
config_dir.mkdir()
config_file = config_dir / "api_connect_config.json"
config_file.write_text("not valid json {{{", encoding="utf-8")
monkeypatch.setattr(auth_keys, "API_JSON_DIR", config_dir)
result = auth_keys.get_key_from_config("openrouter")
assert result is None
mock_logger.error.assert_called_once()
def test_reads_different_providers(self, tmp_path, monkeypatch):
"""Should retrieve keys for different provider names."""
import json
config_dir = tmp_path / "api_json"
config_dir.mkdir()
config_file = config_dir / "api_connect_config.json"
config_file.write_text(
json.dumps(
{
"config": {
"providers": {
"openrouter": {"api_key": "FAKE-sk-or-key"},
"openai": {"api_key": "FAKE-sk-openai-key"},
"anthropic": {"api_key": "FAKE-sk-ant-key"},
}
}
}
),
encoding="utf-8",
)
monkeypatch.setattr(auth_keys, "API_JSON_DIR", config_dir)
assert auth_keys.get_key_from_config("openrouter") == "FAKE-sk-or-key"
assert auth_keys.get_key_from_config("openai") == "FAKE-sk-openai-key"
assert auth_keys.get_key_from_config("anthropic") == "FAKE-sk-ant-key"
# =============================================
# get_validation_rules tests (auth.keys handler)
# get_validation_rules tests (via module entry point)
# =============================================
class TestGetValidationRulesAuthKeys:
"""Tests for auth.keys.get_validation_rules()."""
"""Tests for api_key.get_validation_rules() — module entry point."""
def test_openrouter_rules(self):
"""openrouter should have prefix 'sk-or-v1-' and min_length 40."""
rules = auth_keys.get_validation_rules("openrouter")
rules = api_key.get_validation_rules("openrouter")
assert rules["prefix"] == "sk-or-v1-"
assert rules["min_length"] == 40
def test_openai_rules(self):
"""openai should have prefix 'sk-' and min_length 40."""
rules = auth_keys.get_validation_rules("openai")
rules = api_key.get_validation_rules("openai")
assert rules["prefix"] == "sk-"
assert rules["min_length"] == 40
def test_anthropic_rules(self):
"""anthropic should have prefix 'sk-ant-' and min_length 40."""
rules = auth_keys.get_validation_rules("anthropic")
rules = api_key.get_validation_rules("anthropic")
assert rules["prefix"] == "sk-ant-"
assert rules["min_length"] == 40
def test_unknown_provider_falls_back_to_generic(self):
"""Unknown provider should fall back to generic rules."""
rules = auth_keys.get_validation_rules("unknown_provider")
rules = api_key.get_validation_rules("unknown_provider")
assert rules["min_length"] == 10
assert "prefix" not in rules
def test_generic_rules_directly(self):
"""Requesting 'generic' should return generic rules."""
rules = auth_keys.get_validation_rules("generic")
rules = api_key.get_validation_rules("generic")
assert rules["min_length"] == 10
assert "prefix" not in rules
def test_return_type_is_dict(self):
"""All providers should return a dict."""
for provider in ["openrouter", "openai", "anthropic", "generic", "nonexistent"]:
rules = auth_keys.get_validation_rules(provider)
rules = api_key.get_validation_rules(provider)
assert isinstance(rules, dict)
src/aipass/api/tests/test_critical_paths.py
+40 -112
View File
@@ -11,7 +11,7 @@ Critical path tests for the API branch.
Covers the 4 core functions that form the API request pipeline:
1. get_api_key() - Key retrieval from config JSON and secrets file
1. get_api_key() - Key retrieval from secrets file
2. validate_key() - Key format validation per provider rules
3. get_response() - Main API call orchestrator
4. extract_response() - Response content extraction
@@ -19,9 +19,11 @@ Covers the 4 core functions that form the API request pipeline:
All external dependencies are mocked. File-based tests use tmp_path.
"""
import json
from unittest.mock import patch, MagicMock
from aipass.api.apps.modules import api_key
from aipass.api.apps.modules import openrouter_client
# =============================================
# 1. get_api_key() tests
@@ -29,43 +31,11 @@ from unittest.mock import patch, MagicMock
class TestGetApiKey:
"""Tests for get_api_key() — key retrieval from config and secrets sources."""
"""Tests for fetch_api_key() — key retrieval from secrets file (single source)."""
@patch("aipass.api.apps.handlers.auth.keys.json_handler")
@patch("aipass.api.apps.handlers.auth.keys.API_JSON_DIR")
def test_key_from_config_json(self, mock_api_json_dir, mock_jh, tmp_path):
"""Key found in api_connect_config.json is returned after validation."""
from aipass.api.apps.handlers.auth.keys import get_api_key
config_path = tmp_path / "api_connect_config.json"
config_data = {
"config": {
"providers": {"openrouter": {"api_key": "sk-or-v1-NOTREAL-test-000000000000000000000000000000000000"}}
}
}
config_path.write_text(json.dumps(config_data), encoding="utf-8")
mock_api_json_dir.__truediv__ = lambda self, name: tmp_path / name
result = get_api_key("openrouter")
# do not add real api kets here
assert result == "sk-or-v1-NOTREAL-test-000000000000000000000000000000000000"
mock_jh.log_operation.assert_called_once()
@patch("aipass.api.apps.handlers.auth.keys.json_handler")
@patch("aipass.api.apps.handlers.auth.keys.API_JSON_DIR")
def test_key_from_secrets_file(self, mock_api_json_dir, mock_jh, tmp_path):
"""Key found in ~/.secrets/aipass/.env when config has no key."""
from aipass.api.apps.handlers.auth.keys import get_api_key
# Config file exists but has no key for openrouter
config_path = tmp_path / "api_connect_config.json"
config_data = {"config": {"providers": {}}}
config_path.write_text(json.dumps(config_data), encoding="utf-8")
mock_api_json_dir.__truediv__ = lambda self, name: tmp_path / name
# Create secrets file
def test_key_from_secrets_file(self, mock_jh, tmp_path):
"""Key found in ~/.secrets/aipass/.env is returned after validation."""
secrets_dir = tmp_path / ".secrets" / "aipass"
secrets_dir.mkdir(parents=True)
env_file = secrets_dir / ".env"
@@ -75,47 +45,35 @@ class TestGetApiKey:
)
with patch("aipass.api.apps.handlers.auth.keys.Path") as mock_path_cls:
# Path.home() should return tmp_path so secrets resolve there
mock_path_cls.home.return_value = tmp_path
result = get_api_key("openrouter")
result = api_key.fetch_api_key("openrouter")
assert result == "sk-or-v1-NOTREAL-test-00000000000000000000000000"
mock_jh.log_operation.assert_called_once()
@patch("aipass.api.apps.handlers.auth.keys.json_handler")
@patch("aipass.api.apps.handlers.auth.keys.API_JSON_DIR")
def test_no_key_found_returns_none(self, mock_api_json_dir, mock_jh, tmp_path):
"""Returns None when no key exists in any source."""
from aipass.api.apps.handlers.auth.keys import get_api_key
# Config file with no providers
config_path = tmp_path / "api_connect_config.json"
config_data = {"config": {"providers": {}}}
config_path.write_text(json.dumps(config_data), encoding="utf-8")
mock_api_json_dir.__truediv__ = lambda self, name: tmp_path / name
# No secrets file exists
def test_no_key_found_returns_none(self, mock_jh, tmp_path):
"""Returns None when no secrets file exists."""
with patch("aipass.api.apps.handlers.auth.keys.Path") as mock_path_cls:
mock_home = tmp_path / "nonexistent_home"
mock_path_cls.home.return_value = mock_home
result = get_api_key("openrouter")
mock_path_cls.home.return_value = tmp_path / "nonexistent_home"
result = api_key.fetch_api_key("openrouter")
assert result is None
@patch("aipass.api.apps.handlers.auth.keys.json_handler")
@patch("aipass.api.apps.handlers.auth.keys.API_JSON_DIR")
def test_invalid_key_format_returns_none(self, mock_api_json_dir, mock_jh, tmp_path):
"""Key exists in config but fails validation (wrong prefix) returns None."""
from aipass.api.apps.handlers.auth.keys import get_api_key
def test_invalid_key_format_returns_none(self, mock_jh, tmp_path):
"""Key in secrets with wrong prefix returns None."""
secrets_dir = tmp_path / ".secrets" / "aipass"
secrets_dir.mkdir(parents=True)
env_file = secrets_dir / ".env"
env_file.write_text(
"OPENROUTER_API_KEY=INVALID-PREFIX-key-that-is-long-enough-to-pass\n",
encoding="utf-8",
)
config_path = tmp_path / "api_connect_config.json"
config_data = {"config": {"providers": {"openrouter": {"api_key": "INVALID-PREFIX-key-that-is-long-enough"}}}}
config_path.write_text(json.dumps(config_data), encoding="utf-8")
mock_api_json_dir.__truediv__ = lambda self, name: tmp_path / name
# No secrets file fallback
with patch("aipass.api.apps.handlers.auth.keys.Path") as mock_path_cls:
mock_path_cls.home.return_value = tmp_path / "no_home"
result = get_api_key("openrouter")
mock_path_cls.home.return_value = tmp_path
result = api_key.fetch_api_key("openrouter")
assert result is None
@@ -126,54 +84,40 @@ class TestGetApiKey:
class TestValidateKey:
"""Tests for validate_key() — key format validation per provider rules."""
"""Tests for fetch_validate_key() — key format validation per provider rules."""
def test_valid_openrouter_key(self):
"""Valid openrouter key with correct prefix and length passes."""
from aipass.api.apps.handlers.auth.keys import validate_key
key = "sk-or-v1-NOTREAL-test-000000000000000000000000000000000000"
assert validate_key(key, "openrouter") is True
assert api_key.fetch_validate_key(key, "openrouter") is True
def test_key_too_short(self):
"""Key shorter than min_length fails validation."""
from aipass.api.apps.handlers.auth.keys import validate_key
key = "sk-or-v1-short"
assert len(key) < 20
assert validate_key(key, "openrouter") is False
assert api_key.fetch_validate_key(key, "openrouter") is False
def test_wrong_prefix(self):
"""Key with wrong prefix for provider fails validation."""
from aipass.api.apps.handlers.auth.keys import validate_key
key = "sk-wrong-prefix-but-long-enough-to-pass-length"
assert validate_key(key, "openrouter") is False
assert api_key.fetch_validate_key(key, "openrouter") is False
def test_empty_key(self):
"""Empty string key fails validation."""
from aipass.api.apps.handlers.auth.keys import validate_key
assert validate_key("", "openrouter") is False
assert api_key.fetch_validate_key("", "openrouter") is False
def test_none_key(self):
"""None key fails validation."""
from aipass.api.apps.handlers.auth.keys import validate_key
assert validate_key(None, "openrouter") is False # type: ignore[arg-type]
assert api_key.fetch_validate_key(None, "openrouter") is False # type: ignore[arg-type]
def test_generic_provider_no_prefix_required(self):
"""Generic provider only checks min_length, no prefix required."""
from aipass.api.apps.handlers.auth.keys import validate_key
key = "any-key-that-is-long-enough"
assert validate_key(key, "unknown_provider") is True
assert api_key.fetch_validate_key(key, "unknown_provider") is True
def test_generic_provider_too_short(self):
"""Generic provider rejects keys shorter than 10 chars."""
from aipass.api.apps.handlers.auth.keys import validate_key
assert validate_key("short", "unknown_provider") is False
assert api_key.fetch_validate_key("short", "unknown_provider") is False
# =============================================
@@ -205,8 +149,6 @@ class TestGetResponse:
mock_track,
):
"""Full successful pipeline: detect caller, get key, make request, extract, track."""
from aipass.api.apps.handlers.openrouter.client import get_response
mock_caller_info.return_value = {"caller_name": "test-branch"}
mock_get_key.return_value = "FAKE-sk-or-v1-testkey"
mock_get_client.return_value = MagicMock()
@@ -217,7 +159,7 @@ class TestGetResponse:
"model": "anthropic/claude-3.5-sonnet",
}
result = get_response("What is Python?", model="anthropic/claude-3.5-sonnet")
result = openrouter_client.get_response("What is Python?", model="anthropic/claude-3.5-sonnet")
assert result is not None
assert result["content"] == "Hello, world!"
@@ -229,11 +171,9 @@ class TestGetResponse:
@patch(f"{MODULE}.get_caller_info")
def test_no_model_returns_none(self, mock_caller_info, mock_ensure, mock_get_key):
"""Missing model parameter returns None without making API call."""
from aipass.api.apps.handlers.openrouter.client import get_response
mock_caller_info.return_value = {"caller_name": "test"}
result = get_response("What is Python?", model=None)
result = openrouter_client.get_response("What is Python?", model=None)
assert result is None
mock_get_key.assert_not_called()
@@ -243,12 +183,10 @@ class TestGetResponse:
@patch(f"{MODULE}.get_caller_info")
def test_no_api_key_returns_none(self, mock_caller_info, mock_ensure, mock_get_key):
"""No API key available returns None."""
from aipass.api.apps.handlers.openrouter.client import get_response
mock_caller_info.return_value = {"caller_name": "test"}
mock_get_key.return_value = None
result = get_response("What is Python?", model="anthropic/claude-3.5-sonnet")
result = openrouter_client.get_response("What is Python?", model="anthropic/claude-3.5-sonnet")
assert result is None
@@ -263,8 +201,6 @@ class TestExtractResponse:
def test_valid_response(self):
"""Extracts content, id, and model from a well-formed response."""
from aipass.api.apps.handlers.openrouter.client import extract_response
choice = MagicMock()
choice.message.content = "The answer is 42."
choice.finish_reason = "stop"
@@ -274,7 +210,7 @@ class TestExtractResponse:
response.id = "gen-xyz789"
response.model = "anthropic/claude-3.5-sonnet"
result = extract_response(response)
result = openrouter_client.extract_response(response)
assert result is not None
assert result["content"] == "The answer is 42."
@@ -283,23 +219,17 @@ class TestExtractResponse:
def test_none_response(self):
"""None response returns None."""
from aipass.api.apps.handlers.openrouter.client import extract_response
assert extract_response(None) is None
assert openrouter_client.extract_response(None) is None
def test_response_no_choices(self):
"""Response with empty choices list returns None."""
from aipass.api.apps.handlers.openrouter.client import extract_response
response = MagicMock()
response.choices = []
assert extract_response(response) is None
assert openrouter_client.extract_response(response) is None
def test_response_no_content(self):
"""Response with choice but no content returns None."""
from aipass.api.apps.handlers.openrouter.client import extract_response
choice = MagicMock()
choice.message.content = None
@@ -308,14 +238,12 @@ class TestExtractResponse:
response.id = "gen-empty"
response.model = "test/model"
assert extract_response(response) is None
assert openrouter_client.extract_response(response) is None
def test_response_missing_message(self):
"""Response choice without message attribute returns None."""
from aipass.api.apps.handlers.openrouter.client import extract_response
choice = MagicMock(spec=[]) # No attributes at all
response = MagicMock()
response.choices = [choice]
assert extract_response(response) is None
assert openrouter_client.extract_response(response) is None