Spaces:
Sleeping
Sleeping
| import json | |
| import logging | |
| import unittest | |
| from observability import ( | |
| GenerationStats, | |
| JsonFormatter, | |
| RuntimeMetrics, | |
| sanitize, | |
| summarize_messages, | |
| ) | |
| class ObservabilityTests(unittest.TestCase): | |
| def test_sanitize_redacts_secrets_and_truncates_text(self): | |
| sanitized = sanitize( | |
| { | |
| "authorization": "Bearer secret", | |
| "nested": { | |
| "access-token": "secret", | |
| "x-api-key": "secret", | |
| "content": "abcdefgh", | |
| }, | |
| }, | |
| max_text_chars=4, | |
| ) | |
| self.assertEqual(sanitized["authorization"], "[REDACTED]") | |
| self.assertEqual(sanitized["nested"]["access-token"], "[REDACTED]") | |
| self.assertEqual(sanitized["nested"]["x-api-key"], "[REDACTED]") | |
| self.assertEqual( | |
| sanitized["nested"]["content"], | |
| "abcd...[truncated 4 chars]", | |
| ) | |
| def test_message_summary_excludes_payload_text(self): | |
| summary = summarize_messages( | |
| [ | |
| {"role": "system", "content": "rules"}, | |
| {"role": "user", "content": "hello"}, | |
| { | |
| "role": "assistant", | |
| "content": None, | |
| "tool_calls": [{"function": {"name": "open_page"}}], | |
| }, | |
| ] | |
| ) | |
| self.assertEqual(summary["message_count"], 3) | |
| self.assertEqual(summary["text_chars"], 10) | |
| self.assertEqual(summary["tool_call_count"], 1) | |
| self.assertNotIn("hello", json.dumps(summary)) | |
| def test_json_formatter_includes_structured_fields(self): | |
| record = logging.LogRecord( | |
| "browso.server", | |
| logging.INFO, | |
| __file__, | |
| 1, | |
| "request.completed", | |
| (), | |
| None, | |
| ) | |
| record.event_fields = {"event": "request.completed", "status": 200} | |
| payload = json.loads(JsonFormatter().format(record)) | |
| self.assertEqual(payload["event"], "request.completed") | |
| self.assertEqual(payload["status"], 200) | |
| def test_runtime_metrics_tracks_generation_usage(self): | |
| metrics = RuntimeMetrics() | |
| metrics.record_http(200) | |
| metrics.generation_started() | |
| metrics.generation_finished( | |
| GenerationStats(prompt_tokens=12, completion_tokens=4), | |
| failed=False, | |
| ) | |
| snapshot = metrics.snapshot() | |
| self.assertEqual(snapshot["http_requests"], 1) | |
| self.assertEqual(snapshot["completions"], 1) | |
| self.assertEqual(snapshot["active_generations"], 0) | |
| self.assertEqual(snapshot["prompt_tokens"], 12) | |
| self.assertEqual(snapshot["completion_tokens"], 4) | |
| if __name__ == "__main__": | |
| unittest.main() | |