Files
ai-contract-platform/backend/tests/test_contracts_api.py
T

3468 lines
152 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.api.notifications import require_current_user as require_notification_current_user
from app.api.notifications import get_notification_service
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 StubNotificationService:
def list_notifications(self, current_user, limit=20, scope="mine"):
return {
"items": [
{
"id": "contract:contract-123:approval_decision",
"type": "approval_pending",
"title": "待我审批",
"description": "合同正在审批流转中。",
"priority": "critical",
"status": "open",
"action_label": "处理审批",
"href": "/contracts/contract-123",
"source_type": "contract_attention",
"source_id": "approval_decision",
"contract_id": "contract-123",
"contract_title": "Procurement Template",
"is_mine": True,
"due_at": None,
"created_at": "2026-04-17T00:00:00+00:00",
"metadata": {},
}
],
"total": 1,
"limit": limit,
"scope": scope,
"summary": {"total": 1, "mine": 1, "critical": 1, "high": 0, "by_type": {"approval_pending": 1}, "by_priority": {"critical": 1}},
}
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,
},
"ledger_attention": {
"status": "normal",
"is_mine": generated_by == "user-admin",
"primary": None,
"items": [],
"counts": {"total": 0, "mine": 0, "open_tasks": 0, "pending_approval": 0},
},
"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,
participant_contract_ids=None,
counterparty_name=None,
owner_name=None,
effective_from=None,
effective_to=None,
expires_from=None,
expires_to=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": [
{"index": 1, "diff_key": "diff-1-stub", "change_type": "modified", "old_paragraph_index": 1, "new_paragraph_index": 1, "old_text": "Old", "new_text": "New"},
{"index": 2, "diff_key": "diff-2-stub", "change_type": "added", "old_paragraph_index": None, "new_paragraph_index": 3, "old_text": None, "new_text": "Payment deadline added."},
],
"comments": [],
"comment_summary": {"open": 0, "resolved": 0, "dismissed": 0, "total": 0},
}
def list_generated_contract_version_comments(self, current_user, contract_id, from_version_id=None, to_version_id=None, status_value=None):
self.get_generated_contract(current_user, contract_id)
items = [
{
"id": "gcvc-001",
"tenant_id": current_user["tenant_id"],
"contract_id": contract_id,
"from_version_id": from_version_id or "base",
"to_version_id": to_version_id or "gcv-001",
"diff_key": "diff-1-stub",
"change_type": "modified",
"old_text": "Old",
"new_text": "New",
"comment": "Please align payment wording.",
"status": "open",
"source": "version_diff",
"created_by": current_user["id"],
"created_by_name": "Admin",
"resolved_by": None,
"resolved_by_name": None,
"resolved_at": None,
"metadata": {},
"created_at": "2026-06-19T00:00:00+00:00",
"updated_at": "2026-06-19T00:00:00+00:00",
}
]
if status_value:
items = [item for item in items if item["status"] == status_value]
return {
"contract_id": contract_id,
"items": items,
"total": len(items),
"summary": {"open": len([item for item in items if item["status"] == "open"]), "resolved": 0, "dismissed": 0, "total": len(items)},
}
def create_generated_contract_version_comment(
self,
current_user,
contract_id,
from_version_id=None,
to_version_id=None,
diff_key=None,
change_type="modified",
old_text=None,
new_text=None,
comment=None,
source="version_diff",
metadata=None,
):
self.get_generated_contract(current_user, contract_id)
return {
"id": "gcvc-created-001",
"tenant_id": current_user["tenant_id"],
"contract_id": contract_id,
"from_version_id": from_version_id or "base",
"to_version_id": to_version_id or "gcv-001",
"diff_key": diff_key,
"change_type": change_type,
"old_text": old_text,
"new_text": new_text,
"comment": comment,
"status": "open",
"source": source,
"created_by": current_user["id"],
"created_by_name": "Admin",
"resolved_by": None,
"resolved_by_name": None,
"resolved_at": None,
"metadata": metadata or {},
"created_at": "2026-06-19T00:00:00+00:00",
"updated_at": "2026-06-19T00:00:00+00:00",
}
def update_generated_contract_version_comment(self, current_user, contract_id, comment_id, status_value=None, comment=None):
self.get_generated_contract(current_user, contract_id)
return {
"id": comment_id,
"tenant_id": current_user["tenant_id"],
"contract_id": contract_id,
"from_version_id": "base",
"to_version_id": "gcv-001",
"diff_key": "diff-1-stub",
"change_type": "modified",
"old_text": "Old",
"new_text": "New",
"comment": comment or "Please align payment wording.",
"status": status_value or "resolved",
"source": "version_diff",
"created_by": current_user["id"],
"created_by_name": "Admin",
"resolved_by": current_user["id"] if status_value == "resolved" else None,
"resolved_by_name": "Admin" if status_value == "resolved" else None,
"resolved_at": "2026-06-19T00:00:00+00:00" if status_value == "resolved" else None,
"metadata": {},
"created_at": "2026-06-19T00:00:00+00:00",
"updated_at": "2026-06-19T00:00:00+00:00",
}
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 export_generated_contracts_csv(
self,
current_user,
contract_type=None,
document_type_code=None,
lifecycle_status=None,
keyword=None,
counterparty_name=None,
owner_name=None,
effective_from=None,
effective_to=None,
expires_from=None,
expires_to=None,
limit=1000,
):
listed = self.list_generated_contracts(
current_user=current_user,
page=1,
page_size=limit,
contract_type=contract_type,
document_type_code=document_type_code,
lifecycle_status=lifecycle_status,
keyword=keyword,
counterparty_name=counterparty_name,
owner_name=owner_name,
effective_from=effective_from,
effective_to=effective_to,
expires_from=expires_from,
expires_to=expires_to,
)
lines = ["合同ID,合同名称,相对方,负责人"]
for item in listed["items"]:
lines.append(
",".join(
[
item["id"],
item["template_name"],
item["lifecycle"].get("counterparty_name") or "",
item["lifecycle"].get("owner_name") or item["metadata"].get("generated_by_name") or "",
]
)
)
return {
"file_name": "contract-ledger-test.csv",
"mime_type": "text/csv; charset=utf-8",
"content": ("\ufeff" + "\n".join(lines)).encode("utf-8"),
"count": len(listed["items"]),
"limit": limit,
}
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
app.dependency_overrides[get_notification_service] = lambda: StubNotificationService()
self.client = TestClient(app)
def tearDown(self):
app.dependency_overrides.clear()
def _login(self, tenant_id="tenant-demo-001"):
user = {
"id": "user-admin",
"tenant_id": tenant_id,
"display_name": "Admin",
"role": "admin",
}
app.dependency_overrides[require_current_user] = lambda: user
app.dependency_overrides[require_notification_current_user] = lambda: user
def _login_member(self, tenant_id="tenant-demo-001"):
user = {
"id": "user-member",
"tenant_id": tenant_id,
"display_name": "Member",
"role": "member",
}
app.dependency_overrides[require_current_user] = lambda: user
app.dependency_overrides[require_notification_current_user] = lambda: user
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")
self.assertIn("ledger_attention", body["data"]["items"][0])
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_user_can_list_notifications(self):
self._login()
response = self.client.get("/api/v1/notifications", params={"limit": 5})
self.assertEqual(response.status_code, 200)
body = response.json()["data"]
self.assertEqual(body["total"], 1)
self.assertEqual(body["items"][0]["type"], "approval_pending")
self.assertEqual(body["summary"]["critical"], 1)
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(diff_body["items"][0]["diff_key"], "diff-1-stub")
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_create_and_resolve_generated_contract_version_comment(self):
self._login()
list_response = self.client.get(
"/api/v1/contracts/contract-123/version-comments",
params={"from_version_id": "base", "to_version_id": "gcv-001"},
)
self.assertEqual(list_response.status_code, 200)
self.assertEqual(list_response.json()["data"]["items"][0]["diff_key"], "diff-1-stub")
create_response = self.client.post(
"/api/v1/contracts/contract-123/version-comments",
json={
"from_version_id": "base",
"to_version_id": "gcv-001",
"diff_key": "diff-1-stub",
"change_type": "modified",
"old_text": "Old",
"new_text": "New",
"comment": "Please align payment wording.",
},
)
self.assertEqual(create_response.status_code, 200)
self.assertEqual(create_response.json()["data"]["status"], "open")
self.assertEqual(self.audit_log_service.operations[-1]["action"], "contract_version_comment_create")
update_response = self.client.patch(
"/api/v1/contracts/contract-123/version-comments/gcvc-created-001",
json={"status": "resolved"},
)
self.assertEqual(update_response.status_code, 200)
self.assertEqual(update_response.json()["data"]["status"], "resolved")
self.assertEqual(self.audit_log_service.operations[-1]["action"], "contract_version_comment_update")
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")
export_response = self.client.get("/api/v1/contracts/export?contract_type=employment&limit=20")
self.assertEqual(export_response.status_code, 200)
self.assertIn("text/csv", export_response.headers["content-type"])
self.assertIn("filename*=UTF-8''contract-ledger-test.csv", export_response.headers["content-disposition"])
self.assertIn("contract-456", export_response.content.decode("utf-8-sig"))
self.assertEqual(self.audit_log_service.operations[-1]["action"], "export_contract_ledger")
self.assertEqual(self.audit_log_service.operations[-1]["detail_json"]["count"], 1)
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.generated_contract_version_comment import GeneratedContractVersionComment
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.GeneratedContractVersionComment = GeneratedContractVersionComment
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 _add_generated_contract(
self,
contract_id: str,
*,
generated_by: str = "user-admin",
generated_by_name: str = "Admin",
lifecycle_status: str = "generated",
):
now = datetime.now(timezone.utc)
contract = self.GeneratedContract(
id=contract_id,
tenant_id="tenant-demo-001",
template_id="tpl-approval",
template_name="Approval Template",
contract_type="procurement",
document_type_code="procurement_goods",
generated_by=generated_by,
generated_by_name=generated_by_name,
export_format="docx",
file_id=None,
file_name=f"{contract_id}.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",
lifecycle_status=lifecycle_status,
created_at=now,
)
self.db.add(contract)
self.db.commit()
return contract
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_strips_html_from_structured_payload(self):
now = datetime.now(timezone.utc)
template = self.ContractTemplate(
id="tpl-ai-html-only",
tenant_id="tenant-demo-001",
name="AI HTML Template",
contract_type="guarantee",
description="AI html generation",
content="Guarantee 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 enhance_contract_generation(self, current_user, *, prompt_runtime, prompt, draft_content, metadata=None):
return {
"content": '{"title":"Guarantee Contract","html":"<h1>Guarantee Contract</h1><p>Article 1 guarantee scope.</p>"}',
"raw_content": '{"title":"Guarantee Contract","html":"<h1>Guarantee Contract</h1><p>Article 1 guarantee scope.</p>"}',
"structured_output": {
"title": "Guarantee Contract",
"html": "<h1>Guarantee Contract</h1><p>Article 1 guarantee scope.</p>",
},
"provider": "shared-ai-platform",
"model": "test-model",
"prompt_key": "contract_generation_default",
"prompt_version": "contract_generation_default.v1.0",
}
self.service.shared_ai_platform_client = FakeSharedAIClient()
result = self.service.generate_contract(
current_user=self.current_user,
template_id="tpl-ai-html-only",
variable_values={"supplier_name": "Supplier A"},
options={"ai_enhanced_generation": True},
export_format=ExportFormat.TXT,
)
self.assertEqual(result["generated_content"], "Guarantee Contract\nArticle 1 guarantee scope.")
self.assertNotIn("<h1>", result["generated_content"])
self.assertNotIn("{\"title\"", result["generated_content"])
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_detail_and_preview_normalize_saved_json_content(self):
now = datetime.now(timezone.utc)
saved_json_content = '{"title":"Guarantee Contract","html":"<h1>Guarantee Contract</h1><p>Article 1 guarantee scope.</p>"}'
self.db.add(
self.GeneratedContract(
id="contract-json-content-001",
tenant_id="tenant-demo-001",
template_id="tpl-json",
template_name="Guarantee Template",
contract_type="guarantee",
document_type_code="guarantee_general",
generated_by="user-admin",
generated_by_name="Admin",
export_format="txt",
file_id=None,
file_name="guarantee.txt",
mime_type="text/plain; charset=utf-8",
file_size=10,
generated_content=saved_json_content,
variable_values_json={},
export_options_json={},
file_content_base64="",
status="generated",
created_at=now,
)
)
self.db.commit()
detail = self.service.get_generated_contract(self.current_user, "contract-json-content-001")
preview = self.service.get_generated_contract_preview(self.current_user, "contract-json-content-001")
self.assertEqual(detail["generated_content"], "Guarantee Contract\nArticle 1 guarantee scope.")
self.assertIn("<p>Guarantee Contract</p>", detail["generated_html"])
self.assertEqual(preview["preview_content"], "Guarantee Contract\nArticle 1 guarantee scope.")
self.assertIn("preview_html", preview)
self.assertNotIn("{\"title\"", detail["generated_content"])
self.assertNotIn("<h1>", preview["preview_content"])
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": "",
"generated_by": "user-member",
},
)()
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_get_generated_contract_download_recovers_when_file_storage_read_fails(self):
generated_contract = type(
"GeneratedContractRecord",
(),
{
"id": "contract-recover-001",
"tenant_id": "tenant-demo-001",
"file_id": "file-missing-001",
"file_name": "broken.txt",
"mime_type": "text/plain; charset=utf-8",
"file_content_base64": "",
"generated_by": "user-admin",
"generated_by_name": "Admin",
"generated_content": '{"title":"Guarantee Contract","html":"<p>Recovered clause.</p>"}',
"template_name": "Guarantee Template",
"export_format": "txt",
"export_options_json": {},
},
)()
self.service.generated_contract_repository = type(
"StubGeneratedContractRepository",
(),
{"get_by_id": lambda _self, contract_id, tenant_id: generated_contract},
)()
class FailingFileService:
def get_download_payload(self, current_user, file_id, allow_owner_access=False):
raise HTTPException(status_code=500, detail="file storage unavailable")
self.service.file_service = FailingFileService()
result = self.service.get_generated_contract_download(
current_user={"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"},
contract_id="contract-recover-001",
)
self.assertTrue(result["recovered_from_content"])
self.assertEqual(result["mime_type"], "text/plain; charset=utf-8")
self.assertIn(b"Recovered clause.", result["content"])
self.assertNotIn(b"<p>", result["content"])
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("diff_key", result["items"][0])
self.assertEqual(result["comment_summary"]["total"], 0)
self.assertIn("第三条 验收", result["items"][0]["new_text"])
def test_generated_contract_version_comments_roundtrip(self):
now = datetime.now(timezone.utc)
self.db.add(
self.GeneratedContract(
id="contract-diff-comment-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-comment.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-comment-002",
tenant_id="tenant-demo-001",
contract_id="contract-diff-comment-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()
diff_result = self.service.diff_generated_contract_versions(
current_user=self.current_user,
contract_id="contract-diff-comment-001",
from_version_id="base",
to_version_id="gcv-diff-comment-002",
)
diff_item = diff_result["items"][0]
created = self.service.create_generated_contract_version_comment(
current_user=self.current_user,
contract_id="contract-diff-comment-001",
from_version_id="base",
to_version_id="gcv-diff-comment-002",
diff_key=diff_item["diff_key"],
change_type=diff_item["change_type"],
old_text=diff_item["old_text"],
new_text=diff_item["new_text"],
comment="请补充验收触发条件后再提交审批。",
)
self.assertEqual(created["status"], "open")
self.assertEqual(created["from_version_id"], "base")
self.assertEqual(created["to_version_id"], "gcv-diff-comment-002")
listed = self.service.list_generated_contract_version_comments(
current_user=self.current_user,
contract_id="contract-diff-comment-001",
from_version_id="base",
to_version_id="gcv-diff-comment-002",
)
self.assertEqual(listed["summary"]["open"], 1)
self.assertEqual(listed["items"][0]["diff_key"], diff_item["diff_key"])
diff_with_comment = self.service.diff_generated_contract_versions(
current_user=self.current_user,
contract_id="contract-diff-comment-001",
from_version_id="base",
to_version_id="gcv-diff-comment-002",
)
self.assertEqual(diff_with_comment["comment_summary"]["open"], 1)
updated = self.service.update_generated_contract_version_comment(
current_user=self.current_user,
contract_id="contract-diff-comment-001",
comment_id=created["id"],
status_value="resolved",
)
self.assertEqual(updated["status"], "resolved")
self.assertEqual(updated["resolved_by"], "user-admin")
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_list_generated_contracts_includes_ledger_attention(self):
now = datetime.now(timezone.utc)
self._add_generated_contract(
"contract-ledger-attention-001",
generated_by="user-member",
generated_by_name="Member",
lifecycle_status="internal_review",
)
self.db.add(
self.ContractApprovalStep(
id="approval-ledger-001",
tenant_id="tenant-demo-001",
contract_id="contract-ledger-attention-001",
step_type="legal_review",
status="pending",
title="法务审批",
comment="请处理审批。",
submitter_id="user-member",
submitter_name="Member",
assignee_id="user-approver",
assignee_name="Approver",
created_at=now,
updated_at=now,
)
)
self.db.commit()
result = self.service.list_generated_contracts(
current_user={
"id": "user-approver",
"tenant_id": "tenant-demo-001",
"display_name": "Approver",
"role": "member",
},
page=1,
page_size=10,
)
self.assertEqual(result["total"], 1)
attention = result["items"][0]["ledger_attention"]
self.assertTrue(attention["is_mine"])
self.assertEqual(attention["primary"]["code"], "approval_decision")
self.assertEqual(attention["counts"]["pending_approval"], 1)
def test_list_generated_contracts_supports_ledger_advanced_filters(self):
now = datetime.now(timezone.utc)
alpha = self._add_generated_contract("contract-ledger-filter-alpha", lifecycle_status="effective")
alpha.counterparty_name = "Alpha Supplier"
alpha.owner_name = "Legal Ops"
alpha.effective_at = now - timedelta(days=10)
alpha.expires_at = now + timedelta(days=20)
beta = self._add_generated_contract("contract-ledger-filter-beta", lifecycle_status="effective")
beta.counterparty_name = "Beta Supplier"
beta.owner_name = "Business Owner"
beta.effective_at = now - timedelta(days=30)
beta.expires_at = now + timedelta(days=80)
self.db.add_all([alpha, beta])
self.db.commit()
by_counterparty = self.service.list_generated_contracts(
current_user=self.current_user,
page=1,
page_size=10,
counterparty_name="Alpha",
)
by_owner = self.service.list_generated_contracts(
current_user=self.current_user,
page=1,
page_size=10,
owner_name="Legal",
)
by_expiry = self.service.list_generated_contracts(
current_user=self.current_user,
page=1,
page_size=10,
expires_from=now,
expires_to=now + timedelta(days=30),
)
self.assertEqual([item["id"] for item in by_counterparty["items"]], ["contract-ledger-filter-alpha"])
self.assertEqual([item["id"] for item in by_owner["items"]], ["contract-ledger-filter-alpha"])
self.assertEqual([item["id"] for item in by_expiry["items"]], ["contract-ledger-filter-alpha"])
def test_list_generated_contracts_rejects_invalid_date_range(self):
now = datetime.now(timezone.utc)
with self.assertRaises(ValidationError):
self.service.list_generated_contracts(
current_user=self.current_user,
page=1,
page_size=10,
expires_from=now + timedelta(days=30),
expires_to=now,
)
def test_export_generated_contracts_csv_uses_ledger_filters(self):
now = datetime.now(timezone.utc)
alpha = self._add_generated_contract("contract-ledger-export-alpha", lifecycle_status="effective")
alpha.counterparty_name = "Alpha Supplier"
alpha.owner_name = "Legal Ops"
alpha.effective_at = now - timedelta(days=10)
alpha.expires_at = now + timedelta(days=20)
beta = self._add_generated_contract("contract-ledger-export-beta", lifecycle_status="effective")
beta.counterparty_name = "Beta Supplier"
beta.owner_name = "Business Owner"
beta.effective_at = now - timedelta(days=30)
beta.expires_at = now + timedelta(days=80)
self.db.add_all([alpha, beta])
self.db.commit()
exported = self.service.export_generated_contracts_csv(
current_user=self.current_user,
counterparty_name="Alpha",
expires_to=now + timedelta(days=30),
)
csv_text = exported["content"].decode("utf-8-sig")
self.assertEqual(exported["count"], 1)
self.assertIn("合同ID", csv_text)
self.assertIn("contract-ledger-export-alpha", csv_text)
self.assertIn("Alpha Supplier", csv_text)
self.assertNotIn("contract-ledger-export-beta", csv_text)
def test_legal_role_can_view_and_update_tenant_contracts(self):
self._add_generated_contract(
"contract-legal-visible-001",
generated_by="user-member",
generated_by_name="Member",
)
legal_user = {
"id": "user-legal",
"tenant_id": "tenant-demo-001",
"display_name": "Legal",
"role": "legal",
}
listed = self.service.list_generated_contracts(current_user=legal_user, page=1, page_size=10)
updated = self.service.update_generated_contract_lifecycle(
current_user=legal_user,
contract_id="contract-legal-visible-001",
lifecycle_status="external_review",
lifecycle_note="法务已推进外部审查。",
)
self.assertEqual(listed["total"], 1)
self.assertEqual(updated["lifecycle"]["status"], "external_review")
def test_observer_role_can_view_but_cannot_mutate_or_download_contracts(self):
self._add_generated_contract(
"contract-observer-visible-001",
generated_by="user-member",
generated_by_name="Member",
)
observer = {
"id": "user-observer",
"tenant_id": "tenant-demo-001",
"display_name": "Observer",
"role": "observer",
}
listed = self.service.list_generated_contracts(current_user=observer, page=1, page_size=10)
detail = self.service.get_generated_contract(observer, "contract-observer-visible-001")
self.assertEqual(listed["total"], 1)
self.assertEqual(detail["id"], "contract-observer-visible-001")
self.assertFalse(detail["approval"]["can_submit"])
with self.assertRaises(ForbiddenError):
self.service.export_generated_contracts_csv(observer)
with self.assertRaises(ForbiddenError):
self.service.update_generated_contract_lifecycle(
current_user=observer,
contract_id="contract-observer-visible-001",
lifecycle_status="external_review",
)
with self.assertRaises(ForbiddenError):
self.service.get_generated_contract_download(observer, "contract-observer-visible-001")
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_can_bind_signoff_evidence(self):
now = datetime.now(timezone.utc)
self.db.add(
self.GeneratedContract(
id="contract-signoff-001",
tenant_id="tenant-demo-001",
template_id="tpl-signoff",
template_name="Signoff 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="signoff.docx",
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
file_size=10,
generated_content="Signoff contract content.",
variable_values_json={},
export_options_json={},
file_content_base64="",
status="generated",
lifecycle_status="approved",
created_at=now,
)
)
self.db.add(
self.File(
id="file-signoff-evidence",
tenant_id="tenant-demo-001",
uploaded_by="user-admin",
bucket="default",
object_key="contract-lifecycle-evidence/file-signoff-evidence.pdf",
file_name="signed-contract.pdf",
mime_type="application/pdf",
file_size=512,
file_hash="hash-signoff-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-signoff-001",
task_type="signoff",
title="签署确认",
status_value="completed",
evidence_file_id="file-signoff-evidence",
evidence_note="双方已完成盖章签署。",
)
self.assertEqual(created["task_type"], "signoff")
self.assertEqual(created["status"], "completed")
self.assertEqual(created["evidence_file"]["file_name"], "signed-contract.pdf")
self.assertEqual(created["evidence_note"], "双方已完成盖章签署。")
def test_generated_contract_includes_signoff_archive_evidence_summary(self):
now = datetime.now(timezone.utc)
self._add_generated_contract("contract-signoff-archive-001", lifecycle_status="signed")
self.db.add_all(
[
self.File(
id="file-signoff-archive-signoff",
tenant_id="tenant-demo-001",
uploaded_by="user-admin",
bucket="default",
object_key="contract-lifecycle-evidence/file-signoff.pdf",
file_name="signed-contract.pdf",
mime_type="application/pdf",
file_size=512,
file_hash="hash-signoff",
file_type="contract_lifecycle_evidence",
status="active",
created_at=now,
),
self.File(
id="file-signoff-archive-archive",
tenant_id="tenant-demo-001",
uploaded_by="user-admin",
bucket="default",
object_key="contract-lifecycle-evidence/file-archive.pdf",
file_name="archive-index.pdf",
mime_type="application/pdf",
file_size=256,
file_hash="hash-archive",
file_type="contract_lifecycle_evidence",
status="active",
created_at=now,
),
]
)
self.db.commit()
self.service.create_contract_lifecycle_task(
current_user=self.current_user,
contract_id="contract-signoff-archive-001",
task_type="signoff",
title="签署确认",
status_value="completed",
evidence_file_id="file-signoff-archive-signoff",
evidence_note="双方已完成签署。",
)
self.service.create_contract_lifecycle_task(
current_user=self.current_user,
contract_id="contract-signoff-archive-001",
task_type="termination_archive",
title="归档确认",
status_value="completed",
evidence_file_id="file-signoff-archive-archive",
evidence_note="归档清单已留存。",
)
detail = self.service.get_generated_contract(self.current_user, "contract-signoff-archive-001")
signoff_archive = detail["signoff_archive"]
self.assertTrue(signoff_archive["summary"]["has_signoff_evidence"])
self.assertTrue(signoff_archive["summary"]["has_archive_evidence"])
self.assertEqual(signoff_archive["summary"]["total_evidence_count"], 2)
self.assertEqual(signoff_archive["signoff"]["task"]["title"], "签署确认")
self.assertEqual(signoff_archive["archive"]["task"]["title"], "归档确认")
self.assertEqual(signoff_archive["items"][0]["evidence_file"]["file_name"], "signed-contract.pdf")
listed = self.service.list_generated_contracts(self.current_user, page=1, page_size=10)
listed_item = next(item for item in listed["items"] if item["id"] == "contract-signoff-archive-001")
self.assertEqual(listed_item["signoff_archive"]["summary"]["signoff_evidence_count"], 1)
self.assertEqual(listed_item["signoff_archive"]["summary"]["archive_evidence_count"], 1)
def test_notification_service_builds_contract_todo_notifications(self):
from app.services.contract_approval_service import ContractApprovalService
from app.services.notification_service import NotificationService
approver = {
"id": "user-approver",
"tenant_id": "tenant-demo-001",
"display_name": "Approver",
"role": "member",
}
submitter = {
"id": "user-member",
"tenant_id": "tenant-demo-001",
"display_name": "Member",
"role": "member",
}
self._add_generated_contract("contract-notice-approval", generated_by="user-member", generated_by_name="Member")
approval_service = ContractApprovalService(self.db)
approval_service.submit_approval(
submitter,
contract_id="contract-notice-approval",
title="法务审批",
assignee_id="user-approver",
assignee_name="Approver",
)
service = NotificationService(self.db)
approver_notifications = service.list_notifications(approver)
self.assertGreaterEqual(approver_notifications["summary"]["mine"], 1)
self.assertTrue(any(item["type"] == "approval_pending" for item in approver_notifications["items"]))
approved_contract = self._add_generated_contract("contract-notice-signoff", lifecycle_status="approved")
approved_contract.generated_by = "user-member"
approved_contract.generated_by_name = "Member"
self.db.add(approved_contract)
self.db.commit()
expired_contract = self._add_generated_contract("contract-notice-archive", lifecycle_status="expired")
expired_contract.generated_by = "user-member"
expired_contract.generated_by_name = "Member"
expired_contract.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
self.db.add(expired_contract)
self.db.commit()
owner_notifications = service.list_notifications(submitter, limit=20)
notification_types = {item["type"] for item in owner_notifications["items"]}
self.assertIn("signoff_evidence_missing", notification_types)
self.assertIn("archive_evidence_missing", notification_types)
self.assertIn("renewal_reminder", notification_types)
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
self._add_generated_contract("contract-approval-001")
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)
self.assertEqual(pending_list["items"][0]["sla"]["status"], "normal")
self.assertIn("due_at", pending_list["items"][0]["sla"])
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"], "审批通过。")
self.assertEqual(decided["sla"]["status"], "closed")
self.assertEqual(detail["approval"]["next_action"]["code"], "approved")
self.assertEqual(detail["approval"]["timeline"][-1]["event"], "approved")
def test_contract_approval_sla_summary_counts_pending_steps(self):
from app.services.contract_approval_service import ContractApprovalService
now = datetime.now(timezone.utc)
self.db.add_all(
[
self.GeneratedContract(
id="contract-approval-sla-normal",
tenant_id="tenant-demo-001",
template_id="tpl-sla-normal",
template_name="SLA Normal Contract",
contract_type="procurement",
document_type_code="procurement_goods",
generated_by="user-admin",
generated_by_name="Admin",
export_format="docx",
file_id=None,
file_name="sla-normal.docx",
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
file_size=10,
generated_content="SLA normal contract content.",
variable_values_json={},
export_options_json={},
file_content_base64="",
status="generated",
created_at=now,
),
self.GeneratedContract(
id="contract-approval-sla-soon",
tenant_id="tenant-demo-001",
template_id="tpl-sla-soon",
template_name="SLA Soon 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="sla-soon.docx",
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
file_size=10,
generated_content="SLA soon contract content.",
variable_values_json={},
export_options_json={},
file_content_base64="",
status="generated",
created_at=now,
),
self.GeneratedContract(
id="contract-approval-sla-overdue",
tenant_id="tenant-demo-001",
template_id="tpl-sla-overdue",
template_name="SLA Overdue 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="sla-overdue.docx",
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
file_size=10,
generated_content="SLA overdue contract content.",
variable_values_json={},
export_options_json={},
file_content_base64="",
status="generated",
created_at=now,
),
self.ContractApprovalStep(
id="approval-sla-normal",
tenant_id="tenant-demo-001",
contract_id="contract-approval-sla-normal",
step_type="legal_review",
status="pending",
title="SLA normal",
submitter_id="user-admin",
submitter_name="Admin",
assignee_id="user-admin",
assignee_name="Admin",
created_at=now - timedelta(hours=2),
updated_at=now - timedelta(hours=2),
),
self.ContractApprovalStep(
id="approval-sla-soon",
tenant_id="tenant-demo-001",
contract_id="contract-approval-sla-soon",
step_type="legal_review",
status="pending",
title="SLA due soon",
submitter_id="user-admin",
submitter_name="Admin",
assignee_id="user-admin",
assignee_name="Admin",
created_at=now - timedelta(hours=30),
updated_at=now - timedelta(hours=30),
),
self.ContractApprovalStep(
id="approval-sla-overdue",
tenant_id="tenant-demo-001",
contract_id="contract-approval-sla-overdue",
step_type="legal_review",
status="pending",
title="SLA overdue",
submitter_id="user-admin",
submitter_name="Admin",
assignee_id="user-admin",
assignee_name="Admin",
created_at=now - timedelta(hours=60),
updated_at=now - timedelta(hours=60),
),
]
)
self.db.commit()
approval_service = ContractApprovalService(self.db)
result = approval_service.list_approvals(
current_user=self.current_user,
scope="assigned_to_me",
status_value="pending",
)
statuses = {item["id"]: item["sla"]["status"] for item in result["items"]}
self.assertEqual(statuses["approval-sla-normal"], "normal")
self.assertEqual(statuses["approval-sla-soon"], "due_soon")
self.assertEqual(statuses["approval-sla-overdue"], "overdue")
self.assertEqual(result["summary"]["sla_normal"], 1)
self.assertEqual(result["summary"]["sla_due_soon"], 1)
self.assertEqual(result["summary"]["sla_overdue"], 1)
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")
def test_contract_approval_reject_requires_reason_and_can_resubmit(self):
from app.services.contract_approval_service import ContractApprovalService
self._add_generated_contract("contract-approval-reject-001", generated_by="user-member", generated_by_name="Member")
approval_service = ContractApprovalService(self.db)
submitter = {"id": "user-member", "tenant_id": "tenant-demo-001", "display_name": "Member", "role": "member"}
approver = {"id": "user-legal", "tenant_id": "tenant-demo-001", "display_name": "Legal", "role": "member"}
submitted = approval_service.submit_approval(
current_user=submitter,
contract_id="contract-approval-reject-001",
title="采购合同法务审批",
assignee_id="user-legal",
assignee_name="Legal",
)
with self.assertRaises(ValidationError):
approval_service.decide_approval(
current_user=approver,
approval_id=submitted["id"],
decision="reject",
comment=" ",
)
rejected = approval_service.decide_approval(
current_user=approver,
approval_id=submitted["id"],
decision="reject",
comment="请补充验收标准后重新提交。",
)
self.assertEqual(rejected["status"], "rejected")
self.assertEqual(rejected["timeline"][-1]["event"], "rejected")
detail = self.service.get_generated_contract(submitter, "contract-approval-reject-001")
self.assertEqual(detail["approval"]["status"], "rejected")
self.assertTrue(detail["approval"]["can_submit"])
self.assertEqual(detail["approval"]["latest_rejected"]["comment"], "请补充验收标准后重新提交。")
self.assertEqual(detail["approval"]["next_action"]["code"], "revise_and_resubmit")
resubmitted = approval_service.submit_approval(
current_user=submitter,
contract_id="contract-approval-reject-001",
title="采购合同重新提交",
comment="已补充验收标准,请复核。",
assignee_id="user-legal",
assignee_name="Legal",
)
self.assertEqual(resubmitted["status"], "pending")
detail_after_resubmit = self.service.get_generated_contract(submitter, "contract-approval-reject-001")
self.assertEqual(detail_after_resubmit["approval"]["status"], "pending")
self.assertEqual(detail_after_resubmit["approval"]["total"], 2)
self.assertEqual(detail_after_resubmit["approval"]["timeline"][-1]["event"], "pending")
def test_contract_approval_role_boundaries_allow_participants_only(self):
from app.services.contract_approval_service import ContractApprovalService
self._add_generated_contract("contract-approval-boundary-001", generated_by="user-member", generated_by_name="Member")
approval_service = ContractApprovalService(self.db)
submitter = {"id": "user-member", "tenant_id": "tenant-demo-001", "display_name": "Member", "role": "member"}
approver = {"id": "user-legal", "tenant_id": "tenant-demo-001", "display_name": "Legal", "role": "member"}
outsider = {"id": "user-outsider", "tenant_id": "tenant-demo-001", "display_name": "Outsider", "role": "member"}
submitted = approval_service.submit_approval(
current_user=submitter,
contract_id="contract-approval-boundary-001",
title="采购合同法务审批",
assignee_id="user-legal",
assignee_name="Legal",
)
approver_detail = self.service.get_generated_contract(approver, "contract-approval-boundary-001")
self.assertTrue(approver_detail["approval"]["pending"]["can_decide"])
with self.assertRaises(NotFoundError):
self.service.update_generated_contract_lifecycle(
current_user=approver,
contract_id="contract-approval-boundary-001",
lifecycle_status="approved",
)
with self.assertRaises(NotFoundError):
self.service.get_generated_contract(outsider, "contract-approval-boundary-001")
with self.assertRaises(NotFoundError):
approval_service.decide_approval(
current_user=outsider,
approval_id=submitted["id"],
decision="approve",
comment="越权处理。",
)
def test_contract_approval_legal_role_can_view_all_and_decide(self):
from app.services.contract_approval_service import ContractApprovalService
self._add_generated_contract("contract-approval-legal-001", generated_by="user-member", generated_by_name="Member")
approval_service = ContractApprovalService(self.db)
submitter = {"id": "user-member", "tenant_id": "tenant-demo-001", "display_name": "Member", "role": "member"}
legal_user = {"id": "user-legal", "tenant_id": "tenant-demo-001", "display_name": "Legal", "role": "legal"}
submitted = approval_service.submit_approval(
current_user=submitter,
contract_id="contract-approval-legal-001",
title="采购合同法务审批",
assignee_id="user-other-approver",
assignee_name="Other Approver",
)
listed = approval_service.list_approvals(current_user=legal_user, scope="all")
decided = approval_service.decide_approval(
current_user=legal_user,
approval_id=submitted["id"],
decision="approve",
comment="法务通过。",
)
self.assertEqual(listed["total"], 1)
self.assertEqual(decided["status"], "approved")
def test_contract_approval_observer_is_readonly(self):
from app.services.contract_approval_service import ContractApprovalService
self._add_generated_contract("contract-approval-observer-001", generated_by="user-member", generated_by_name="Member")
approval_service = ContractApprovalService(self.db)
submitter = {"id": "user-member", "tenant_id": "tenant-demo-001", "display_name": "Member", "role": "member"}
observer = {"id": "user-observer", "tenant_id": "tenant-demo-001", "display_name": "Observer", "role": "observer"}
submitted = approval_service.submit_approval(
current_user=submitter,
contract_id="contract-approval-observer-001",
title="采购合同法务审批",
assignee_id="user-member",
assignee_name="Member",
)
listed = approval_service.list_approvals(current_user=observer, scope="all")
self.assertEqual(listed["total"], 1)
with self.assertRaises(NotFoundError):
approval_service.submit_approval(
current_user=observer,
contract_id="contract-approval-observer-001",
title="观察者提交审批",
)
with self.assertRaises(ForbiddenError):
approval_service.decide_approval(
current_user=observer,
approval_id=submitted["id"],
decision="approve",
comment="观察者越权处理。",
)
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))