Spaces:
Running
Running
File size: 4,417 Bytes
6853143 | 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 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | #!/usr/bin/env python3
"""
Test script for document analyses integration with lawyer prompts
"""
import asyncio
import sys
sys.path.append('.')
from agent_api import CyberLegalAPI, DocumentAnalysis
import logging
logging.basicConfig(level=logging.INFO)
async def test_lawyer_with_documents():
"""Test lawyer agent with document analyses"""
api = CyberLegalAPI()
# Create sample document analyses
doc_analyses = [
DocumentAnalysis(
file_name="employment_contract.pdf",
summary="Employment contract with termination clause",
actors="John Doe (employee), TechCorp (employer)",
key_details="30-day notice period, 2 months severance, confidentiality clause"
),
DocumentAnalysis(
file_name="nda_agreement.pdf",
summary="Non-disclosure agreement",
actors="TechCorp, ThirdParty Inc",
key_details="5-year duration, penalties for breach, return of materials"
)
]
# Test request with document analyses
from agent_api import ChatRequest, Message
request = ChatRequest(
message="What are the termination obligations according to the employment contract?",
conversationHistory=[],
userType="lawyer",
jurisdiction="Romania",
documentAnalyses=doc_analyses
)
print("Testing lawyer agent with document analyses...")
print("="*60)
try:
response = await api.process_request(request)
print(f"✅ Response generated successfully")
print(f"Processing time: {response.processing_time:.2f}s")
print(f"Response: {response.response[:500]}...")
print("="*60)
return True
except Exception as e:
print(f"❌ Error: {e}")
return False
async def test_lawyer_without_documents():
"""Test lawyer agent without document analyses"""
api = CyberLegalAPI()
from agent_api import ChatRequest
request = ChatRequest(
message="What are the key provisions of GDPR?",
conversationHistory=[],
userType="lawyer",
jurisdiction="Romania",
documentAnalyses=None
)
print("\nTesting lawyer agent without document analyses...")
print("="*60)
try:
response = await api.process_request(request)
print(f"✅ Response generated successfully")
print(f"Processing time: {response.processing_time:.2f}s")
print(f"Response: {response.response[:500]}...")
print("="*60)
return True
except Exception as e:
print(f"❌ Error: {e}")
return False
async def test_client_user():
"""Test client user (should ignore document analyses)"""
api = CyberLegalAPI()
doc_analyses = [
DocumentAnalysis(
file_name="sample.pdf",
summary="Sample document",
actors="Person A, Person B",
key_details="Some details"
)
]
from agent_api import ChatRequest
request = ChatRequest(
message="What is GDPR?",
conversationHistory=[],
userType="client",
jurisdiction="Romania",
documentAnalyses=doc_analyses # Should be ignored for clients
)
print("\nTesting client user (document analyses should be ignored)...")
print("="*60)
try:
response = await api.process_request(request)
print(f"✅ Response generated successfully")
print(f"Processing time: {response.processing_time:.2f}s")
print(f"Response: {response.response[:500]}...")
print("="*60)
return True
except Exception as e:
print(f"❌ Error: {e}")
return False
async def main():
"""Run all tests"""
print("Starting document analyses integration tests...")
print("="*60)
results = []
# Test 1: Lawyer with documents
results.append(await test_lawyer_with_documents())
# Test 2: Lawyer without documents
results.append(await test_lawyer_without_documents())
# Test 3: Client (should ignore documents)
results.append(await test_client_user())
print("\n" + "="*60)
print("Test Summary:")
print(f"Passed: {sum(results)}/{len(results)}")
print("="*60)
return all(results)
if __name__ == "__main__":
success = asyncio.run(main())
sys.exit(0 if success else 1)
|