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()