Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import json | |
| import sys | |
| import unittest | |
| from pathlib import Path | |
| from unittest.mock import AsyncMock, patch | |
| SERVICE_ROOT = Path(__file__).resolve().parents[1] | |
| if str(SERVICE_ROOT) not in sys.path: | |
| sys.path.insert(0, str(SERVICE_ROOT)) | |
| from tabular_agent import run_chat, run_review # noqa: E402 | |
| class PlaybookFlowTests(unittest.IsolatedAsyncioTestCase): | |
| async def test_run_review_builds_canonical_payload_and_normalizes_findings( | |
| self, | |
| ) -> None: | |
| raw_review = json.dumps( | |
| { | |
| "workflow_id": "buyer", | |
| "findings": [ | |
| { | |
| "id": "finding_001", | |
| "risk_level": "high", | |
| "clause_type": "Payment", | |
| "issue": "Payment period is too long.", | |
| "comment": "Reduce the payment period.", | |
| "target_text": "Payment is due in 90 days.", | |
| "start_paragraph_id": "P_0001", | |
| "end_paragraph_id": "P_0001", | |
| "confidence": 0.9, | |
| } | |
| ], | |
| } | |
| ) | |
| mocked_post = AsyncMock( | |
| return_value=( | |
| raw_review, | |
| {"totalTokenCount": 42}, | |
| {"candidates": [{"content": {"parts": [{"text": raw_review}]}}]}, | |
| ) | |
| ) | |
| with patch("tabular_agent._post", mocked_post): | |
| result, usage, _trace_id = await run_review( | |
| review_text="[P_0001] Payment is due in 90 days.", | |
| document_text="Payment is due in 90 days.", | |
| document_name="contract.docx", | |
| playbook_id="buyer", | |
| playbook_name="Buyer Playbook", | |
| playbook_text="Payment must be due within 30 days.", | |
| max_findings=10, | |
| ) | |
| self.assertEqual(result.workflow_id, "buyer") | |
| self.assertEqual(len(result.findings), 1) | |
| self.assertEqual(result.findings[0].issue, "Payment period is too long.") | |
| self.assertEqual(result.model, "TabularAgent") | |
| self.assertEqual(usage, {"totalTokenCount": 42}) | |
| call = mocked_post.await_args | |
| self.assertEqual(call.kwargs["feature_id"], "python-review") | |
| payload_text = call.args[0]["messages"][1]["content"] | |
| self.assertIn("SYSTEM INSTRUCTIONS:", payload_text) | |
| self.assertIn("USER INPUT:", payload_text) | |
| self.assertIn("Buyer Playbook", payload_text) | |
| self.assertIn("Payment must be due within 30 days.", payload_text) | |
| self.assertIn("Payment is due in 90 days.", payload_text) | |
| async def test_run_chat_builds_canonical_payload_and_renders_sources( | |
| self, | |
| ) -> None: | |
| source_url = "https://example.test/payment-terms" | |
| raw_chat = json.dumps( | |
| { | |
| "answer_paragraphs": [ | |
| { | |
| "text": "The clause exceeds the playbook position.", | |
| "citation_urls": [source_url], | |
| } | |
| ], | |
| "sources": [ | |
| { | |
| "title": "Payment Terms Guidance", | |
| "url": source_url, | |
| } | |
| ], | |
| } | |
| ) | |
| mocked_post = AsyncMock( | |
| return_value=( | |
| raw_chat, | |
| {"totalTokenCount": 24}, | |
| {"candidates": [{"content": {"parts": [{"text": raw_chat}]}}]}, | |
| ) | |
| ) | |
| with patch("tabular_agent._post", mocked_post): | |
| text, usage, _trace_id = await run_chat( | |
| message="Does this payment clause comply?", | |
| document_text="Payment is due in 90 days.", | |
| document_name="contract.docx", | |
| playbook_id="buyer", | |
| playbook_name="Buyer Playbook", | |
| role_brief="Protect the buyer.", | |
| ) | |
| self.assertIn("The clause exceeds the playbook position.", text) | |
| self.assertIn( | |
| "[Payment Terms Guidance](https://example.test/payment-terms)", | |
| text, | |
| ) | |
| self.assertIn("Sources:", text) | |
| self.assertEqual(usage, {"totalTokenCount": 24}) | |
| call = mocked_post.await_args | |
| self.assertEqual(call.kwargs["feature_id"], "python-chat") | |
| payload_text = call.args[0]["messages"][1]["content"] | |
| self.assertIn("SYSTEM INSTRUCTIONS:", payload_text) | |
| self.assertIn("USER INPUT:", payload_text) | |
| self.assertIn("Protect the buyer.", payload_text) | |
| self.assertIn("Does this payment clause comply?", payload_text) | |
| self.assertIn("Payment is due in 90 days.", payload_text) | |
| if __name__ == "__main__": | |
| unittest.main() | |