cyuqei
2335 lines
102 KiB
Python
2335 lines
102 KiB
Python
import json
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import HTTPException
|
|
from app.core.exceptions import ConflictError, ForbiddenError, NotFoundError, ValidationError
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.contracts import get_audit_log_service, get_contract_generation_service, get_tenant_service, require_current_user
|
|
from app.main import app
|
|
from app.services.contract_generation_service import ExportFormat
|
|
|
|
|
|
class StubAuditLogService:
|
|
def __init__(self):
|
|
self.operations = []
|
|
|
|
def record_event(self, **kwargs):
|
|
self.operations.append(kwargs)
|
|
return kwargs
|
|
|
|
|
|
class StubTenantService:
|
|
def __init__(self, blocked_action=None):
|
|
self.blocked_action = blocked_action
|
|
self.calls = []
|
|
|
|
def assert_billing_allows(self, current_user, action, requested_file_size_bytes=0):
|
|
self.calls.append((action, requested_file_size_bytes))
|
|
if action == self.blocked_action:
|
|
raise HTTPException(status_code=402, detail="quota exceeded")
|
|
|
|
|
|
class StubContractGenerationService:
|
|
def __init__(self):
|
|
self.last_generate_options = None
|
|
def _execute_deferred_auto_review(self, contract_id, tenant_id, review_user):
|
|
pass # stub for async auto review background task
|
|
def _auto_review_summary(self, review_id="review-auto-123", status="completed"):
|
|
return {
|
|
"review_id": review_id,
|
|
"review_title": "Procurement Template",
|
|
"status": status,
|
|
"review_as": "party_a",
|
|
"contract_type": "procurement",
|
|
"risk_level": "high",
|
|
"current_stage": "report",
|
|
"current_stage_label": "Report generated",
|
|
"status_detail": "Auto review completed",
|
|
"failure_reason": None,
|
|
"next_action": "Open review report",
|
|
"can_retry": False,
|
|
"review_opinion": "High risk terms found",
|
|
"recommendation": "manual_review",
|
|
"source_file_id": "file-generated-123",
|
|
"source_file_name": "contract.pdf",
|
|
"source_text_length": 12,
|
|
"processing_time_ms": 1200,
|
|
"detail_url": f"/reviews/{review_id}",
|
|
"report_download_url": f"/api/v1/reviews/{review_id}/report/download",
|
|
"report_file_name": f"review-report-{review_id}.pdf",
|
|
"chat_url": f"/chat?review={review_id}",
|
|
}
|
|
|
|
def _record(
|
|
self,
|
|
record_id,
|
|
tenant_id="tenant-demo-001",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
):
|
|
return {
|
|
"id": record_id,
|
|
"template_id": "tpl-123",
|
|
"template_name": "Procurement Template",
|
|
"contract_type": contract_type,
|
|
"document_type_code": document_type_code,
|
|
"generated_content": "Contract with Supplier A",
|
|
"generated_html": "<article>Contract with Supplier A</article>",
|
|
"variable_values": {"supplier_name": "Supplier A"},
|
|
"elements_snapshot": None,
|
|
"missing_fields": [],
|
|
"ai_hints": [],
|
|
"related_legal_articles": [],
|
|
"generated_at": "2026-04-17T00:00:00+00:00",
|
|
"export_format": "pdf",
|
|
"file_name": "contract.pdf",
|
|
"file_size": 128,
|
|
"preview_url": f"/api/v1/contracts/{record_id}/preview",
|
|
"download_url": f"/api/v1/contracts/{record_id}/download",
|
|
"auto_review": self._auto_review_summary(f"review-{record_id}"),
|
|
"lifecycle": {
|
|
"status": "generated",
|
|
"counterparty_name": None,
|
|
"owner_name": None,
|
|
"effective_at": None,
|
|
"expires_at": None,
|
|
"days_to_expiry": None,
|
|
"due_status": "not_set",
|
|
"note": None,
|
|
"updated_by": None,
|
|
"updated_at": None,
|
|
},
|
|
"metadata": {
|
|
"generated_by": generated_by,
|
|
"generated_by_name": generated_by_name,
|
|
"tenant_id": tenant_id,
|
|
"original_template": "Procurement Template",
|
|
"document_type_code": document_type_code,
|
|
"export_options": {"watermark": True},
|
|
"variables_used": 1,
|
|
"asset_file_id": "file-generated-123",
|
|
"missing_field_count": 0,
|
|
"related_legal_article_count": 0,
|
|
"generation_source": None,
|
|
},
|
|
"tenant_id": tenant_id,
|
|
}
|
|
|
|
def list_generated_contracts(self, current_user, page, page_size, contract_type=None, document_type_code=None, lifecycle_status=None, keyword=None):
|
|
records = [
|
|
self._record("contract-123"),
|
|
self._record("contract-456", contract_type="employment", document_type_code="labor_contract"),
|
|
self._record(
|
|
"contract-member-owned",
|
|
tenant_id="tenant-demo-001",
|
|
generated_by="user-member",
|
|
generated_by_name="Member",
|
|
),
|
|
self._record("contract-other-tenant", tenant_id="tenant-other-001"),
|
|
]
|
|
tenant_records = [record for record in records if record["tenant_id"] == current_user["tenant_id"]]
|
|
if current_user.get("role") != "admin":
|
|
tenant_records = [
|
|
record
|
|
for record in tenant_records
|
|
if record["metadata"]["generated_by"] == current_user["id"]
|
|
]
|
|
if contract_type:
|
|
tenant_records = [record for record in tenant_records if record["contract_type"] == contract_type]
|
|
if document_type_code:
|
|
tenant_records = [record for record in tenant_records if record["document_type_code"] == document_type_code]
|
|
if lifecycle_status:
|
|
tenant_records = [record for record in tenant_records if record["lifecycle"]["status"] == lifecycle_status]
|
|
if keyword:
|
|
normalized_keyword = keyword.strip().lower()
|
|
tenant_records = [
|
|
record
|
|
for record in tenant_records
|
|
if normalized_keyword in record["template_name"].lower()
|
|
or normalized_keyword in record["file_name"].lower()
|
|
or normalized_keyword in record["metadata"]["generated_by_name"].lower()
|
|
]
|
|
start = (page - 1) * page_size
|
|
return {
|
|
"items": [{key: value for key, value in item.items() if key != "tenant_id"} for item in tenant_records[start : start + page_size]],
|
|
"total": len(tenant_records),
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
def update_generated_contract_lifecycle(
|
|
self,
|
|
current_user,
|
|
contract_id,
|
|
lifecycle_status,
|
|
counterparty_name=None,
|
|
owner_name=None,
|
|
effective_at=None,
|
|
expires_at=None,
|
|
lifecycle_note=None,
|
|
):
|
|
record = self.get_generated_contract(current_user, contract_id)
|
|
record["lifecycle"] = {
|
|
"status": lifecycle_status,
|
|
"counterparty_name": counterparty_name,
|
|
"owner_name": owner_name,
|
|
"effective_at": effective_at.isoformat() if effective_at else None,
|
|
"expires_at": expires_at.isoformat() if expires_at else None,
|
|
"days_to_expiry": 20,
|
|
"due_status": "due_soon",
|
|
"note": lifecycle_note,
|
|
"updated_by": current_user["id"],
|
|
"updated_at": "2026-05-21T00:00:00+00:00",
|
|
}
|
|
return record
|
|
|
|
def _lifecycle_task(self, contract_id="contract-123", task_id="clt-001", status_value="open"):
|
|
return {
|
|
"id": task_id,
|
|
"tenant_id": "tenant-demo-001",
|
|
"contract_id": contract_id,
|
|
"task_key": "renewal:contract-123",
|
|
"task_type": "renewal_reminder",
|
|
"title": "Renewal reminder",
|
|
"description": "Confirm renewal decision",
|
|
"priority": "high",
|
|
"status": status_value,
|
|
"due_at": "2026-06-01T00:00:00+00:00",
|
|
"due_status": "due_soon",
|
|
"days_to_due": 7,
|
|
"completed_by": "user-admin" if status_value == "completed" else None,
|
|
"completed_at": "2026-05-21T00:00:00+00:00" if status_value == "completed" else None,
|
|
"evidence_file_id": "file-evidence-001" if status_value == "completed" else None,
|
|
"evidence_file": {
|
|
"id": "file-evidence-001",
|
|
"file_name": "termination-proof.pdf",
|
|
"file_type": "contract_lifecycle_evidence",
|
|
"mime_type": "application/pdf",
|
|
"file_size": 128,
|
|
"download_url": "/api/v1/files/file-evidence-001/download",
|
|
"created_at": "2026-05-21T00:00:00+00:00",
|
|
} if status_value == "completed" else None,
|
|
"evidence_note": "Archive proof" if status_value == "completed" else None,
|
|
"metadata": {"source": "manual"},
|
|
"created_by": "user-admin",
|
|
"created_at": "2026-05-21T00:00:00+00:00",
|
|
"updated_at": "2026-05-21T00:00:00+00:00",
|
|
}
|
|
|
|
def list_contract_lifecycle_tasks(self, current_user, contract_id, status_value=None, task_type=None):
|
|
self.get_generated_contract(current_user, contract_id)
|
|
task = self._lifecycle_task(contract_id=contract_id)
|
|
if status_value and task["status"] != status_value:
|
|
items = []
|
|
elif task_type and task["task_type"] != task_type:
|
|
items = []
|
|
else:
|
|
items = [task]
|
|
return {
|
|
"contract_id": contract_id,
|
|
"items": items,
|
|
"total": len(items),
|
|
"summary": {"open": len(items), "in_progress": 0, "completed": 0, "skipped": 0, "archived": 0, "overdue": 0, "due_soon": len(items), "with_evidence": 0},
|
|
}
|
|
|
|
def sync_contract_lifecycle_tasks(self, current_user, contract_id):
|
|
result = self.list_contract_lifecycle_tasks(current_user, contract_id)
|
|
result["synced"] = True
|
|
result["reason"] = "renewal_reminder_synced"
|
|
return result
|
|
|
|
def create_contract_lifecycle_task(
|
|
self,
|
|
current_user,
|
|
contract_id,
|
|
task_type,
|
|
title,
|
|
description=None,
|
|
priority="medium",
|
|
status_value="open",
|
|
due_at=None,
|
|
evidence_file_id=None,
|
|
evidence_note=None,
|
|
):
|
|
self.get_generated_contract(current_user, contract_id)
|
|
task = self._lifecycle_task(contract_id=contract_id, task_id="clt-created", status_value=status_value)
|
|
task["task_type"] = task_type
|
|
task["title"] = title
|
|
task["description"] = description
|
|
task["priority"] = priority
|
|
task["due_at"] = due_at.isoformat() if due_at else None
|
|
task["evidence_file_id"] = evidence_file_id
|
|
task["evidence_note"] = evidence_note
|
|
return task
|
|
|
|
def update_contract_lifecycle_task(
|
|
self,
|
|
current_user,
|
|
contract_id,
|
|
task_id,
|
|
task_type=None,
|
|
title=None,
|
|
description=None,
|
|
priority=None,
|
|
status_value=None,
|
|
due_at=None,
|
|
evidence_file_id=None,
|
|
evidence_note=None,
|
|
):
|
|
self.get_generated_contract(current_user, contract_id)
|
|
task = self._lifecycle_task(contract_id=contract_id, task_id=task_id, status_value=status_value or "open")
|
|
if task_type is not None:
|
|
task["task_type"] = task_type
|
|
if title is not None:
|
|
task["title"] = title
|
|
if description is not None:
|
|
task["description"] = description
|
|
if priority is not None:
|
|
task["priority"] = priority
|
|
if due_at is not None:
|
|
task["due_at"] = due_at.isoformat()
|
|
if evidence_file_id is not None:
|
|
task["evidence_file_id"] = evidence_file_id
|
|
if evidence_note is not None:
|
|
task["evidence_note"] = evidence_note
|
|
return task
|
|
|
|
def list_available_templates(self, current_user, contract_type, document_type_code=None, template_category=None, industry_category=None):
|
|
return [
|
|
{
|
|
"id": "tpl-123",
|
|
"name": "Procurement Template",
|
|
"contract_type": contract_type or "procurement",
|
|
"document_type_code": document_type_code or "procurement_goods",
|
|
"description": "Standard procurement contract",
|
|
"variables": [{"name": "supplier_name", "label": "Supplier", "required": True}],
|
|
"preview_excerpt": "Buyer purchases goods from supplier.",
|
|
}
|
|
]
|
|
|
|
def preview_contract(self, current_user, template_id, variable_values):
|
|
if template_id == "tpl-missing":
|
|
raise HTTPException(status_code=404, detail="template not found")
|
|
if "甲方名称" in variable_values:
|
|
return {
|
|
"preview_content": f"甲方：{variable_values.get('甲方名称', '')}\n乙方：{variable_values.get('乙方名称', '')}",
|
|
"template_name": "Property Service Template",
|
|
"variable_summary": [
|
|
{
|
|
"name": "甲方名称",
|
|
"label": "甲方名称",
|
|
"required": True,
|
|
"provided": "甲方名称" in variable_values,
|
|
"value_preview": variable_values.get("甲方名称", ""),
|
|
"is_empty": not bool(variable_values.get("甲方名称", "").strip()),
|
|
},
|
|
{
|
|
"name": "乙方名称",
|
|
"label": "乙方名称",
|
|
"required": True,
|
|
"provided": "乙方名称" in variable_values,
|
|
"value_preview": variable_values.get("乙方名称", ""),
|
|
"is_empty": not bool(variable_values.get("乙方名称", "").strip()),
|
|
},
|
|
],
|
|
}
|
|
return {
|
|
"preview_content": f"Contract with {variable_values.get('supplier_name', '')}",
|
|
"template_name": "Procurement Template",
|
|
"variable_summary": [
|
|
{
|
|
"name": "supplier_name",
|
|
"label": "Supplier",
|
|
"required": True,
|
|
"provided": "supplier_name" in variable_values,
|
|
"value_preview": variable_values.get("supplier_name", ""),
|
|
"is_empty": not bool(variable_values.get("supplier_name", "").strip()),
|
|
}
|
|
],
|
|
}
|
|
|
|
def get_generated_contract(self, current_user, contract_id):
|
|
records = [
|
|
item
|
|
for item in self.list_generated_contracts(current_user, page=1, page_size=50)["items"]
|
|
if item["id"] == contract_id
|
|
]
|
|
if not records:
|
|
raise HTTPException(status_code=404, detail="contract not found")
|
|
return records[0]
|
|
|
|
def list_generated_contract_versions(self, current_user, contract_id):
|
|
self.get_generated_contract(current_user, contract_id)
|
|
return {
|
|
"contract_id": contract_id,
|
|
"base_version": {"version_number": 1, "version_label": "v1", "content": "Contract with Supplier A"},
|
|
"items": [
|
|
{
|
|
"id": "gcv-001",
|
|
"contract_id": contract_id,
|
|
"version_number": 2,
|
|
"version_label": "v2",
|
|
"content": "Contract with Supplier A. Payment deadline added.",
|
|
"locate_status": "applied",
|
|
"detail_url": f"/contracts/{contract_id}?version=gcv-001",
|
|
}
|
|
],
|
|
"total": 1,
|
|
}
|
|
|
|
def get_generated_contract_version(self, current_user, contract_id, version_id):
|
|
versions = self.list_generated_contract_versions(current_user, contract_id)
|
|
for item in versions["items"]:
|
|
if item["id"] == version_id:
|
|
return item
|
|
raise HTTPException(status_code=404, detail="contract version not found")
|
|
|
|
def diff_generated_contract_versions(self, current_user, contract_id, from_version_id=None, to_version_id=None):
|
|
self.get_generated_contract(current_user, contract_id)
|
|
return {
|
|
"contract_id": contract_id,
|
|
"from_version": {"id": "base", "version_number": 1, "version_label": "v1"},
|
|
"to_version": {"id": to_version_id or "gcv-001", "version_number": 2, "version_label": "v2"},
|
|
"summary": {
|
|
"added": 1,
|
|
"removed": 0,
|
|
"modified": 1,
|
|
"unchanged": 1,
|
|
"change_count": 2,
|
|
"from_paragraph_count": 2,
|
|
"to_paragraph_count": 3,
|
|
},
|
|
"items": [
|
|
{"change_type": "modified", "old_paragraph_index": 1, "new_paragraph_index": 1, "old_text": "Old", "new_text": "New"},
|
|
{"change_type": "added", "old_paragraph_index": None, "new_paragraph_index": 3, "old_text": None, "new_text": "Payment deadline added."},
|
|
],
|
|
}
|
|
|
|
def export_generated_contract_version(self, current_user, contract_id, version_id=None, export_format=ExportFormat.DOCX):
|
|
self.get_generated_contract(current_user, contract_id)
|
|
return {
|
|
"id": contract_id,
|
|
"contract_id": contract_id,
|
|
"version": {"id": version_id or "latest", "version_number": 2, "version_label": "v2"},
|
|
"file_name": "contract-v2.docx",
|
|
"mime_type": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
"content": b"docx-bytes",
|
|
"file_size": 10,
|
|
"export_format": export_format.value,
|
|
}
|
|
|
|
def create_generated_contract_version(
|
|
self,
|
|
current_user,
|
|
contract_id,
|
|
content,
|
|
change_type="editor_save",
|
|
change_summary=None,
|
|
source_version_id=None,
|
|
original_text=None,
|
|
replacement_text=None,
|
|
locate_status="applied",
|
|
):
|
|
self.get_generated_contract(current_user, contract_id)
|
|
return {
|
|
"id": "gcv-editor-001",
|
|
"tenant_id": current_user["tenant_id"],
|
|
"contract_id": contract_id,
|
|
"review_id": None,
|
|
"review_risk_action_id": None,
|
|
"version_number": 3,
|
|
"version_label": "v3",
|
|
"source_version_id": source_version_id or "gcv-001",
|
|
"change_type": change_type,
|
|
"change_summary": change_summary or "Editor save",
|
|
"original_text": original_text,
|
|
"replacement_text": replacement_text,
|
|
"content": content,
|
|
"locate_status": locate_status,
|
|
"created_by": current_user["id"],
|
|
"created_at": "2026-05-20T00:00:00+00:00",
|
|
"detail_url": f"/contracts/{contract_id}?version=gcv-editor-001",
|
|
}
|
|
|
|
def generate_contract(
|
|
self,
|
|
current_user,
|
|
template_id,
|
|
variable_values,
|
|
export_format,
|
|
options,
|
|
document_type_code=None,
|
|
elements_snapshot=None,
|
|
element_schema_id=None,
|
|
element_schema_version_label=None,
|
|
fill_missing_variables=False,
|
|
):
|
|
self.last_generate_options = options
|
|
if template_id == "tpl-missing":
|
|
raise HTTPException(status_code=404, detail="template not found")
|
|
if "甲方名称" in variable_values:
|
|
content = f"甲方：{variable_values.get('甲方名称', '')}\n乙方：{variable_values.get('乙方名称', '')}"
|
|
contract_type = "property_service"
|
|
document_type_code = document_type_code or "property_service"
|
|
template_name = "Property Service Template"
|
|
else:
|
|
content = f"Contract with {variable_values.get('supplier_name', '')}"
|
|
contract_type = "procurement"
|
|
document_type_code = document_type_code or "procurement_goods"
|
|
template_name = "Procurement Template"
|
|
metadata_generation_source = {
|
|
"generation_mode": "ai_enhanced_generation" if options.get("ai_enhanced_generation") else "template_render",
|
|
"ai_enhancement_applied": bool(options.get("ai_enhanced_generation") and not options.get("simulate_ai_fallback")),
|
|
"fallback_used": bool(options.get("simulate_ai_fallback")),
|
|
"fallback_reason": "simulated shared ai failure" if options.get("simulate_ai_fallback") else None,
|
|
"ai_provider": "shared-ai-platform" if options.get("ai_enhanced_generation") else None,
|
|
"ai_model": "mock-model" if options.get("ai_enhanced_generation") else None,
|
|
"prompt_key": "contract_generation_default" if options.get("ai_enhanced_generation") else None,
|
|
"prompt_version": "contract_generation_default.v1.0" if options.get("ai_enhanced_generation") else None,
|
|
}
|
|
return {
|
|
"id": "contract-123",
|
|
"template_id": template_id,
|
|
"template_name": template_name,
|
|
"contract_type": contract_type,
|
|
"document_type_code": document_type_code,
|
|
"generated_content": content,
|
|
"generated_html": "<article>Contract</article>",
|
|
"variable_values": variable_values,
|
|
"elements_snapshot": elements_snapshot,
|
|
"missing_fields": [],
|
|
"ai_hints": [],
|
|
"related_legal_articles": [],
|
|
"generated_at": "2026-04-17T00:00:00+00:00",
|
|
"export_format": export_format.value,
|
|
"file_name": f"contract.{export_format.value}",
|
|
"file_size": 128,
|
|
"file_content_base64": "ZmlsZQ==",
|
|
"preview_url": "/api/v1/contracts/contract-123/preview",
|
|
"download_url": "/api/v1/contracts/contract-123/download",
|
|
"auto_review": self._auto_review_summary("review-auto-123"),
|
|
"lifecycle": {
|
|
"status": "generated",
|
|
"counterparty_name": None,
|
|
"owner_name": None,
|
|
"effective_at": None,
|
|
"expires_at": None,
|
|
"days_to_expiry": None,
|
|
"due_status": "not_set",
|
|
"note": None,
|
|
"updated_by": None,
|
|
"updated_at": None,
|
|
},
|
|
"metadata": {
|
|
"generated_by": current_user["id"],
|
|
"generated_by_name": current_user.get("display_name", "Admin"),
|
|
"tenant_id": current_user["tenant_id"],
|
|
"original_template": "Procurement Template",
|
|
"document_type_code": document_type_code or "procurement_goods",
|
|
"elements_snapshot": elements_snapshot,
|
|
"element_schema_id": element_schema_id,
|
|
"element_schema_version_label": element_schema_version_label,
|
|
"export_options": options,
|
|
"variables_used": len(variable_values),
|
|
"missing_field_count": 0,
|
|
"related_legal_article_count": 0,
|
|
"generation_source": metadata_generation_source,
|
|
},
|
|
}
|
|
|
|
def generate_contract_from_elements(
|
|
self,
|
|
current_user,
|
|
document_type_code,
|
|
elements,
|
|
export_format,
|
|
options,
|
|
template_id=None,
|
|
variable_values=None,
|
|
):
|
|
result = self.generate_contract(
|
|
current_user=current_user,
|
|
template_id=template_id or "tpl-auto-selected",
|
|
variable_values={"supplier_name": (elements.get("party_b") or {}).get("name", "seller"), **(variable_values or {})},
|
|
export_format=export_format,
|
|
options=options,
|
|
document_type_code=document_type_code,
|
|
elements_snapshot=elements,
|
|
element_schema_id="ldes-auto-v1",
|
|
element_schema_version_label="v1.0",
|
|
fill_missing_variables=True,
|
|
)
|
|
result["missing_fields"] = [{"field_key": "payment.terms", "label": "Payment terms", "placeholder": "please fill"}]
|
|
result["ai_hints"] = [{"code": "missing_required_fields", "level": "warning", "message": "Fields still need input"}]
|
|
result["related_legal_articles"] = [{"source_id": "lart-civil-585", "title": "Civil Code Article 585"}]
|
|
result["metadata"]["missing_field_count"] = 1
|
|
result["metadata"]["related_legal_article_count"] = 1
|
|
result["metadata"]["generation_source"] = {"mode": "structured_elements", "template_selection": "auto"}
|
|
return result
|
|
|
|
def get_generated_contract_preview(self, current_user, contract_id):
|
|
if contract_id == "contract-missing":
|
|
raise HTTPException(status_code=404, detail="contract not found")
|
|
return {
|
|
"id": contract_id,
|
|
"template_name": "Procurement Template",
|
|
"contract_type": "procurement",
|
|
"preview_content": "Contract with Supplier A",
|
|
"export_format": "pdf",
|
|
"file_name": "contract.pdf",
|
|
"generated_at": "2026-04-17T00:00:00+00:00",
|
|
"preview_url": f"/api/v1/contracts/{contract_id}/preview",
|
|
"download_url": f"/api/v1/contracts/{contract_id}/download",
|
|
}
|
|
|
|
def get_generated_contract_download(self, current_user, contract_id):
|
|
if contract_id == "contract-missing":
|
|
raise HTTPException(status_code=404, detail="contract not found")
|
|
return {
|
|
"id": contract_id,
|
|
"file_name": "contract.pdf",
|
|
"mime_type": "application/pdf",
|
|
"content": b"pdf-bytes",
|
|
}
|
|
|
|
|
|
class ContractsApiTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.audit_log_service = StubAuditLogService()
|
|
self.tenant_service = StubTenantService()
|
|
app.dependency_overrides[get_contract_generation_service] = lambda: StubContractGenerationService()
|
|
app.dependency_overrides[get_audit_log_service] = lambda: self.audit_log_service
|
|
app.dependency_overrides[get_tenant_service] = lambda: self.tenant_service
|
|
self.client = TestClient(app)
|
|
|
|
def tearDown(self):
|
|
app.dependency_overrides.clear()
|
|
|
|
def _login(self, tenant_id="tenant-demo-001"):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": tenant_id,
|
|
"display_name": "Admin",
|
|
"role": "admin",
|
|
}
|
|
|
|
def _login_member(self, tenant_id="tenant-demo-001"):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-member",
|
|
"tenant_id": tenant_id,
|
|
"display_name": "Member",
|
|
"role": "member",
|
|
}
|
|
|
|
def test_user_can_list_contract_templates(self):
|
|
self._login()
|
|
response = self.client.get("/api/v1/contracts/templates", params={"contract_type": "procurement"})
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"][0]["document_type_code"], "procurement_goods")
|
|
|
|
def test_user_can_list_generated_contracts_with_filters(self):
|
|
self._login()
|
|
response = self.client.get("/api/v1/contracts", params={"document_type_code": "labor_contract"})
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["total"], 1)
|
|
self.assertEqual(body["data"]["items"][0]["document_type_code"], "labor_contract")
|
|
self.assertEqual(body["data"]["items"][0]["lifecycle"]["status"], "generated")
|
|
|
|
def test_member_generated_contract_list_only_returns_owned_records(self):
|
|
self._login_member()
|
|
response = self.client.get("/api/v1/contracts")
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["total"], 1)
|
|
self.assertEqual(body["data"]["items"][0]["id"], "contract-member-owned")
|
|
|
|
def test_user_can_update_contract_lifecycle_and_write_audit_log(self):
|
|
self._login()
|
|
response = self.client.patch(
|
|
"/api/v1/contracts/contract-123/lifecycle",
|
|
json={
|
|
"lifecycle_status": "effective",
|
|
"counterparty_name": "Supplier A",
|
|
"owner_name": "Legal Ops",
|
|
"effective_at": "2026-05-21T00:00:00+08:00",
|
|
"expires_at": "2026-06-10T00:00:00+08:00",
|
|
"lifecycle_note": "已签署，进入履约跟踪。",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["lifecycle"]["status"], "effective")
|
|
self.assertEqual(body["data"]["lifecycle"]["counterparty_name"], "Supplier A")
|
|
self.assertEqual(body["data"]["lifecycle"]["due_status"], "due_soon")
|
|
self.assertEqual(self.audit_log_service.operations[0]["action"], "contract_lifecycle_update")
|
|
self.assertEqual(self.audit_log_service.operations[0]["detail_json"]["counterparty_name"], "Supplier A")
|
|
|
|
def test_user_can_manage_contract_lifecycle_tasks_and_write_audit_logs(self):
|
|
self._login()
|
|
listed = self.client.get("/api/v1/contracts/contract-123/lifecycle/tasks")
|
|
self.assertEqual(listed.status_code, 200)
|
|
self.assertEqual(listed.json()["data"]["items"][0]["task_type"], "renewal_reminder")
|
|
|
|
synced = self.client.post("/api/v1/contracts/contract-123/lifecycle/tasks/sync")
|
|
self.assertEqual(synced.status_code, 200)
|
|
self.assertTrue(synced.json()["data"]["synced"])
|
|
self.assertEqual(self.audit_log_service.operations[-1]["action"], "contract_lifecycle_task_sync")
|
|
|
|
created = self.client.post(
|
|
"/api/v1/contracts/contract-123/lifecycle/tasks",
|
|
json={
|
|
"task_type": "performance_milestone",
|
|
"title": "验收节点",
|
|
"description": "确认验收材料",
|
|
"priority": "high",
|
|
"status_value": "open",
|
|
"due_at": "2026-06-01T00:00:00+08:00",
|
|
},
|
|
)
|
|
self.assertEqual(created.status_code, 200)
|
|
self.assertEqual(created.json()["data"]["task_type"], "performance_milestone")
|
|
self.assertEqual(self.audit_log_service.operations[-1]["action"], "contract_lifecycle_task_create")
|
|
|
|
updated = self.client.put(
|
|
"/api/v1/contracts/contract-123/lifecycle/tasks/clt-created",
|
|
json={
|
|
"status_value": "completed",
|
|
"evidence_file_id": "file-evidence-001",
|
|
"evidence_note": "终止归档证据已绑定",
|
|
},
|
|
)
|
|
self.assertEqual(updated.status_code, 200)
|
|
self.assertEqual(updated.json()["data"]["status"], "completed")
|
|
self.assertEqual(updated.json()["data"]["evidence_file_id"], "file-evidence-001")
|
|
self.assertEqual(self.audit_log_service.operations[-1]["action"], "contract_lifecycle_task_update")
|
|
|
|
def test_generated_contracts_are_tenant_isolated(self):
|
|
self._login("tenant-other-001")
|
|
response = self.client.get("/api/v1/contracts")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"]["items"][0]["id"], "contract-other-tenant")
|
|
|
|
def test_user_can_get_generated_contract_detail(self):
|
|
self._login()
|
|
response = self.client.get("/api/v1/contracts/contract-123")
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["id"], "contract-123")
|
|
self.assertEqual(body["data"]["metadata"]["document_type_code"], "procurement_goods")
|
|
self.assertIn("generated_content", body["data"])
|
|
|
|
def test_user_can_list_and_get_generated_contract_versions(self):
|
|
self._login()
|
|
response = self.client.get("/api/v1/contracts/contract-123/versions")
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["base_version"]["version_label"], "v1")
|
|
self.assertEqual(body["data"]["items"][0]["version_label"], "v2")
|
|
|
|
detail = self.client.get("/api/v1/contracts/contract-123/versions/gcv-001")
|
|
self.assertEqual(detail.status_code, 200)
|
|
self.assertEqual(detail.json()["data"]["id"], "gcv-001")
|
|
|
|
def test_user_can_diff_and_download_generated_contract_version(self):
|
|
self._login()
|
|
diff_response = self.client.get("/api/v1/contracts/contract-123/versions/diff")
|
|
self.assertEqual(diff_response.status_code, 200)
|
|
diff_body = diff_response.json()["data"]
|
|
self.assertEqual(diff_body["summary"]["change_count"], 2)
|
|
self.assertEqual(diff_body["items"][0]["change_type"], "modified")
|
|
self.assertEqual(self.audit_log_service.operations[-1]["action"], "contract_version_diff")
|
|
|
|
download_response = self.client.get("/api/v1/contracts/contract-123/versions/gcv-001/download")
|
|
self.assertEqual(download_response.status_code, 200)
|
|
self.assertEqual(
|
|
download_response.headers["content-type"],
|
|
"application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
)
|
|
self.assertEqual(download_response.content, b"docx-bytes")
|
|
self.assertEqual(self.audit_log_service.operations[-1]["action"], "download_contract_version")
|
|
|
|
def test_user_can_create_generated_contract_version_and_write_audit_log(self):
|
|
self._login()
|
|
response = self.client.post(
|
|
"/api/v1/contracts/contract-123/versions",
|
|
json={
|
|
"content": "Contract with Supplier A. Acceptance clause added.",
|
|
"change_type": "clause_recommendation",
|
|
"change_summary": "Apply acceptance clause",
|
|
"original_text": "",
|
|
"replacement_text": "Acceptance clause added.",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["version_label"], "v3")
|
|
self.assertEqual(body["data"]["change_type"], "clause_recommendation")
|
|
self.assertEqual(self.audit_log_service.operations[0]["action"], "contract_version_create")
|
|
self.assertEqual(self.audit_log_service.operations[0]["detail_json"]["contract_id"], "contract-123")
|
|
|
|
def test_user_can_generate_contract_and_write_audit_log(self):
|
|
self._login()
|
|
response = self.client.post(
|
|
"/api/v1/contracts/generate",
|
|
data={
|
|
"template_id": "tpl-123",
|
|
"variable_values_json": '{"supplier_name":"Supplier A"}',
|
|
"export_format": "pdf",
|
|
"options_json": '{"watermark":true}',
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["export_format"], "pdf")
|
|
self.assertEqual(body["data"]["metadata"]["export_options"], {"watermark": True})
|
|
self.assertEqual([call[0] for call in self.tenant_service.calls], ["generation"])
|
|
self.assertEqual(self.audit_log_service.operations[0]["action"], "contract_generation")
|
|
|
|
def test_user_can_generate_contract_with_chinese_variable_keys(self):
|
|
self._login()
|
|
payload = {
|
|
"甲方名称": "甲方物业公司",
|
|
"乙方名称": "乙方服务公司",
|
|
}
|
|
response = self.client.post(
|
|
"/api/v1/contracts/generate",
|
|
data={
|
|
"template_id": "tpl-123",
|
|
"variable_values_json": json.dumps(payload, ensure_ascii=False),
|
|
"export_format": "docx",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["variable_values"]["甲方名称"], "甲方物业公司")
|
|
self.assertEqual(body["data"]["variable_values"]["乙方名称"], "乙方服务公司")
|
|
|
|
def test_user_can_generate_contract_from_elements_with_auto_template(self):
|
|
self._login()
|
|
response = self.client.post(
|
|
"/api/v1/contracts/generate",
|
|
data={
|
|
"document_type_code": "procurement_goods",
|
|
"elements_json": '{"party_a":{"name":"buyer"},"party_b":{"name":"seller"},"goods":{"name":"server"}}',
|
|
"export_format": "html",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["template_id"], "tpl-auto-selected")
|
|
self.assertEqual(body["data"]["missing_fields"][0]["field_key"], "payment.terms")
|
|
self.assertEqual(body["data"]["related_legal_articles"][0]["source_id"], "lart-civil-585")
|
|
self.assertEqual(self.audit_log_service.operations[0]["detail_json"]["missing_field_count"], 1)
|
|
|
|
def test_generate_contract_handles_review_billing_blocked(self):
|
|
# auto_review 已改为异步队列，不阻塞生成流程
|
|
self._login()
|
|
app.dependency_overrides[get_tenant_service] = lambda: StubTenantService(blocked_action="review")
|
|
response = self.client.post(
|
|
"/api/v1/contracts/generate",
|
|
data={
|
|
"template_id": "tpl-123",
|
|
"variable_values_json": '{"supplier_name":"Supplier A"}',
|
|
"export_format": "pdf",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()["data"]
|
|
self.assertIsNotNone(body["id"])
|
|
|
|
def test_generate_contract_returns_billing_block_message(self):
|
|
self._login()
|
|
app.dependency_overrides[get_tenant_service] = lambda: StubTenantService(blocked_action="generation")
|
|
response = self.client.post(
|
|
"/api/v1/contracts/generate",
|
|
data={
|
|
"template_id": "tpl-123",
|
|
"variable_values_json": '{"supplier_name":"Supplier A"}',
|
|
"export_format": "pdf",
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 402)
|
|
|
|
def test_user_can_get_preview_download_and_error_responses(self):
|
|
self._login()
|
|
preview = self.client.post(
|
|
"/api/v1/contracts/preview",
|
|
data={"template_id": "tpl-123", "variable_values_json": '{"supplier_name":"Supplier A"}'},
|
|
)
|
|
self.assertEqual(preview.status_code, 200)
|
|
self.assertIn("Supplier A", preview.json()["data"]["preview_content"])
|
|
|
|
chinese_preview = self.client.post(
|
|
"/api/v1/contracts/preview",
|
|
data={
|
|
"template_id": "tpl-123",
|
|
"variable_values_json": json.dumps(
|
|
{"甲方名称": "甲方物业公司", "乙方名称": "乙方服务公司"},
|
|
ensure_ascii=False,
|
|
),
|
|
},
|
|
)
|
|
self.assertEqual(chinese_preview.status_code, 200)
|
|
self.assertEqual(
|
|
chinese_preview.json()["data"]["variable_summary"][0]["name"],
|
|
"甲方名称",
|
|
)
|
|
|
|
download = self.client.get("/api/v1/contracts/contract-123/download")
|
|
self.assertEqual(download.status_code, 200)
|
|
self.assertEqual(download.content, b"pdf-bytes")
|
|
self.assertEqual(self.audit_log_service.operations[-1]["action"], "download_contract")
|
|
|
|
invalid_json = self.client.post(
|
|
"/api/v1/contracts/preview",
|
|
data={"template_id": "tpl-123", "variable_values_json": "{bad json}"},
|
|
)
|
|
self.assertEqual(invalid_json.status_code, 400)
|
|
|
|
invalid_format = self.client.post(
|
|
"/api/v1/contracts/generate",
|
|
data={"template_id": "tpl-123", "variable_values_json": '{"supplier_name":"Supplier A"}', "export_format": "xlsx"},
|
|
)
|
|
self.assertEqual(invalid_format.status_code, 400)
|
|
|
|
missing_template = self.client.post(
|
|
"/api/v1/contracts/generate",
|
|
data={"template_id": "tpl-missing", "variable_values_json": '{"supplier_name":"Supplier A"}'},
|
|
)
|
|
self.assertEqual(missing_template.status_code, 404)
|
|
|
|
|
|
class ContractGenerationServiceTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.db import Base
|
|
from app.models.contract_lifecycle_task import ContractLifecycleTask
|
|
from app.models.contract_approval_step import ContractApprovalStep
|
|
from app.models.contract_template import ContractTemplate
|
|
from app.models.file import File
|
|
from app.models.generated_contract import GeneratedContract
|
|
from app.models.generated_contract_version import GeneratedContractVersion
|
|
from app.models.legal_article import LegalArticle
|
|
from app.models.legal_document_element_schema import LegalDocumentElementSchema
|
|
from app.models.legal_document_type import LegalDocumentType
|
|
from app.models.user import User
|
|
from app.services.contract_generation_service import ContractGenerationService
|
|
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(bind=self.engine)
|
|
self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)
|
|
self.db = self.SessionLocal()
|
|
self.ContractTemplate = ContractTemplate
|
|
self.ContractLifecycleTask = ContractLifecycleTask
|
|
self.ContractApprovalStep = ContractApprovalStep
|
|
self.File = File
|
|
self.GeneratedContract = GeneratedContract
|
|
self.GeneratedContractVersion = GeneratedContractVersion
|
|
self.LegalArticle = LegalArticle
|
|
self.LegalDocumentType = LegalDocumentType
|
|
self.LegalDocumentElementSchema = LegalDocumentElementSchema
|
|
self.User = User
|
|
self.service = ContractGenerationService(self.db)
|
|
self.service.file_service = type(
|
|
"StubFileService",
|
|
(),
|
|
{
|
|
"create_file_record": lambda _self, **kwargs: {
|
|
"id": "file-generated-123",
|
|
"file_name": kwargs["file_name"],
|
|
"mime_type": kwargs["mime_type"],
|
|
"file_size": len(kwargs["content"]),
|
|
"file_type": kwargs["file_type"],
|
|
}
|
|
},
|
|
)()
|
|
|
|
class StubReviewService:
|
|
def __init__(self):
|
|
self.create_calls = []
|
|
|
|
def create_review_task(self, **kwargs):
|
|
self.create_calls.append(kwargs)
|
|
return {"id": "review-auto-generated", "status": "completed"}
|
|
|
|
def get_review_detail(self, current_user, review_id):
|
|
return {
|
|
"id": review_id,
|
|
"contract_type": "procurement",
|
|
"review_as": "party_a",
|
|
"status": "completed",
|
|
"current_stage": "report",
|
|
"current_stage_label": "Report generated",
|
|
"status_detail": "done",
|
|
"workflow_state": {"risk_level": "low"},
|
|
"ai_trace": {"status": "completed"},
|
|
"source_document": {"file_name": "contract.docx"},
|
|
}
|
|
|
|
def get_review_result(self, current_user, review_id):
|
|
return {
|
|
"basic_info": {"contract_type": "procurement"},
|
|
"risk_summary": {"overall": "low"},
|
|
"risk_items": [],
|
|
"review_opinion": "ok",
|
|
"recommendation": "approved",
|
|
}
|
|
|
|
def get_review_report(self, current_user, review_id):
|
|
return {"file_name": f"review-report-{review_id}.pdf", "download_url": f"/api/v1/reviews/{review_id}/report/download"}
|
|
|
|
self.review_service = StubReviewService()
|
|
self.service.review_service = self.review_service
|
|
self.current_user = {"id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "Admin", "role": "admin"}
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
self.engine.dispose()
|
|
|
|
def test_generation_prompt_preferred_keys_route_to_specialized_defaults(self):
|
|
cases = [
|
|
("technical_service", "technical-service", "technical_service_default"),
|
|
("software_development", "software-development", "software_development_default"),
|
|
("lease_house", "lease-house", "lease_house_default"),
|
|
("property_service", "property-service", "property_service_default"),
|
|
("equity_transfer", "equity-transfer", "equity_transfer_default"),
|
|
("investment_agreement", "investment", "investment_agreement_default"),
|
|
("intercompany_loan_agreement", "intercompany-loan", "intercompany_loan_agreement_default"),
|
|
("valuation_adjustment_agreement", "valuation-adjustment", "valuation_adjustment_agreement_default"),
|
|
("shareholder_resolution", "shareholder-resolution", "shareholder_resolution_default"),
|
|
("power_of_attorney", "power-of-attorney", "power_of_attorney_default"),
|
|
("legal_opinion", "legal-opinion", "legal_opinion_default"),
|
|
("cooperation_framework", "cooperation-framework", "cooperation_framework_default"),
|
|
("demand_letter", "demand-letter", "demand_letter_default"),
|
|
("rescission_notice", "rescission-notice", "rescission_notice_default"),
|
|
("sales_goods", "sales", "sales_default"),
|
|
("commission_service", "commission-service", "commission_service_default"),
|
|
("agency_distribution_agreement", "agency-distribution", "agency_distribution_default"),
|
|
("franchise_agreement", "franchise", "franchise_default"),
|
|
("software_saas", "software-saas", "software_saas_default"),
|
|
("labor_service_agreement", "labor-service", "labor_service_default"),
|
|
]
|
|
|
|
for document_type_code, contract_type, expected_key in cases:
|
|
with self.subTest(document_type_code=document_type_code, contract_type=contract_type):
|
|
preferred_keys = self.service._build_generation_prompt_preferred_keys(
|
|
document_type_code,
|
|
contract_type,
|
|
)
|
|
self.assertEqual(preferred_keys[0], expected_key)
|
|
self.assertIn("contract_generation_default", preferred_keys)
|
|
self.assertEqual(len(preferred_keys), len(set(preferred_keys)))
|
|
|
|
def test_ai_enhanced_generation_uses_shared_ai_content(self):
|
|
now = datetime.now(timezone.utc)
|
|
template = self.ContractTemplate(
|
|
id="tpl-ai-success",
|
|
tenant_id="tenant-demo-001",
|
|
name="AI Template",
|
|
contract_type="procurement",
|
|
description="AI enhanced generation",
|
|
content="Contract with {{supplier_name}}",
|
|
variables_json=[{"name": "supplier_name", "label": "Supplier", "required": True}],
|
|
enabled=True,
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.db.add(template)
|
|
self.db.commit()
|
|
|
|
class FakeSharedAIClient:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def enhance_contract_generation(self, current_user, *, prompt_runtime, prompt, draft_content, metadata=None):
|
|
self.calls.append({
|
|
"current_user": current_user,
|
|
"prompt_runtime": prompt_runtime,
|
|
"prompt": prompt,
|
|
"draft_content": draft_content,
|
|
"metadata": metadata,
|
|
})
|
|
return {
|
|
"content": "AI enhanced contract for Supplier A",
|
|
"provider": "shared-ai-platform",
|
|
"model": "test-model",
|
|
"prompt_key": "contract_generation_default",
|
|
"prompt_version": "contract_generation_default.v1.0",
|
|
}
|
|
|
|
fake_client = FakeSharedAIClient()
|
|
self.service.shared_ai_platform_client = fake_client
|
|
|
|
result = self.service.generate_contract(
|
|
current_user=self.current_user,
|
|
template_id="tpl-ai-success",
|
|
variable_values={"supplier_name": "Supplier A"},
|
|
options={"ai_enhanced_generation": True},
|
|
)
|
|
|
|
generation_source = result["metadata"]["generation_source"]
|
|
self.assertEqual(result["generated_content"], "AI enhanced contract for Supplier A")
|
|
# AI enhancement call + AI quality evaluation call
|
|
self.assertEqual(len(fake_client.calls), 2)
|
|
self.assertTrue(generation_source["ai_enhancement_applied"])
|
|
self.assertFalse(generation_source["fallback_used"])
|
|
self.assertEqual(generation_source["ai_provider"], "shared-ai-platform")
|
|
|
|
def test_ai_enhanced_generation_uses_html_from_structured_shared_ai_payload(self):
|
|
now = datetime.now(timezone.utc)
|
|
template = self.ContractTemplate(
|
|
id="tpl-ai-structured",
|
|
tenant_id="tenant-demo-001",
|
|
name="AI Structured Template",
|
|
contract_type="procurement",
|
|
description="AI structured generation",
|
|
content="Contract with {{supplier_name}}",
|
|
variables_json=[{"name": "supplier_name", "label": "Supplier", "required": True}],
|
|
enabled=True,
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.db.add(template)
|
|
self.db.commit()
|
|
|
|
class FakeSharedAIClient:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def enhance_contract_generation(self, current_user, *, prompt_runtime, prompt, draft_content, metadata=None):
|
|
self.calls.append({
|
|
"current_user": current_user,
|
|
"prompt_runtime": prompt_runtime,
|
|
"prompt": prompt,
|
|
"draft_content": draft_content,
|
|
"metadata": metadata,
|
|
})
|
|
return {
|
|
"content": "第一条 纯文本兜底",
|
|
"raw_content": '{"title":"采购合同","html":"第一条 结构化合同正文"}',
|
|
"structured_output": {"title": "采购合同", "html": "第一条 结构化合同正文"},
|
|
"provider": "shared-ai-platform",
|
|
"model": "test-model",
|
|
"prompt_key": "contract_generation_default",
|
|
"prompt_version": "contract_generation_default.v1.0",
|
|
}
|
|
|
|
fake_client = FakeSharedAIClient()
|
|
self.service.shared_ai_platform_client = fake_client
|
|
|
|
result = self.service.generate_contract(
|
|
current_user=self.current_user,
|
|
template_id="tpl-ai-structured",
|
|
variable_values={"supplier_name": "Supplier A"},
|
|
options={"ai_enhanced_generation": True},
|
|
)
|
|
|
|
generation_source = result["metadata"]["generation_source"]
|
|
self.assertEqual(result["generated_content"], "第一条 结构化合同正文")
|
|
self.assertEqual(len(fake_client.calls), 2)
|
|
self.assertTrue(generation_source["ai_enhancement_applied"])
|
|
self.assertFalse(generation_source["fallback_used"])
|
|
|
|
def test_ai_enhanced_generation_falls_back_to_template_content(self):
|
|
now = datetime.now(timezone.utc)
|
|
template = self.ContractTemplate(
|
|
id="tpl-ai-fallback",
|
|
tenant_id="tenant-demo-001",
|
|
name="AI Fallback Template",
|
|
contract_type="procurement",
|
|
description="AI enhanced generation fallback",
|
|
content="Contract with {{supplier_name}}",
|
|
variables_json=[{"name": "supplier_name", "label": "Supplier", "required": True}],
|
|
enabled=True,
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.db.add(template)
|
|
self.db.commit()
|
|
|
|
class FailingSharedAIClient:
|
|
def enhance_contract_generation(self, current_user, *, prompt_runtime, prompt, draft_content, metadata=None):
|
|
raise RuntimeError("shared ai unavailable")
|
|
|
|
self.service.shared_ai_platform_client = FailingSharedAIClient()
|
|
|
|
result = self.service.generate_contract(
|
|
current_user=self.current_user,
|
|
template_id="tpl-ai-fallback",
|
|
variable_values={"supplier_name": "Supplier A"},
|
|
options={"ai_enhanced_generation": True},
|
|
)
|
|
|
|
generation_source = result["metadata"]["generation_source"]
|
|
self.assertEqual(result["generated_content"], "Contract with Supplier A")
|
|
self.assertFalse(generation_source["ai_enhancement_applied"])
|
|
self.assertTrue(generation_source["fallback_used"])
|
|
self.assertIn("shared ai unavailable", generation_source["fallback_reason"])
|
|
self.assertIn("AI 增强生成未生效", result["generation_quality"]["warning_issues"][-1])
|
|
|
|
def _seed_law_doc_004_data(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.LegalDocumentType(
|
|
id="ldt-procurement",
|
|
code="procurement_goods",
|
|
name="Procurement goods",
|
|
category="contract",
|
|
industry="trade",
|
|
scenario="purchase",
|
|
risk_level="medium",
|
|
is_active=True,
|
|
sort_order=10,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.LegalDocumentElementSchema(
|
|
id="ldes-procurement-v1",
|
|
document_type_code="procurement_goods",
|
|
version_label="v1.0",
|
|
schema_json={
|
|
"sections": [
|
|
{
|
|
"key": "base",
|
|
"title": "Base",
|
|
"fields": [
|
|
{"key": "party_a.name", "label": "Buyer", "type": "text", "required": True},
|
|
{"key": "party_b.name", "label": "Supplier", "type": "text", "required": True},
|
|
{"key": "goods.name", "label": "Goods", "type": "text", "required": True},
|
|
{"key": "price.total", "label": "Amount", "type": "money", "required": True},
|
|
{"key": "payment.terms", "label": "Payment", "type": "textarea", "required": True},
|
|
],
|
|
}
|
|
]
|
|
},
|
|
status="published",
|
|
published_at=now,
|
|
published_by="seed",
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.ContractTemplate(
|
|
id="tpl-elements",
|
|
tenant_id="tenant-demo-001",
|
|
name="Procurement Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
description="Element based template",
|
|
content="Buyer {{buyer}} buys {{product_name}} from {{supplier}} for {{amount}}. Payment: {{payment_terms}}. Extra: {{missing_required}}.",
|
|
variables_json=[
|
|
{"name": "buyer", "label": "Buyer", "required": True},
|
|
{"name": "supplier", "label": "Supplier", "required": True},
|
|
{"name": "product_name", "label": "Goods", "required": True},
|
|
{"name": "amount", "label": "Amount", "required": True},
|
|
{"name": "payment_terms", "label": "Payment", "required": True},
|
|
{"name": "missing_required", "label": "Extra", "required": True},
|
|
],
|
|
enabled=True,
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.LegalArticle(
|
|
id="lart-civil-585",
|
|
tenant_id="tenant-demo-001",
|
|
source_id="civil-code",
|
|
source_title="中华人民共和国民法典",
|
|
source_type="law",
|
|
authority="NPC",
|
|
region="CN",
|
|
article_no="Article 585",
|
|
article_text="当事人可以约定一方违约时应当根据违约情况向对方支付一定数额的违约金。",
|
|
tags_json=["违约", "买卖合同"],
|
|
index_status="indexed",
|
|
published_status="published",
|
|
published_at=now,
|
|
published_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
def _seed_service_law_doc_data(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.LegalDocumentType(
|
|
id="ldt-service",
|
|
code="service_general",
|
|
name="Service general",
|
|
category="contract",
|
|
industry="service",
|
|
scenario="service-delivery",
|
|
risk_level="medium",
|
|
is_active=True,
|
|
sort_order=20,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.LegalDocumentElementSchema(
|
|
id="ldes-service-v1",
|
|
document_type_code="service_general",
|
|
version_label="v1.0",
|
|
schema_json={
|
|
"sections": [
|
|
{
|
|
"key": "base",
|
|
"title": "Base",
|
|
"fields": [
|
|
{"key": "party_a.name", "label": "Buyer", "type": "text", "required": True},
|
|
{"key": "party_b.name", "label": "Supplier", "type": "text", "required": True},
|
|
{"key": "service.name", "label": "Service", "type": "text", "required": True},
|
|
{"key": "service.scope", "label": "Scope", "type": "textarea", "required": True},
|
|
{"key": "deliverables", "label": "Deliverables", "type": "textarea", "required": True},
|
|
{"key": "service.period", "label": "Period", "type": "text", "required": True},
|
|
{"key": "service_fee", "label": "Fee", "type": "money", "required": True},
|
|
{"key": "payment_terms", "label": "Payment", "type": "textarea", "required": True},
|
|
{"key": "acceptance.standard", "label": "Acceptance", "type": "textarea", "required": True},
|
|
{"key": "data.scope", "label": "Data Scope", "type": "textarea", "required": True},
|
|
{"key": "confidentiality.scope", "label": "Confidentiality", "type": "textarea", "required": True},
|
|
{"key": "breach_liability", "label": "Breach", "type": "textarea", "required": True},
|
|
{
|
|
"key": "dispute_resolution.method",
|
|
"label": "Dispute",
|
|
"type": "select",
|
|
"required": True,
|
|
"options": ["仲裁", "诉讼"],
|
|
},
|
|
{"key": "sign_date", "label": "Sign Date", "type": "date", "required": True},
|
|
],
|
|
}
|
|
]
|
|
},
|
|
status="published",
|
|
published_at=now,
|
|
published_by="seed",
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.ContractTemplate(
|
|
id="tpl-service-elements",
|
|
tenant_id="tenant-demo-001",
|
|
name="Service Template",
|
|
contract_type="service",
|
|
document_type_code="service_general",
|
|
description="Element based service template",
|
|
content=(
|
|
"Buyer {{buyer}} engages {{supplier}} for {{service_name}}. "
|
|
"Scope: {{service_scope}}. Deliverables: {{deliverables}}. "
|
|
"Period: {{service_period}}. Fee: {{service_fee}}. "
|
|
"Payment: {{payment_terms}}. Data: {{data_scope}}. "
|
|
"Dispute: {{dispute_resolution}}. Extra: {{missing_required}}."
|
|
),
|
|
variables_json=[
|
|
{"name": "buyer", "label": "Buyer", "required": True},
|
|
{"name": "supplier", "label": "Supplier", "required": True},
|
|
{"name": "service_name", "label": "Service", "required": True},
|
|
{"name": "service_scope", "label": "Scope", "required": True},
|
|
{"name": "deliverables", "label": "Deliverables", "required": True},
|
|
{"name": "service_period", "label": "Period", "required": True},
|
|
{"name": "service_fee", "label": "Fee", "required": True},
|
|
{"name": "payment_terms", "label": "Payment", "required": True},
|
|
{"name": "acceptance_standard", "label": "Acceptance", "required": True},
|
|
{"name": "data_scope", "label": "Data Scope", "required": True},
|
|
{"name": "confidentiality_scope", "label": "Confidentiality", "required": True},
|
|
{"name": "breach_liability", "label": "Breach", "required": True},
|
|
{"name": "dispute_resolution", "label": "Dispute", "required": True},
|
|
{"name": "sign_date", "label": "Sign Date", "required": True},
|
|
{"name": "missing_required", "label": "Extra", "required": True},
|
|
],
|
|
enabled=True,
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.LegalArticle(
|
|
id="lart-service-509",
|
|
tenant_id="tenant-demo-001",
|
|
source_id="lart-service-509",
|
|
source_title="中华人民共和国民法典",
|
|
source_type="law",
|
|
authority="NPC",
|
|
region="CN",
|
|
article_no="Article 509",
|
|
article_text="当事人应当按照约定全面履行自己的义务。服务合同应明确服务范围、验收、数据处理和违约责任。",
|
|
tags_json=["服务合同", "数据处理", "验收"],
|
|
index_status="indexed",
|
|
published_status="published",
|
|
published_at=now,
|
|
published_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
def test_generate_contract_accepts_json_column_variable_definitions(self):
|
|
now = datetime.now(timezone.utc)
|
|
template = self.ContractTemplate(
|
|
id="tpl-real-json",
|
|
tenant_id="tenant-demo-001",
|
|
name="Procurement Template",
|
|
contract_type="procurement",
|
|
description="JSON variables",
|
|
content="Contract with {{supplier_name}}",
|
|
variables_json=[{"name": "supplier_name", "label": "Supplier", "required": True}],
|
|
enabled=True,
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.db.add(template)
|
|
self.db.commit()
|
|
result = self.service.generate_contract(
|
|
current_user=self.current_user,
|
|
template_id="tpl-real-json",
|
|
variable_values={"supplier_name": "Supplier A"},
|
|
)
|
|
self.assertEqual(result["template_id"], "tpl-real-json")
|
|
self.assertEqual(result["variable_values"]["supplier_name"], "Supplier A")
|
|
# auto_review 已改为异步队列，不再同步调用 create_review_task
|
|
# auto_review 已改为异步，不再同步调用 review_service.create_review_task
|
|
self.assertIn(result.get("auto_review", {}).get("status"), (None, "pending"))
|
|
self.assertEqual(
|
|
result["metadata"]["prompt_runtime"]["contract_generation"]["prompt_version"],
|
|
"contract_generation.template_render.v1",
|
|
)
|
|
|
|
def test_property_service_chinese_variable_template_supports_preview_and_generate(self):
|
|
now = datetime.now(timezone.utc)
|
|
template = self.ContractTemplate(
|
|
id="tpl-property-cn",
|
|
tenant_id="tenant-demo-001",
|
|
name="物业服务合同模板",
|
|
contract_type="property_service",
|
|
document_type_code="property_service",
|
|
description="Chinese variable template",
|
|
content="甲方：{{甲方名称}}\n乙方：{{乙方名称}}\n项目：{{项目/物业名称}}",
|
|
variables_json=[
|
|
{"name": "甲方名称", "label": "甲方名称", "required": True},
|
|
{"name": "乙方名称", "label": "乙方名称", "required": True},
|
|
{"name": "项目/物业名称", "label": "项目/物业名称", "required": True},
|
|
],
|
|
enabled=True,
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.db.add(template)
|
|
self.db.commit()
|
|
|
|
variable_values = {
|
|
"甲方名称": "甲方物业公司",
|
|
"乙方名称": "乙方服务公司",
|
|
"项目/物业名称": "测试小区",
|
|
}
|
|
|
|
preview = self.service.preview_contract(
|
|
current_user=self.current_user,
|
|
template_id="tpl-property-cn",
|
|
variable_values=variable_values,
|
|
)
|
|
self.assertIn("甲方物业公司", preview["preview_content"])
|
|
self.assertEqual(preview["variable_summary"][0]["name"], "甲方名称")
|
|
self.assertTrue(preview["variable_summary"][0]["provided"])
|
|
|
|
result = self.service.generate_contract(
|
|
current_user=self.current_user,
|
|
template_id="tpl-property-cn",
|
|
variable_values=variable_values,
|
|
)
|
|
self.assertEqual(result["template_id"], "tpl-property-cn")
|
|
self.assertEqual(result["document_type_code"], "property_service")
|
|
self.assertEqual(result["variable_values"]["甲方名称"], "甲方物业公司")
|
|
self.assertIn("测试小区", result["generated_content"])
|
|
|
|
def test_generate_contract_from_elements_auto_selects_template_and_returns_law_doc_004_payload(self):
|
|
self._seed_law_doc_004_data()
|
|
result = self.service.generate_contract_from_elements(
|
|
current_user=self.current_user,
|
|
document_type_code="procurement_goods",
|
|
elements={
|
|
"party_a": {"name": "Buyer Co"},
|
|
"party_b": {"name": "Supplier Co"},
|
|
"goods": {"name": "Servers"},
|
|
"price": {"total": "180000"},
|
|
},
|
|
export_format=ExportFormat.HTML,
|
|
)
|
|
self.assertEqual(result["template_id"], "tpl-elements")
|
|
self.assertEqual(result["document_type_code"], "procurement_goods")
|
|
self.assertIn("missing-fields", result["generated_html"])
|
|
self.assertIn("payment.terms", [item["field_key"] for item in result["missing_fields"]])
|
|
self.assertIn("missing_required", [item["field_key"] for item in result["missing_fields"]])
|
|
self.assertEqual(result["related_legal_articles"][0]["source_id"], "lart-civil-585")
|
|
self.assertTrue(result["ai_hints"])
|
|
self.assertIn("generated_html", result)
|
|
self.assertEqual(result["metadata"]["generation_source"]["template_selection"], "auto")
|
|
# auto_review 已改为异步，不再同步调用 review_service.create_review_task
|
|
|
|
def test_generate_contract_from_elements_auto_selects_service_template_and_returns_law_doc_payload(self):
|
|
self._seed_service_law_doc_data()
|
|
result = self.service.generate_contract_from_elements(
|
|
current_user=self.current_user,
|
|
document_type_code="service_general",
|
|
elements={
|
|
"party_a": {"name": "Buyer Co"},
|
|
"party_b": {"name": "Supplier Co"},
|
|
"service": {
|
|
"name": "Operations Support Service",
|
|
"scope": "Operation support and weekly reporting",
|
|
"period": "2026-06-01 to 2027-05-31",
|
|
},
|
|
"deliverables": "Weekly reports and monthly recap",
|
|
"service_fee": "360000",
|
|
"payment_terms": "Quarterly in arrears",
|
|
"acceptance": {"standard": "Reports are complete and traceable"},
|
|
"data": {"scope": "Only work order ids and contact details"},
|
|
"confidentiality": {"scope": "Business data, customer data, and pricing"},
|
|
"breach_liability": "Delay, leakage, or unauthorized subcontracting must be compensated",
|
|
"dispute_resolution": {"method": "诉讼"},
|
|
"sign_date": "2026-05-23",
|
|
},
|
|
export_format=ExportFormat.HTML,
|
|
)
|
|
self.assertEqual(result["template_id"], "tpl-service-elements")
|
|
self.assertEqual(result["document_type_code"], "service_general")
|
|
self.assertIn("missing-fields", result["generated_html"])
|
|
self.assertIn("missing_required", [item["field_key"] for item in result["missing_fields"]])
|
|
self.assertEqual(result["related_legal_articles"][0]["source_id"], "lart-service-509")
|
|
self.assertEqual(result["metadata"]["generation_source"]["template_selection"], "auto")
|
|
# auto_review 已改为异步，不再同步调用 review_service.create_review_task
|
|
|
|
def test_get_generated_contract_download_allows_owner_access_to_file_payload(self):
|
|
generated_contract = type(
|
|
"GeneratedContractRecord",
|
|
(),
|
|
{
|
|
"id": "contract-owner-001",
|
|
"tenant_id": "tenant-demo-001",
|
|
"file_id": "file-owner-001",
|
|
"file_name": "contract.txt",
|
|
"mime_type": "text/plain",
|
|
"file_content_base64": "",
|
|
},
|
|
)()
|
|
captured = {}
|
|
self.service.generated_contract_repository = type(
|
|
"StubGeneratedContractRepository",
|
|
(),
|
|
{"get_by_id": lambda _self, contract_id, tenant_id: generated_contract},
|
|
)()
|
|
self.service.file_service = type(
|
|
"StubFileService",
|
|
(),
|
|
{
|
|
"get_download_payload": lambda _self, current_user, file_id, allow_owner_access=False: captured.setdefault(
|
|
"payload",
|
|
{
|
|
"current_user": current_user,
|
|
"file_id": file_id,
|
|
"allow_owner_access": allow_owner_access,
|
|
"file_name": "contract.txt",
|
|
"mime_type": "text/plain",
|
|
"content": b"contract-bytes",
|
|
},
|
|
)
|
|
},
|
|
)()
|
|
result = self.service.get_generated_contract_download(
|
|
current_user={"id": "user-member", "tenant_id": "tenant-demo-001", "role": "member"},
|
|
contract_id="contract-owner-001",
|
|
)
|
|
self.assertEqual(result["file_name"], "contract.txt")
|
|
self.assertTrue(captured["payload"]["allow_owner_access"])
|
|
|
|
def test_create_generated_contract_version_saves_editor_content(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.ContractTemplate(
|
|
id="tpl-editor",
|
|
tenant_id="tenant-demo-001",
|
|
name="Editor Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
description="Editor template",
|
|
content="Original contract content.",
|
|
variables_json=[],
|
|
enabled=True,
|
|
created_by="seed",
|
|
updated_by="seed",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-editor-001",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-editor",
|
|
template_name="Editor Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="editor.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Original contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
result = self.service.create_generated_contract_version(
|
|
current_user=self.current_user,
|
|
contract_id="contract-editor-001",
|
|
content="Original contract content.\nAcceptance clause added.",
|
|
change_type="clause_recommendation",
|
|
change_summary="Apply acceptance clause",
|
|
replacement_text="Acceptance clause added.",
|
|
)
|
|
|
|
self.assertEqual(result["version_label"], "v2")
|
|
self.assertEqual(result["change_type"], "clause_recommendation")
|
|
self.assertIn("Acceptance clause added", result["content"])
|
|
|
|
def test_create_generated_contract_version_rejects_unchanged_content(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-editor-unchanged",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-editor",
|
|
template_name="Editor Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="editor.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Original contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
with self.assertRaises(ConflictError) as ctx:
|
|
self.service.create_generated_contract_version(
|
|
current_user=self.current_user,
|
|
contract_id="contract-editor-unchanged",
|
|
content="Original contract content.",
|
|
)
|
|
|
|
self.assertEqual(ctx.exception.status_code, 409)
|
|
|
|
def test_create_generated_contract_version_rejects_non_owner_member(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-editor-owner-only",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-editor",
|
|
template_name="Editor Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="editor-owner-only.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Original contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
with self.assertRaises(NotFoundError) as ctx:
|
|
self.service.create_generated_contract_version(
|
|
current_user={
|
|
"id": "user-member",
|
|
"tenant_id": "tenant-demo-001",
|
|
"display_name": "Member",
|
|
"role": "member",
|
|
},
|
|
contract_id="contract-editor-owner-only",
|
|
content="Original contract content.\nAcceptance clause added.",
|
|
change_summary="member edit attempt",
|
|
)
|
|
|
|
self.assertEqual(ctx.exception.status_code, 404)
|
|
|
|
def test_diff_generated_contract_versions_returns_paragraph_changes(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-diff-001",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-diff",
|
|
template_name="Diff Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="diff.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="第一条 标的\n第二条 价款",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.GeneratedContractVersion(
|
|
id="gcv-diff-002",
|
|
tenant_id="tenant-demo-001",
|
|
contract_id="contract-diff-001",
|
|
version_number=2,
|
|
version_label="v2",
|
|
change_type="editor_save",
|
|
change_summary="补充验收",
|
|
content="第一条 标的\n第二条 价款\n第三条 验收",
|
|
locate_status="applied",
|
|
created_by="user-admin",
|
|
created_at=now + timedelta(minutes=1),
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
result = self.service.diff_generated_contract_versions(
|
|
current_user=self.current_user,
|
|
contract_id="contract-diff-001",
|
|
)
|
|
|
|
self.assertEqual(result["from_version"]["version_label"], "v1")
|
|
self.assertEqual(result["to_version"]["version_label"], "v2")
|
|
self.assertEqual(result["summary"]["added"], 1)
|
|
self.assertEqual(result["items"][0]["change_type"], "added")
|
|
self.assertIn("第三条 验收", result["items"][0]["new_text"])
|
|
|
|
def test_export_generated_contract_version_returns_docx_payload(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-export-version-001",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-export",
|
|
template_name="Export Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="export.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="第一条 标的\n第二条 价款",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.GeneratedContractVersion(
|
|
id="gcv-export-002",
|
|
tenant_id="tenant-demo-001",
|
|
contract_id="contract-export-version-001",
|
|
version_number=2,
|
|
version_label="v2",
|
|
change_type="editor_save",
|
|
change_summary="补充交付",
|
|
content="第一条 标的\n第二条 价款\n第三条 交付",
|
|
locate_status="applied",
|
|
created_by="user-admin",
|
|
created_at=now + timedelta(minutes=1),
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
result = self.service.export_generated_contract_version(
|
|
current_user=self.current_user,
|
|
contract_id="contract-export-version-001",
|
|
version_id="gcv-export-002",
|
|
)
|
|
|
|
self.assertEqual(result["export_format"], "docx")
|
|
self.assertEqual(result["version"]["version_label"], "v2")
|
|
self.assertTrue(result["file_name"].endswith(".docx"))
|
|
self.assertTrue(result["content"].startswith(b"PK"))
|
|
|
|
def test_update_generated_contract_lifecycle_tracks_expiry_and_owner(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-lifecycle-001",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-lifecycle",
|
|
template_name="Lifecycle Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="lifecycle.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Lifecycle contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
result = self.service.update_generated_contract_lifecycle(
|
|
current_user=self.current_user,
|
|
contract_id="contract-lifecycle-001",
|
|
lifecycle_status="effective",
|
|
counterparty_name="Supplier A",
|
|
owner_name="Legal Ops",
|
|
effective_at=now,
|
|
expires_at=now + timedelta(days=20),
|
|
lifecycle_note="已签署，进入履约跟踪。",
|
|
)
|
|
|
|
self.assertEqual(result["lifecycle"]["status"], "effective")
|
|
self.assertEqual(result["lifecycle"]["counterparty_name"], "Supplier A")
|
|
self.assertEqual(result["lifecycle"]["owner_name"], "Legal Ops")
|
|
self.assertEqual(result["lifecycle"]["due_status"], "due_soon")
|
|
self.assertEqual(result["lifecycle"]["days_to_expiry"], 20)
|
|
self.assertEqual(result["lifecycle"]["updated_by"], "user-admin")
|
|
|
|
def test_sync_contract_lifecycle_tasks_creates_renewal_reminder(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-renewal-001",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-renewal",
|
|
template_name="Renewal Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="renewal.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Renewal contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
lifecycle_status="effective",
|
|
expires_at=now + timedelta(days=20),
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
result = self.service.sync_contract_lifecycle_tasks(
|
|
current_user=self.current_user,
|
|
contract_id="contract-renewal-001",
|
|
)
|
|
|
|
self.assertTrue(result["synced"])
|
|
self.assertEqual(result["items"][0]["task_type"], "renewal_reminder")
|
|
self.assertEqual(result["items"][0]["priority"], "high")
|
|
self.assertEqual(result["summary"]["open"], 1)
|
|
persisted = self.db.query(self.ContractLifecycleTask).filter_by(contract_id="contract-renewal-001").one()
|
|
self.assertEqual(persisted.task_key, "renewal:contract-renewal-001")
|
|
|
|
def test_lifecycle_task_workbench_scopes_to_owned_contracts(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add_all(
|
|
[
|
|
self.GeneratedContract(
|
|
id="contract-task-owned",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-task-owned",
|
|
template_name="Owned Task Contract",
|
|
contract_type="service",
|
|
document_type_code="technical_service",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="owned-task.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Owned task content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
),
|
|
self.GeneratedContract(
|
|
id="contract-task-other",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-task-other",
|
|
template_name="Other Task Contract",
|
|
contract_type="service",
|
|
document_type_code="technical_service",
|
|
generated_by="user-other",
|
|
generated_by_name="Other",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="other-task.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Other task content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
),
|
|
self.ContractLifecycleTask(
|
|
id="task-owned",
|
|
tenant_id="tenant-demo-001",
|
|
contract_id="contract-task-owned",
|
|
task_key="manual:owned",
|
|
task_type="renewal_reminder",
|
|
title="Owned renewal reminder",
|
|
priority="high",
|
|
status="open",
|
|
due_at=now + timedelta(days=3),
|
|
created_by="user-admin",
|
|
created_at=now,
|
|
),
|
|
self.ContractLifecycleTask(
|
|
id="task-other",
|
|
tenant_id="tenant-demo-001",
|
|
contract_id="contract-task-other",
|
|
task_key="manual:other",
|
|
task_type="renewal_reminder",
|
|
title="Other renewal reminder",
|
|
priority="high",
|
|
status="open",
|
|
due_at=now + timedelta(days=3),
|
|
created_by="user-other",
|
|
created_at=now,
|
|
),
|
|
]
|
|
)
|
|
self.db.commit()
|
|
|
|
member_result = self.service.list_contract_lifecycle_task_workbench(
|
|
current_user={"id": "user-admin", "tenant_id": "tenant-demo-001", "display_name": "Admin", "role": "business"},
|
|
)
|
|
self.assertEqual(member_result["total"], 1)
|
|
self.assertEqual(member_result["items"][0]["contract_id"], "contract-task-owned")
|
|
self.assertEqual(member_result["items"][0]["contract"]["template_name"], "Owned Task Contract")
|
|
|
|
admin_result = self.service.list_contract_lifecycle_task_workbench(current_user=self.current_user)
|
|
self.assertEqual(admin_result["total"], 2)
|
|
|
|
def test_contract_lifecycle_task_can_bind_archive_evidence(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-archive-001",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-archive",
|
|
template_name="Archive Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="archive.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Archive contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
lifecycle_status="terminated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.add(
|
|
self.File(
|
|
id="file-archive-evidence",
|
|
tenant_id="tenant-demo-001",
|
|
uploaded_by="user-admin",
|
|
bucket="default",
|
|
object_key="contract-lifecycle-evidence/file-archive-evidence.pdf",
|
|
file_name="termination-proof.pdf",
|
|
mime_type="application/pdf",
|
|
file_size=256,
|
|
file_hash="hash-archive-evidence",
|
|
file_type="contract_lifecycle_evidence",
|
|
status="active",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
created = self.service.create_contract_lifecycle_task(
|
|
current_user=self.current_user,
|
|
contract_id="contract-archive-001",
|
|
task_type="termination_archive",
|
|
title="终止归档",
|
|
status_value="completed",
|
|
evidence_file_id="file-archive-evidence",
|
|
evidence_note="双方确认终止并归档。",
|
|
)
|
|
|
|
self.assertEqual(created["status"], "completed")
|
|
self.assertEqual(created["evidence_file"]["file_name"], "termination-proof.pdf")
|
|
self.assertEqual(created["due_status"], "closed")
|
|
listed = self.service.list_contract_lifecycle_tasks(self.current_user, "contract-archive-001")
|
|
self.assertEqual(listed["summary"]["completed"], 1)
|
|
self.assertEqual(listed["summary"]["with_evidence"], 1)
|
|
|
|
def test_contract_lifecycle_task_rejects_missing_evidence_file(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-missing-evidence",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-archive",
|
|
template_name="Archive Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="archive.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Archive contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
with self.assertRaises(NotFoundError) as ctx:
|
|
self.service.create_contract_lifecycle_task(
|
|
current_user=self.current_user,
|
|
contract_id="contract-missing-evidence",
|
|
task_type="termination_archive",
|
|
title="终止归档",
|
|
evidence_file_id="file-missing",
|
|
)
|
|
|
|
self.assertEqual(ctx.exception.status_code, 404)
|
|
|
|
def test_update_generated_contract_lifecycle_rejects_invalid_range(self):
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-lifecycle-invalid",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-lifecycle",
|
|
template_name="Lifecycle Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="lifecycle.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Lifecycle contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
with self.assertRaises(ValidationError) as ctx:
|
|
self.service.update_generated_contract_lifecycle(
|
|
current_user=self.current_user,
|
|
contract_id="contract-lifecycle-invalid",
|
|
lifecycle_status="effective",
|
|
effective_at=now,
|
|
expires_at=now - timedelta(days=1),
|
|
)
|
|
|
|
self.assertEqual(ctx.exception.status_code, 400)
|
|
|
|
def test_contract_approval_submit_and_approve_updates_lifecycle(self):
|
|
from app.services.contract_approval_service import ContractApprovalService
|
|
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add(
|
|
self.GeneratedContract(
|
|
id="contract-approval-001",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-approval",
|
|
template_name="Approval Template",
|
|
contract_type="procurement",
|
|
document_type_code="procurement_goods",
|
|
generated_by="user-admin",
|
|
generated_by_name="Admin",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="approval.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Approval contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
approval_service = ContractApprovalService(self.db)
|
|
submitted = approval_service.submit_approval(
|
|
current_user=self.current_user,
|
|
contract_id="contract-approval-001",
|
|
title="采购合同法务审批",
|
|
comment="请重点关注付款和验收条款。",
|
|
assignee_id="user-admin",
|
|
assignee_name="Admin",
|
|
)
|
|
|
|
self.assertEqual(submitted["status"], "pending")
|
|
self.assertEqual(submitted["contract"]["lifecycle_status"], "internal_review")
|
|
pending_list = approval_service.list_approvals(
|
|
current_user=self.current_user,
|
|
scope="assigned_to_me",
|
|
status_value="pending",
|
|
)
|
|
self.assertEqual(pending_list["total"], 1)
|
|
self.assertEqual(pending_list["summary"]["assigned_to_me"], 1)
|
|
|
|
decided = approval_service.decide_approval(
|
|
current_user=self.current_user,
|
|
approval_id=submitted["id"],
|
|
decision="approve",
|
|
comment="审批通过。",
|
|
)
|
|
|
|
self.assertEqual(decided["status"], "approved")
|
|
contract = self.db.query(self.GeneratedContract).filter_by(id="contract-approval-001").one()
|
|
self.assertEqual(contract.lifecycle_status, "approved")
|
|
detail = self.service.get_generated_contract(self.current_user, "contract-approval-001")
|
|
self.assertEqual(detail["approval"]["status"], "approved")
|
|
self.assertEqual(detail["approval"]["latest"]["comment"], "审批通过。")
|
|
|
|
def test_contract_approval_assignees_and_default_admin_assignment(self):
|
|
from app.services.contract_approval_service import ContractApprovalService
|
|
|
|
now = datetime.now(timezone.utc)
|
|
self.db.add_all(
|
|
[
|
|
self.User(
|
|
id="user-admin-approval",
|
|
tenant_id="tenant-demo-001",
|
|
username="approval-admin@example.com",
|
|
display_name="Approval Admin",
|
|
password_hash="hash",
|
|
role="admin",
|
|
status="active",
|
|
created_at=now,
|
|
),
|
|
self.User(
|
|
id="user-legal-approval",
|
|
tenant_id="tenant-demo-001",
|
|
username="approval-legal@example.com",
|
|
display_name="Approval Legal",
|
|
password_hash="hash",
|
|
role="legal",
|
|
status="active",
|
|
created_at=now,
|
|
),
|
|
self.User(
|
|
id="user-disabled-approval",
|
|
tenant_id="tenant-demo-001",
|
|
username="approval-disabled@example.com",
|
|
display_name="Disabled User",
|
|
password_hash="hash",
|
|
role="legal",
|
|
status="disabled",
|
|
created_at=now,
|
|
),
|
|
self.GeneratedContract(
|
|
id="contract-approval-default-admin",
|
|
tenant_id="tenant-demo-001",
|
|
template_id="tpl-approval-default",
|
|
template_name="Default Approval Template",
|
|
contract_type="service",
|
|
document_type_code="technical_service",
|
|
generated_by="user-business-owner",
|
|
generated_by_name="Business Owner",
|
|
export_format="docx",
|
|
file_id=None,
|
|
file_name="approval-default.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=10,
|
|
generated_content="Approval contract content.",
|
|
variable_values_json={},
|
|
export_options_json={},
|
|
file_content_base64="",
|
|
status="generated",
|
|
created_at=now,
|
|
),
|
|
]
|
|
)
|
|
self.db.commit()
|
|
|
|
approval_service = ContractApprovalService(self.db)
|
|
business_user = {
|
|
"id": "user-business-owner",
|
|
"tenant_id": "tenant-demo-001",
|
|
"display_name": "Business Owner",
|
|
"role": "business",
|
|
}
|
|
|
|
assignees = approval_service.list_assignees(business_user)
|
|
assignee_ids = [item["id"] for item in assignees["items"]]
|
|
self.assertIn("user-admin-approval", assignee_ids)
|
|
self.assertIn("user-legal-approval", assignee_ids)
|
|
self.assertNotIn("user-disabled-approval", assignee_ids)
|
|
self.assertTrue(next(item for item in assignees["items"] if item["id"] == "user-admin-approval")["recommended"])
|
|
|
|
submitted = approval_service.submit_approval(
|
|
current_user=business_user,
|
|
contract_id="contract-approval-default-admin",
|
|
title="服务合同审批",
|
|
comment="请法务关注验收条款。",
|
|
)
|
|
|
|
self.assertEqual(submitted["status"], "pending")
|
|
self.assertEqual(submitted["assignee_id"], "user-admin-approval")
|
|
self.assertEqual(submitted["assignee_name"], "Approval Admin")
|
|
|
|
|
|
class ClauseSuggestionRuleTests(unittest.TestCase):
|
|
def setUp(self):
|
|
from app.services.contract_generation_service import ContractGenerationService
|
|
|
|
self.service = ContractGenerationService(db=None)
|
|
|
|
def test_build_clause_suggestions_prefers_review_risk_rules(self):
|
|
generation_quality = {
|
|
"section_coverage": {
|
|
"has_payment": True,
|
|
"has_acceptance_or_delivery": True,
|
|
"has_breach": True,
|
|
"has_dispute_resolution": True,
|
|
},
|
|
"warning_issues": [],
|
|
"placeholder_count": 0,
|
|
}
|
|
auto_review = {
|
|
"status": "completed",
|
|
"risk_level": "high",
|
|
"review_opinion": "合同存在自动续约和免责条款风险。",
|
|
"recommendation": "建议补充续约通知期，并限制免责范围。",
|
|
"risk_items": [
|
|
{"title": "自动续约风险", "description": "自动续约条款对我方不利", "suggestion": "补充退出机制"},
|
|
{"title": "免责条款风险", "description": "免责范围过宽", "suggestion": "增加 carve-out"},
|
|
],
|
|
}
|
|
|
|
suggestions = self.service._build_clause_completion_suggestions(generation_quality, auto_review)
|
|
|
|
self.assertTrue(any(item["code"] == "renewal_notice_guardrail" for item in suggestions))
|
|
self.assertTrue(any(item["code"] == "liability_carve_out" for item in suggestions))
|
|
self.assertTrue(all(item.get("source") == "risk_review" for item in suggestions[:2]))
|
|
|
|
def test_build_clause_suggestions_falls_back_to_quality_rules(self):
|
|
generation_quality = {
|
|
"section_coverage": {
|
|
"has_payment": False,
|
|
"has_acceptance_or_delivery": True,
|
|
"has_breach": False,
|
|
"has_dispute_resolution": True,
|
|
},
|
|
"warning_issues": ["缺少付款安排"],
|
|
"placeholder_count": 1,
|
|
}
|
|
auto_review = {"status": "failed", "risk_level": None}
|
|
|
|
suggestions = self.service._build_clause_completion_suggestions(generation_quality, auto_review)
|
|
|
|
codes = {item["code"] for item in suggestions}
|
|
self.assertIn("payment", codes)
|
|
self.assertIn("breach", codes)
|
|
self.assertIn("placeholder_cleanup", codes)
|
|
self.assertTrue(all(item.get("source") == "quality" for item in suggestions))
|