feat(config): cles des sources connectees et recherche a cle editables depuis la page Configurations (#103)
This commit is contained in:
@@ -0,0 +1,163 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user