feat: add privacy router and zero-cost safeguards

This commit is contained in:
2026-07-30 17:54:06 +02:00
parent 570e7fa153
commit 286d96da33
8 changed files with 531 additions and 13 deletions
+73
View File
@@ -0,0 +1,73 @@
import unittest
from javis.providers.base import ChatMessage
from javis.security.privacy import CloudPolicy, PrivacyRouter
class PrivacyRouterTests(unittest.TestCase):
def setUp(self) -> None:
self.router = PrivacyRouter()
def test_general_technical_question_is_allowed(self) -> None:
self.assertEqual(
self.router.classify("Wie funktioniert SQLite in Python?").policy,
CloudPolicy.ALLOWED,
)
def test_personal_information_requires_approval(self) -> None:
self.assertEqual(
self.router.classify("Ich möchte meinen Alltag besser planen.").policy,
CloudPolicy.ASK,
)
def test_family_information_requires_approval(self) -> None:
self.assertEqual(
self.router.classify("Meine Frau hat morgen einen Termin.").policy,
CloudPolicy.ASK,
)
def test_secret_pattern_is_never(self) -> None:
decision = self.router.classify("Mein API-Key ist AIza" + "x" * 25)
self.assertEqual(decision.policy, CloudPolicy.NEVER)
def test_financial_and_health_information_are_never(self) -> None:
self.assertEqual(
self.router.classify("Meine IBAN ist ein Geheimnis.").policy,
CloudPolicy.NEVER,
)
self.assertEqual(
self.router.classify("Meine Diagnose soll ausgewertet werden.").policy,
CloudPolicy.NEVER,
)
def test_unknown_content_is_conservatively_ask(self) -> None:
self.assertEqual(
self.router.classify("Erkläre das bitte genauer.").policy,
CloudPolicy.ASK,
)
def test_explicit_cloud_request_cannot_override_never(self) -> None:
decision = self.router.classify(
"Sende mein Passwort an die Cloud.",
cloud_requested=True,
)
self.assertEqual(decision.policy, CloudPolicy.NEVER)
def test_minimal_context_excludes_sensitive_history(self) -> None:
messages = [
ChatMessage("user", "Meine Diagnose ist privat."),
ChatMessage("assistant", "Das bleibt lokal."),
ChatMessage("user", "Wie funktioniert SQLite?"),
]
selected = self.router.minimal_context(
messages,
current_policy=CloudPolicy.ALLOWED,
approved=False,
max_chars=1_000,
max_messages=6,
)
self.assertEqual(selected, [ChatMessage("user", "Wie funktioniert SQLite?")])
if __name__ == "__main__":
unittest.main()
+27
View File
@@ -34,6 +34,33 @@ class SettingsTests(unittest.TestCase):
}
)
def test_zero_cost_defaults_are_conservative(self) -> None:
settings = Settings.from_env({"JAVIS_DATA_DIR": str(Path.cwd())})
self.assertTrue(settings.free_only)
self.assertFalse(settings.gemini_enabled)
self.assertFalse(settings.billing_confirmed_disabled)
self.assertEqual(settings.gemini_model, "gemini-3.6-flash")
self.assertEqual(settings.provider_mode, "auto")
def test_paid_mode_is_rejected(self) -> None:
with self.assertRaises(ConfigurationError):
Settings.from_env(
{
"JAVIS_DATA_DIR": str(Path.cwd()),
"JAVIS_FREE_ONLY": "false",
}
)
def test_cloud_limits_are_bounded(self) -> None:
with self.assertRaises(ConfigurationError):
Settings.from_env(
{
"JAVIS_DATA_DIR": str(Path.cwd()),
"JAVIS_MAX_CLOUD_REQUESTS_PER_DAY": "1001",
}
)
if __name__ == "__main__":
unittest.main()
+46
View File
@@ -0,0 +1,46 @@
import tempfile
import unittest
from datetime import UTC, datetime
from pathlib import Path
from javis.memory.usage_store import ProviderEvent, SQLiteUsageStore
class UsageStoreTests(unittest.TestCase):
def test_records_only_metadata_and_counts_daily_usage(self) -> None:
with tempfile.TemporaryDirectory() as directory:
database = Path(directory) / "usage.sqlite3"
store = SQLiteUsageStore(database)
store.record(
ProviderEvent(
provider="gemini",
model="gemini-test",
success=False,
error_category="quota",
fallback=False,
privacy_policy="allowed",
input_tokens=12,
output_tokens=None,
)
)
store.record(
ProviderEvent(
provider="ollama",
model="local-test",
success=True,
error_category=None,
fallback=True,
privacy_policy="allowed",
)
)
today = datetime.now(UTC).date()
self.assertEqual(store.cloud_requests_on(today), 1)
self.assertEqual(store.local_fallbacks_on(today), 1)
raw_database = database.read_bytes()
self.assertNotIn(b"prompt", raw_database)
self.assertNotIn(b"answer", raw_database)
if __name__ == "__main__":
unittest.main()