mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 20:35:22 +08:00
214 lines
8.6 KiB
Python
214 lines
8.6 KiB
Python
"""Parser telemetry is scoped to the calling installation binding."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
from langbot_plugin.api.definition.components.page import PageRequest
|
|
from langbot_plugin.api.proxies.invocation import bind_invocation, current_config
|
|
from langbot_plugin.entities.io.context import InstallationBinding
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
|
|
|
|
def _binding(installation: str, *, workspace: str = "workspace-1", instance: str = "instance-1"):
|
|
return InstallationBinding(
|
|
instance_uuid=instance,
|
|
workspace_uuid=workspace,
|
|
installation_uuid=installation,
|
|
runtime_revision=1,
|
|
artifact_digest="a" * 64,
|
|
)
|
|
|
|
|
|
class _StubPlugin:
|
|
"""Minimal BasePlugin surface: per-invocation config + runtime handler."""
|
|
|
|
def __init__(self, handler):
|
|
self.plugin_runtime_handler = handler
|
|
|
|
def get_config(self) -> dict:
|
|
return dict(current_config(self.plugin_runtime_handler) or {})
|
|
|
|
def get_plugin_config(self) -> dict:
|
|
return self.get_config()
|
|
|
|
|
|
class ParserTelemetryTests(unittest.TestCase):
|
|
def test_snapshot_records_only_operational_metadata(self) -> None:
|
|
from components.observability.telemetry import ParserTelemetry
|
|
|
|
telemetry = ParserTelemetry(recent_limit=2)
|
|
telemetry.record_parse(
|
|
filename="/tmp/report.pdf",
|
|
mime_type="application/pdf",
|
|
extension="pdf",
|
|
extension_source="mime_type",
|
|
duration_ms=12.345,
|
|
text_chars=1200,
|
|
sections_count=4,
|
|
metadata={
|
|
"has_tables": True,
|
|
"has_images": True,
|
|
"images_count": 2,
|
|
"vision_used": True,
|
|
"vision_tasks_count": 3,
|
|
"vision_images_described_count": 2,
|
|
"vision_failed_count": 1,
|
|
},
|
|
)
|
|
|
|
snapshot = telemetry.snapshot()
|
|
|
|
self.assertEqual(snapshot["summary"]["total_parses"], 1)
|
|
self.assertEqual(snapshot["summary"]["failed_parses"], 0)
|
|
self.assertEqual(snapshot["summary"]["tables_detected"], 1)
|
|
self.assertEqual(snapshot["summary"]["images_detected"], 1)
|
|
self.assertEqual(snapshot["summary"]["vision_tasks_count"], 3)
|
|
self.assertEqual(snapshot["distributions"]["extensions"][0]["key"], "pdf")
|
|
self.assertEqual(snapshot["recent_parses"][0]["filename"], "report.pdf")
|
|
self.assertNotIn("text", snapshot["recent_parses"][0])
|
|
self.assertNotIn("file_content", snapshot["recent_parses"][0])
|
|
|
|
def test_ring_buffers_and_errors_are_bounded(self) -> None:
|
|
from components.observability.telemetry import ParserTelemetry
|
|
|
|
telemetry = ParserTelemetry(recent_limit=1)
|
|
telemetry.record_parse(
|
|
filename="ok.txt",
|
|
mime_type="text/plain",
|
|
extension="txt",
|
|
extension_source="mime_type",
|
|
duration_ms=1,
|
|
text_chars=2,
|
|
sections_count=1,
|
|
)
|
|
telemetry.record_parse(
|
|
filename="bad.pdf",
|
|
mime_type="application/pdf",
|
|
extension="pdf",
|
|
extension_source="mime_type",
|
|
duration_ms=2,
|
|
text_chars=0,
|
|
sections_count=0,
|
|
metadata={"parser_failed": True, "parse_error": "broken"},
|
|
)
|
|
|
|
snapshot = telemetry.snapshot()
|
|
|
|
self.assertEqual(snapshot["summary"]["total_parses"], 2)
|
|
self.assertEqual(snapshot["summary"]["failed_parses"], 1)
|
|
self.assertEqual(len(snapshot["recent_parses"]), 1)
|
|
self.assertEqual(snapshot["recent_parses"][0]["filename"], "bad.pdf")
|
|
self.assertEqual(len(snapshot["recent_errors"]), 1)
|
|
self.assertEqual(snapshot["recent_errors"][0]["parse_error"], "broken")
|
|
|
|
|
|
class TelemetryBindingScopeTests(unittest.IsolatedAsyncioTestCase):
|
|
"""Two installations through one component object must not share records."""
|
|
|
|
def setUp(self) -> None:
|
|
from components.general_parsers.general_parsers import GeneralParsers
|
|
from components.pages.observability import ParserObservabilityPage
|
|
|
|
# ONE object graph: a single handler, plugin, Parser and Page instance
|
|
# serve both installations, as in the shared-runtime worker.
|
|
self.handler = SimpleNamespace()
|
|
self.plugin = _StubPlugin(self.handler)
|
|
self.parser = GeneralParsers()
|
|
self.parser.plugin = self.plugin
|
|
self.page = ParserObservabilityPage()
|
|
self.page.plugin = self.plugin
|
|
|
|
async def _record_parse(self, binding, filename: str, config: dict | None = None) -> None:
|
|
from langbot_plugin.api.entities.builtin.rag.models import ParseContext
|
|
|
|
with bind_invocation(self.handler, config=config or {}, binding=binding):
|
|
await self.parser.parse(
|
|
ParseContext(file_content=b"hello world", filename=filename, mime_type="text/plain")
|
|
)
|
|
|
|
async def _snapshot(self, binding, endpoint: str = "/snapshot", method: str = "GET"):
|
|
with bind_invocation(self.handler, binding=binding):
|
|
return await self.page.handle_api(PageRequest(endpoint=endpoint, method=method))
|
|
|
|
async def test_records_are_not_visible_across_installations(self) -> None:
|
|
from components.observability import get_telemetry, release_telemetry
|
|
|
|
binding_a = _binding("installation-a")
|
|
binding_b = _binding("installation-b")
|
|
self.addCleanup(release_telemetry, binding_a)
|
|
self.addCleanup(release_telemetry, binding_b)
|
|
|
|
await self._record_parse(binding_a, "tenant-a-secret.pdf")
|
|
await self._record_parse(binding_b, "tenant-b-notes.md")
|
|
|
|
snapshot_b = await self._snapshot(binding_b)
|
|
self.assertIsNone(snapshot_b.error)
|
|
filenames_b = [row["filename"] for row in snapshot_b.data["recent_parses"]]
|
|
self.assertIn("tenant-b-notes.md", filenames_b)
|
|
# Pre-fix the module-level singleton leaked A's record into B's page.
|
|
self.assertNotIn("tenant-a-secret.pdf", filenames_b)
|
|
self.assertEqual(snapshot_b.data["summary"]["total_parses"], 1)
|
|
|
|
snapshot_a = await self._snapshot(binding_a)
|
|
filenames_a = [row["filename"] for row in snapshot_a.data["recent_parses"]]
|
|
self.assertIn("tenant-a-secret.pdf", filenames_a)
|
|
self.assertNotIn("tenant-b-notes.md", filenames_a)
|
|
|
|
self.assertEqual(get_telemetry(binding_a).snapshot()["summary"]["total_parses"], 1)
|
|
self.assertEqual(get_telemetry(binding_b).snapshot()["summary"]["total_parses"], 1)
|
|
|
|
async def test_clear_only_resets_the_calling_installation(self) -> None:
|
|
from components.observability import release_telemetry
|
|
|
|
binding_a = _binding("installation-clear-a")
|
|
binding_b = _binding("installation-clear-b")
|
|
self.addCleanup(release_telemetry, binding_a)
|
|
self.addCleanup(release_telemetry, binding_b)
|
|
|
|
await self._record_parse(binding_a, "clear-a.txt")
|
|
await self._record_parse(binding_b, "clear-b.txt")
|
|
|
|
cleared = await self._snapshot(binding_a, endpoint="/clear", method="POST")
|
|
self.assertIsNone(cleared.error)
|
|
self.assertEqual(cleared.data["summary"]["total_parses"], 0)
|
|
|
|
snapshot_b = await self._snapshot(binding_b)
|
|
self.assertEqual(snapshot_b.data["summary"]["total_parses"], 1)
|
|
self.assertEqual(snapshot_b.data["recent_parses"][0]["filename"], "clear-b.txt")
|
|
|
|
async def test_revocation_drops_only_that_installation(self) -> None:
|
|
from components.observability import get_telemetry, release_telemetry
|
|
|
|
binding_a = _binding("installation-revoke-a")
|
|
binding_b = _binding("installation-revoke-b")
|
|
self.addCleanup(release_telemetry, binding_a)
|
|
self.addCleanup(release_telemetry, binding_b)
|
|
|
|
await self._record_parse(binding_a, "revoked-a.txt")
|
|
await self._record_parse(binding_b, "kept-b.txt")
|
|
|
|
release_telemetry(binding_a)
|
|
|
|
self.assertEqual(get_telemetry(binding_a).snapshot()["summary"]["total_parses"], 0)
|
|
self.assertEqual(get_telemetry(binding_b).snapshot()["summary"]["total_parses"], 1)
|
|
|
|
async def test_unknown_endpoint_fails(self) -> None:
|
|
from components.observability import release_telemetry
|
|
|
|
binding = _binding("installation-unknown-endpoint")
|
|
self.addCleanup(release_telemetry, binding)
|
|
|
|
response = await self._snapshot(binding, endpoint="/missing", method="GET")
|
|
self.assertIsNotNone(response.error)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|