164 lines
6.2 KiB
Python
164 lines
6.2 KiB
Python
"""Unit tests for the tool/connected-source key store (#103).
|
|
|
|
Covers ``backend.tools.secrets`` (precedence, masking, whitelist) and the
|
|
``/api/config/tool-keys`` endpoints (masked GET, POST, DELETE, admin-only).
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from backend.tools.secrets import (
|
|
TOOL_KEY_NAMES,
|
|
delete_tool_key,
|
|
get_tool_key,
|
|
is_secret_name,
|
|
mask_value,
|
|
set_tool_key,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def key_store(tmp_path, monkeypatch):
|
|
"""Isolated key store directory."""
|
|
monkeypatch.setenv("OBSIGATE_DATA_DIR", str(tmp_path))
|
|
yield tmp_path
|
|
|
|
|
|
def _write_store(tmp_path, data):
|
|
(tmp_path / "api_keys.json").write_text(json.dumps(data), encoding="utf-8")
|
|
|
|
|
|
class TestGetToolKey:
|
|
def test_stored_value_takes_precedence_over_env(self, key_store, monkeypatch):
|
|
_write_store(key_store, {"OBSIGATE_GITHUB_TOKEN": "stored-token"})
|
|
monkeypatch.setenv("OBSIGATE_GITHUB_TOKEN", "env-token")
|
|
assert get_tool_key("OBSIGATE_GITHUB_TOKEN") == "stored-token"
|
|
|
|
def test_env_fallback_when_not_stored(self, key_store, monkeypatch):
|
|
monkeypatch.setenv("OBSIGATE_GITEA_TOKEN", "env-token")
|
|
assert get_tool_key("OBSIGATE_GITEA_TOKEN") == "env-token"
|
|
|
|
def test_missing_everywhere_returns_empty(self, key_store, monkeypatch):
|
|
monkeypatch.delenv("OBSIGATE_GITHUB_TOKEN", raising=False)
|
|
assert get_tool_key("OBSIGATE_GITHUB_TOKEN") == ""
|
|
|
|
def test_non_whitelisted_name_uses_env_only(self, key_store, monkeypatch):
|
|
_write_store(key_store, {"OTHER_KEY": "stored"})
|
|
monkeypatch.setenv("OTHER_KEY", "env")
|
|
assert get_tool_key("OTHER_KEY") == "env"
|
|
|
|
def test_whitelist_covers_expected_names(self):
|
|
assert set(TOOL_KEY_NAMES) == {
|
|
"OBSIGATE_TAVILY_API_KEY",
|
|
"OBSIGATE_BRAVE_API_KEY",
|
|
"OBSIGATE_SERPAPI_API_KEY",
|
|
"OBSIGATE_EXA_API_KEY",
|
|
"OBSIGATE_GITEA_URL",
|
|
"OBSIGATE_GITEA_TOKEN",
|
|
"OBSIGATE_GITHUB_TOKEN",
|
|
}
|
|
|
|
|
|
class TestSetDelete:
|
|
def test_set_then_get_roundtrip(self, key_store):
|
|
set_tool_key("OBSIGATE_TAVILY_API_KEY", "tvly-1234")
|
|
assert get_tool_key("OBSIGATE_TAVILY_API_KEY") == "tvly-1234"
|
|
|
|
def test_set_empty_value_deletes_entry(self, key_store):
|
|
set_tool_key("OBSIGATE_TAVILY_API_KEY", "tvly-1234")
|
|
set_tool_key("OBSIGATE_TAVILY_API_KEY", "")
|
|
assert get_tool_key("OBSIGATE_TAVILY_API_KEY") == ""
|
|
assert "OBSIGATE_TAVILY_API_KEY" not in json.loads(
|
|
(key_store / "api_keys.json").read_text(encoding="utf-8")
|
|
)
|
|
|
|
def test_delete_removes_and_reports(self, key_store):
|
|
set_tool_key("OBSIGATE_GITEA_URL", "https://git.example.net")
|
|
assert delete_tool_key("OBSIGATE_GITEA_URL") is True
|
|
assert delete_tool_key("OBSIGATE_GITEA_URL") is False
|
|
|
|
def test_unknown_name_rejected(self, key_store):
|
|
with pytest.raises(ValueError):
|
|
set_tool_key("NOT_WHITELISTED", "x")
|
|
with pytest.raises(ValueError):
|
|
delete_tool_key("NOT_WHITELISTED")
|
|
|
|
def test_store_keeps_other_entries(self, key_store):
|
|
_write_store(key_store, {"DEEPSEEK_API_KEY": "sk-existing"})
|
|
set_tool_key("OBSIGATE_EXA_API_KEY", "exa-key")
|
|
data = json.loads((key_store / "api_keys.json").read_text(encoding="utf-8"))
|
|
assert data["DEEPSEEK_API_KEY"] == "sk-existing"
|
|
assert data["OBSIGATE_EXA_API_KEY"] == "exa-key"
|
|
|
|
|
|
class TestMasking:
|
|
def test_urls_returned_clear(self):
|
|
assert mask_value("OBSIGATE_GITEA_URL", "https://git.example.net") == \
|
|
"https://git.example.net"
|
|
|
|
def test_tokens_masked(self):
|
|
masked = mask_value("OBSIGATE_GITHUB_TOKEN", "ghp_abcdefgh1234")
|
|
assert masked.startswith("ghp_")
|
|
assert "abcdefgh1234" not in masked
|
|
assert "..." in masked
|
|
|
|
def test_short_secret_fully_masked(self):
|
|
assert mask_value("OBSIGATE_EXA_API_KEY", "abc") == "***"
|
|
|
|
def test_secret_detection(self):
|
|
assert is_secret_name("OBSIGATE_GITEA_TOKEN")
|
|
assert is_secret_name("OBSIGATE_TAVILY_API_KEY")
|
|
assert not is_secret_name("OBSIGATE_GITEA_URL")
|
|
|
|
|
|
class TestToolKeysAPI:
|
|
def _login(self, admin_client):
|
|
resp = admin_client.post(
|
|
"/api/auth/login", json={"username": "admin", "password": "chab30"}
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
token = resp.json()["access_token"]
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
def test_get_masks_tokens_shows_urls(self, admin_client, key_store):
|
|
headers = self._login(admin_client)
|
|
_write_store(key_store, {
|
|
"OBSIGATE_GITEA_URL": "https://git.example.net",
|
|
"OBSIGATE_GITHUB_TOKEN": "ghp_abcdefgh1234",
|
|
})
|
|
resp = admin_client.get("/api/config/tool-keys", headers=headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["OBSIGATE_GITEA_URL"] == "https://git.example.net"
|
|
assert "abcdefgh1234" not in data["OBSIGATE_GITHUB_TOKEN"]
|
|
assert data["OBSIGATE_GITHUB_TOKEN"].startswith("ghp_")
|
|
|
|
def test_post_roundtrip_then_delete(self, admin_client, key_store):
|
|
headers = self._login(admin_client)
|
|
resp = admin_client.post(
|
|
"/api/config/tool-keys",
|
|
headers=headers,
|
|
json={"OBSIGATE_EXA_API_KEY": "exa-key-1234"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
assert get_tool_key("OBSIGATE_EXA_API_KEY") == "exa-key-1234"
|
|
|
|
resp = admin_client.delete("/api/config/tool-keys/OBSIGATE_EXA_API_KEY", headers=headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "deleted"
|
|
assert get_tool_key("OBSIGATE_EXA_API_KEY") == ""
|
|
|
|
def test_post_unknown_name_rejected(self, admin_client, key_store):
|
|
headers = self._login(admin_client)
|
|
resp = admin_client.post(
|
|
"/api/config/tool-keys", headers=headers, json={"MY_SECRET": "x"}
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
def test_requires_admin(self, admin_client, key_store):
|
|
resp = admin_client.get("/api/config/tool-keys", headers={})
|
|
assert resp.status_code in (401, 403)
|