Spaces:
Running
Running
File size: 2,084 Bytes
047d1bc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | import math
import unittest
from api.auth_guard import require_private_state_machine
from api.private_state import RagSearchIn, SkillPatternIn, TelegramConfigIn, _as_epoch_ms, _cosine_similarity, _parse_vector, router
class PrivateStateContractTests(unittest.TestCase):
def test_router_is_namespaced_under_private_state(self):
self.assertEqual(router.prefix, "/api/private-state")
paths = {route.path for route in router.routes}
self.assertIn("/api/private-state/sessions", paths)
self.assertIn("/api/private-state/tasks", paths)
self.assertIn("/api/private-state/rag/index", paths)
self.assertIn("/api/private-state/rag/search", paths)
self.assertEqual(router.dependencies[0].dependency, require_private_state_machine)
def test_rag_search_allows_lexical_fallback_without_embedding(self):
payload = RagSearchIn(query="contesto progetto", query_embedding=None)
self.assertEqual(payload.query, "contesto progetto")
self.assertIsNone(payload.query_embedding)
def test_rag_search_rejects_non_finite_embedding(self):
with self.assertRaises(ValueError):
RagSearchIn(query_embedding=[1.0, math.inf])
def test_skill_pattern_rejects_empty_tool_sequence(self):
with self.assertRaises(ValueError):
SkillPatternIn(
id="pattern", task_signature="task", tool_sequence=[" "],
success_count=0, total_count=1, last_used=1, confidence=0,
)
def test_timestamp_is_normalized_for_browser_clients(self):
self.assertEqual(_as_epoch_ms("1970-01-01T00:00:01+00:00"), 1_000)
self.assertEqual(_as_epoch_ms("not-a-timestamp"), 0)
def test_private_payload_has_bounded_required_fields(self):
config = TelegramConfigIn(bot_token="token", chat_id="chat")
self.assertEqual(config.chat_id, "chat")
self.assertEqual(_parse_vector("[1, 2]"), [1.0, 2.0])
self.assertAlmostEqual(_cosine_similarity([1.0, 0.0], [1.0, 0.0]), 1.0)
if __name__ == "__main__":
unittest.main()
|