854 lines
34 KiB
Python
854 lines
34 KiB
Python
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.api.tenant_settings import get_audit_log_service, get_tenant_service, require_current_user
|
|
from app.db import Base
|
|
from app.main import app
|
|
from app.models.audit_log import AuditLog
|
|
from app.models.commercial_invoice import CommercialInvoice
|
|
from app.models.commercial_order import CommercialOrder
|
|
from app.models.file import File
|
|
from app.models.generated_contract import GeneratedContract
|
|
from app.models.knowledge_base import KnowledgeBase
|
|
from app.models.knowledge_base_document import KnowledgeBaseDocument
|
|
from app.models.message import Message
|
|
from app.models.review_task import ReviewTask
|
|
from app.services.tenant_service import TenantService
|
|
|
|
|
|
def create_tenant_service_for_validation():
|
|
engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, future=True)()
|
|
service = TenantService(session)
|
|
service.ensure_default_tenant()
|
|
return service
|
|
|
|
|
|
class StubAuditLogService:
|
|
def __init__(self):
|
|
self.events = []
|
|
|
|
def record_event(self, **kwargs):
|
|
self.events.append(kwargs)
|
|
return kwargs
|
|
|
|
|
|
class StubTenantService:
|
|
def get_current_settings(self, current_user):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return {
|
|
"id": current_user["tenant_id"],
|
|
"name": "演示企业",
|
|
"code": "demo",
|
|
"status": "active",
|
|
"plan_name": "professional",
|
|
"max_users": 20,
|
|
"max_storage_mb": 10240,
|
|
"created_at": "2026-04-16T00:00:00+00:00",
|
|
"updated_at": "2026-04-16T00:00:00+00:00",
|
|
}
|
|
|
|
def update_current_settings(self, current_user, name, code, status_value):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return {
|
|
"id": current_user["tenant_id"],
|
|
"name": name,
|
|
"code": code,
|
|
"status": status_value,
|
|
"plan_name": "professional",
|
|
"max_users": 20,
|
|
"max_storage_mb": 10240,
|
|
"created_at": "2026-04-16T00:00:00+00:00",
|
|
"updated_at": "2026-04-16T00:00:00+00:00",
|
|
}
|
|
|
|
def get_current_billing(self, current_user):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return {
|
|
"tenant": {"id": current_user["tenant_id"], "name": "婕旂ず浼佷笟"},
|
|
"plan": {"code": "trial", "billing_status": "trialing", "trial_days_remaining": 7},
|
|
"metrics": [
|
|
{
|
|
"code": "reviews",
|
|
"label": "合同审查次数",
|
|
"used": 12,
|
|
"quota": 30,
|
|
"unit": "次",
|
|
"ratio": 0.4,
|
|
"percentage": 40,
|
|
"status": "ok",
|
|
}
|
|
],
|
|
"warnings": [],
|
|
"summary": {"overall_status": "healthy", "warning_count": 0},
|
|
"upgrade_action": {"href": "/pricing"},
|
|
}
|
|
|
|
def list_commercial_orders(self, current_user, limit=50):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return [
|
|
{
|
|
"id": "commercial-order-1",
|
|
"tenant_id": current_user["tenant_id"],
|
|
"requested_by": current_user["id"],
|
|
"order_no": "ORD-202605180001",
|
|
"order_type": "subscription",
|
|
"status": "pending_payment",
|
|
"plan_code": "team",
|
|
"plan_name": "团队版",
|
|
"billing_cycle": "monthly",
|
|
"amount_cents": 199900,
|
|
"currency": "CNY",
|
|
"payment_provider": "offline",
|
|
"invoice_status": "requested",
|
|
"contact_name": "demo",
|
|
"contact_phone": "13800000000",
|
|
"contact_email": "demo@example.com",
|
|
"note": "upgrade",
|
|
"review_quota_monthly": 800,
|
|
"generation_quota_monthly": 500,
|
|
"knowledge_base_quota": 5,
|
|
"storage_quota_mb": 10240,
|
|
"max_users": 20,
|
|
"max_storage_mb": 10240,
|
|
"paid_at": None,
|
|
"activated_at": None,
|
|
"canceled_at": None,
|
|
"created_at": "2026-05-18T00:00:00+00:00",
|
|
"updated_at": "2026-05-18T00:00:00+00:00",
|
|
}
|
|
][:limit]
|
|
|
|
def list_commercial_invoices(self, current_user, limit=50):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return [
|
|
{
|
|
"id": "commercial-invoice-1",
|
|
"tenant_id": current_user["tenant_id"],
|
|
"order_id": "commercial-order-1",
|
|
"requested_by": current_user["id"],
|
|
"issued_by": None,
|
|
"invoice_type": "vat_general",
|
|
"status": "requested",
|
|
"invoice_title": "示例企业",
|
|
"tax_number": "91310000TEST000001",
|
|
"registered_address": "上海市",
|
|
"registered_phone": "021-00000000",
|
|
"bank_name": "测试银行",
|
|
"bank_account": "6222000000000000",
|
|
"receiver_email": "finance@example.com",
|
|
"receiver_phone": "13800000000",
|
|
"amount_cents": 199900,
|
|
"currency": "CNY",
|
|
"invoice_no": None,
|
|
"invoice_file_id": None,
|
|
"note": "待开票",
|
|
"issued_at": None,
|
|
"created_at": "2026-05-18T00:00:00+00:00",
|
|
"updated_at": "2026-05-18T00:00:00+00:00",
|
|
}
|
|
][:limit]
|
|
|
|
def create_commercial_order(self, current_user, **kwargs):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return {
|
|
"id": "commercial-order-created",
|
|
"tenant_id": current_user["tenant_id"],
|
|
"requested_by": current_user["id"],
|
|
"order_no": "ORD-202605180002",
|
|
"order_type": kwargs["order_type"],
|
|
"status": "pending_payment",
|
|
"plan_code": kwargs["plan_code"],
|
|
"plan_name": kwargs.get("plan_name") or "团队版",
|
|
"billing_cycle": kwargs["billing_cycle"],
|
|
"amount_cents": kwargs["amount_cents"],
|
|
"currency": kwargs["currency"],
|
|
"payment_provider": kwargs["payment_provider"],
|
|
"invoice_status": "requested" if kwargs["invoice_required"] else "not_requested",
|
|
"contact_name": kwargs.get("contact_name"),
|
|
"contact_phone": kwargs.get("contact_phone"),
|
|
"contact_email": kwargs.get("contact_email"),
|
|
"note": kwargs.get("note"),
|
|
"review_quota_monthly": kwargs.get("review_quota_monthly"),
|
|
"generation_quota_monthly": kwargs.get("generation_quota_monthly"),
|
|
"knowledge_base_quota": kwargs.get("knowledge_base_quota"),
|
|
"storage_quota_mb": kwargs.get("storage_quota_mb"),
|
|
"max_users": kwargs.get("max_users"),
|
|
"max_storage_mb": kwargs.get("max_storage_mb"),
|
|
"paid_at": None,
|
|
"activated_at": None,
|
|
"canceled_at": None,
|
|
"created_at": "2026-05-18T00:00:00+00:00",
|
|
"updated_at": "2026-05-18T00:00:00+00:00",
|
|
}
|
|
|
|
def activate_commercial_order(self, current_user, order_id, payment_provider="offline"):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return {
|
|
"order": {
|
|
"id": order_id,
|
|
"tenant_id": current_user["tenant_id"],
|
|
"requested_by": current_user["id"],
|
|
"order_no": "ORD-202605180002",
|
|
"order_type": "subscription",
|
|
"status": "activated",
|
|
"plan_code": "team",
|
|
"plan_name": "团队版",
|
|
"billing_cycle": "monthly",
|
|
"amount_cents": 199900,
|
|
"currency": "CNY",
|
|
"payment_provider": payment_provider,
|
|
"invoice_status": "requested",
|
|
"contact_name": "demo",
|
|
"contact_phone": "13800000000",
|
|
"contact_email": "demo@example.com",
|
|
"note": "paid",
|
|
"review_quota_monthly": 800,
|
|
"generation_quota_monthly": 500,
|
|
"knowledge_base_quota": 5,
|
|
"storage_quota_mb": 10240,
|
|
"max_users": 20,
|
|
"max_storage_mb": 10240,
|
|
"paid_at": "2026-05-18T00:00:00+00:00",
|
|
"activated_at": "2026-05-18T00:00:00+00:00",
|
|
"canceled_at": None,
|
|
"created_at": "2026-05-18T00:00:00+00:00",
|
|
"updated_at": "2026-05-18T00:00:00+00:00",
|
|
},
|
|
"tenant": {
|
|
"id": current_user["tenant_id"],
|
|
"name": "demo",
|
|
"code": "demo",
|
|
"status": "active",
|
|
"plan_name": "团队版",
|
|
"plan_code": "team",
|
|
"billing_status": "active",
|
|
"trial_started_at": None,
|
|
"trial_ends_at": None,
|
|
"review_quota_monthly": 800,
|
|
"generation_quota_monthly": 500,
|
|
"knowledge_base_quota": 5,
|
|
"storage_quota_mb": 10240,
|
|
"max_users": 20,
|
|
"max_storage_mb": 10240,
|
|
"created_at": "2026-05-18T00:00:00+00:00",
|
|
"updated_at": "2026-05-18T00:00:00+00:00",
|
|
},
|
|
}
|
|
|
|
def request_commercial_invoice(self, current_user, order_id, **kwargs):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return {
|
|
"id": "commercial-invoice-created",
|
|
"tenant_id": current_user["tenant_id"],
|
|
"order_id": order_id,
|
|
"requested_by": current_user["id"],
|
|
"issued_by": None,
|
|
"invoice_type": kwargs["invoice_type"],
|
|
"status": "requested",
|
|
"invoice_title": kwargs["invoice_title"],
|
|
"tax_number": kwargs["tax_number"],
|
|
"registered_address": kwargs.get("registered_address"),
|
|
"registered_phone": kwargs.get("registered_phone"),
|
|
"bank_name": kwargs.get("bank_name"),
|
|
"bank_account": kwargs.get("bank_account"),
|
|
"receiver_email": kwargs.get("receiver_email"),
|
|
"receiver_phone": kwargs.get("receiver_phone"),
|
|
"amount_cents": 199900,
|
|
"currency": "CNY",
|
|
"invoice_no": None,
|
|
"invoice_file_id": None,
|
|
"note": kwargs.get("note"),
|
|
"issued_at": None,
|
|
"created_at": "2026-05-18T00:00:00+00:00",
|
|
"updated_at": "2026-05-18T00:00:00+00:00",
|
|
}
|
|
|
|
def issue_commercial_invoice(self, current_user, invoice_id, invoice_no, invoice_file_id=None, note=None):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return {
|
|
"id": invoice_id,
|
|
"tenant_id": current_user["tenant_id"],
|
|
"order_id": "commercial-order-1",
|
|
"requested_by": current_user["id"],
|
|
"issued_by": current_user["id"],
|
|
"invoice_type": "vat_general",
|
|
"status": "issued",
|
|
"invoice_title": "示例企业",
|
|
"tax_number": "91310000TEST000001",
|
|
"registered_address": "上海市",
|
|
"registered_phone": "021-00000000",
|
|
"bank_name": "测试银行",
|
|
"bank_account": "6222000000000000",
|
|
"receiver_email": "finance@example.com",
|
|
"receiver_phone": "13800000000",
|
|
"amount_cents": 199900,
|
|
"currency": "CNY",
|
|
"invoice_no": invoice_no,
|
|
"invoice_file_id": invoice_file_id,
|
|
"note": note,
|
|
"issued_at": "2026-05-18T00:00:00+00:00",
|
|
"created_at": "2026-05-18T00:00:00+00:00",
|
|
"updated_at": "2026-05-18T00:00:00+00:00",
|
|
}
|
|
|
|
def update_current_billing_plan(self, current_user, **kwargs):
|
|
if current_user["role"] != "admin":
|
|
raise HTTPException(status_code=403, detail="forbidden")
|
|
return {
|
|
"id": current_user["tenant_id"],
|
|
"name": "demo",
|
|
"code": "demo",
|
|
"status": "active",
|
|
"plan_name": kwargs["plan_name"],
|
|
"plan_code": kwargs["plan_code"],
|
|
"billing_status": kwargs["billing_status"],
|
|
"trial_started_at": None,
|
|
"trial_ends_at": kwargs["trial_ends_at"].isoformat() if kwargs.get("trial_ends_at") else None,
|
|
"review_quota_monthly": kwargs.get("review_quota_monthly"),
|
|
"generation_quota_monthly": kwargs.get("generation_quota_monthly"),
|
|
"knowledge_base_quota": kwargs.get("knowledge_base_quota"),
|
|
"storage_quota_mb": kwargs.get("storage_quota_mb"),
|
|
"max_users": kwargs.get("max_users"),
|
|
"max_storage_mb": kwargs.get("max_storage_mb"),
|
|
"created_at": "2026-05-18T00:00:00+00:00",
|
|
"updated_at": "2026-05-18T00:00:00+00:00",
|
|
}
|
|
|
|
|
|
class TenantSettingsApiTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.audit_log_service = StubAuditLogService()
|
|
app.dependency_overrides[get_tenant_service] = lambda: StubTenantService()
|
|
app.dependency_overrides[get_audit_log_service] = lambda: self.audit_log_service
|
|
self.client = TestClient(app)
|
|
|
|
def tearDown(self):
|
|
app.dependency_overrides.clear()
|
|
|
|
def test_admin_can_get_tenant_settings(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.get("/api/v1/tenant-settings/me")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"]["code"], "demo")
|
|
|
|
def test_admin_can_get_tenant_billing(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.get("/api/v1/tenant-settings/billing")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()["data"]
|
|
self.assertEqual(body["plan"]["code"], "trial")
|
|
self.assertEqual(body["metrics"][0]["code"], "reviews")
|
|
|
|
def test_admin_can_list_commercial_orders(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.get("/api/v1/tenant-settings/billing/orders")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"][0]["order_no"], "ORD-202605180001")
|
|
|
|
def test_admin_can_list_commercial_invoices(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.get("/api/v1/tenant-settings/billing/invoices")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"][0]["invoice_title"], "示例企业")
|
|
|
|
def test_admin_can_create_commercial_order(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/api/v1/tenant-settings/billing/orders",
|
|
data={
|
|
"order_type": "subscription",
|
|
"plan_code": "team",
|
|
"plan_name": "团队版",
|
|
"billing_cycle": "monthly",
|
|
"amount_cents": "199900",
|
|
"payment_provider": "offline",
|
|
"invoice_required": "true",
|
|
"review_quota_monthly": "800",
|
|
"generation_quota_monthly": "500",
|
|
"knowledge_base_quota": "5",
|
|
"storage_quota_mb": "10240",
|
|
"max_users": "20",
|
|
"max_storage_mb": "10240",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()["data"]
|
|
self.assertEqual(body["status"], "pending_payment")
|
|
self.assertEqual(body["invoice_status"], "requested")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "create_commercial_order")
|
|
|
|
def test_admin_can_activate_commercial_order(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/api/v1/tenant-settings/billing/orders/commercial-order-created/activate",
|
|
data={"payment_provider": "bank_transfer"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()["data"]
|
|
self.assertEqual(body["order"]["status"], "activated")
|
|
self.assertEqual(body["tenant"]["billing_status"], "active")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "activate_commercial_order")
|
|
|
|
def test_admin_can_request_commercial_invoice(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/api/v1/tenant-settings/billing/orders/commercial-order-created/invoice",
|
|
data={
|
|
"invoice_type": "vat_general",
|
|
"invoice_title": "示例企业",
|
|
"tax_number": "91310000TEST000001",
|
|
"registered_address": "上海市",
|
|
"registered_phone": "021-00000000",
|
|
"bank_name": "测试银行",
|
|
"bank_account": "6222000000000000",
|
|
"receiver_email": "finance@example.com",
|
|
"receiver_phone": "13800000000",
|
|
"note": "请开电子普票",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()["data"]
|
|
self.assertEqual(body["status"], "requested")
|
|
self.assertEqual(body["tax_number"], "91310000TEST000001")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "request_commercial_invoice")
|
|
|
|
def test_admin_can_issue_commercial_invoice(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.post(
|
|
"/api/v1/tenant-settings/billing/invoices/commercial-invoice-created/issue",
|
|
data={"invoice_no": "INV-20260518-001", "invoice_file_id": "file-invoice-1", "note": "已开票"},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()["data"]
|
|
self.assertEqual(body["status"], "issued")
|
|
self.assertEqual(body["invoice_no"], "INV-20260518-001")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "issue_commercial_invoice")
|
|
|
|
def test_admin_can_update_billing_plan(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.put(
|
|
"/api/v1/tenant-settings/billing/plan",
|
|
data={
|
|
"plan_code": "professional",
|
|
"plan_name": "专业版",
|
|
"billing_status": "active",
|
|
"review_quota_monthly": "4000",
|
|
"generation_quota_monthly": "2000",
|
|
"knowledge_base_quota": "20",
|
|
"storage_quota_mb": "51200",
|
|
"max_users": "80",
|
|
"max_storage_mb": "51200",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()["data"]
|
|
self.assertEqual(body["plan_code"], "professional")
|
|
self.assertEqual(body["billing_status"], "active")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "update_commercial_billing_plan")
|
|
|
|
def test_admin_can_update_tenant_settings(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-admin",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "admin",
|
|
}
|
|
|
|
response = self.client.put(
|
|
"/api/v1/tenant-settings/me",
|
|
data={
|
|
"name": "演示企业集团",
|
|
"code": "demo-group",
|
|
"status": "disabled",
|
|
},
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
body = response.json()
|
|
self.assertEqual(body["data"]["name"], "演示企业集团")
|
|
self.assertEqual(body["data"]["status"], "disabled")
|
|
self.assertEqual(self.audit_log_service.events[0]["action"], "update_tenant_settings")
|
|
self.assertEqual(
|
|
self.audit_log_service.events[0]["detail_json"],
|
|
{
|
|
"name": "演示企业集团",
|
|
"code": "demo-group",
|
|
"status": "disabled",
|
|
"plan_name": "professional",
|
|
"max_users": 20,
|
|
"max_storage_mb": 10240,
|
|
},
|
|
)
|
|
|
|
def test_rejects_blank_tenant_identity_fields(self):
|
|
service = create_tenant_service_for_validation()
|
|
|
|
with self.assertRaises(HTTPException) as name_context:
|
|
service.update_current_settings(
|
|
{"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"},
|
|
" ",
|
|
"demo",
|
|
"active",
|
|
)
|
|
self.assertEqual(name_context.exception.status_code, 400)
|
|
self.assertEqual(name_context.exception.detail, "tenant name is required")
|
|
|
|
with self.assertRaises(HTTPException) as code_context:
|
|
service.update_current_settings(
|
|
{"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"},
|
|
"演示企业",
|
|
" ",
|
|
"active",
|
|
)
|
|
self.assertEqual(code_context.exception.status_code, 400)
|
|
self.assertEqual(code_context.exception.detail, "tenant code is required")
|
|
|
|
def test_member_cannot_access_tenant_settings(self):
|
|
app.dependency_overrides[require_current_user] = lambda: {
|
|
"id": "user-member",
|
|
"tenant_id": "tenant-demo-001",
|
|
"role": "member",
|
|
}
|
|
|
|
response = self.client.get("/api/v1/tenant-settings/me")
|
|
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertEqual(response.json()["message"], "forbidden")
|
|
|
|
def test_billing_summary_uses_current_month_usage_and_warnings(self):
|
|
engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, future=True)()
|
|
try:
|
|
service = TenantService(session)
|
|
service.ensure_default_tenant()
|
|
tenant_id = "tenant-demo-001"
|
|
user_id = "user-admin"
|
|
now = datetime.now(timezone.utc)
|
|
session.add_all(
|
|
[
|
|
ReviewTask(
|
|
id=f"review-{index}",
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
title=f"审查-{index}",
|
|
created_at=now,
|
|
)
|
|
for index in range(30)
|
|
]
|
|
)
|
|
session.add(
|
|
GeneratedContract(
|
|
id="contract-1",
|
|
tenant_id=tenant_id,
|
|
template_id="tpl-1",
|
|
template_name="NDA",
|
|
generated_by=user_id,
|
|
export_format="docx",
|
|
file_name="nda.docx",
|
|
mime_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
|
|
file_size=1024,
|
|
generated_content="content",
|
|
file_content_base64="Y29udGVudA==",
|
|
created_at=now,
|
|
)
|
|
)
|
|
session.add(
|
|
File(
|
|
id="file-1",
|
|
tenant_id=tenant_id,
|
|
uploaded_by=user_id,
|
|
bucket="contracts",
|
|
object_key="contracts/file-1",
|
|
file_name="contract.pdf",
|
|
mime_type="application/pdf",
|
|
file_size=1024 * 1024,
|
|
file_hash="hash",
|
|
file_type="contract",
|
|
created_at=now,
|
|
)
|
|
)
|
|
session.add(
|
|
KnowledgeBase(
|
|
id="kb-1",
|
|
tenant_id=tenant_id,
|
|
name="合同知识库",
|
|
status="active",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
)
|
|
session.add(
|
|
Message(
|
|
id="msg-1",
|
|
tenant_id=tenant_id,
|
|
conversation_id="conv-1",
|
|
user_id=user_id,
|
|
role="user",
|
|
content="违约金条款是否合理？",
|
|
created_at=now,
|
|
)
|
|
)
|
|
session.add(
|
|
AuditLog(
|
|
id="audit-1",
|
|
tenant_id=tenant_id,
|
|
user_id=user_id,
|
|
action="download_report",
|
|
resource_type="review",
|
|
resource_id="review-1",
|
|
created_at=now,
|
|
)
|
|
)
|
|
tenant = service.repository.get_by_id(tenant_id)
|
|
tenant.trial_ends_at = now - timedelta(days=1)
|
|
session.commit()
|
|
|
|
summary = service.get_current_billing({"id": user_id, "tenant_id": tenant_id, "role": "admin"})
|
|
|
|
metrics = {item["code"]: item for item in summary["metrics"]}
|
|
self.assertEqual(metrics["reviews"]["used"], 30)
|
|
self.assertEqual(metrics["reviews"]["status"], "exceeded")
|
|
self.assertEqual(metrics["generations"]["used"], 1)
|
|
self.assertEqual(metrics["storage"]["used"], 1.0)
|
|
self.assertEqual(metrics["knowledge_bases"]["used"], 1)
|
|
self.assertEqual(metrics["chat_messages"]["used"], 1)
|
|
self.assertEqual(metrics["report_downloads"]["used"], 1)
|
|
self.assertEqual(summary["summary"]["overall_status"], "blocked")
|
|
self.assertTrue(any(warning["code"] == "trial_expired" for warning in summary["warnings"]))
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
def test_billing_guard_blocks_expired_trial(self):
|
|
service = create_tenant_service_for_validation()
|
|
tenant = service.repository.get_by_id("tenant-demo-001")
|
|
tenant.trial_ends_at = datetime.now(timezone.utc) - timedelta(days=1)
|
|
service.db.commit()
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
service.assert_billing_allows({"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"}, "review")
|
|
|
|
self.assertEqual(context.exception.status_code, 402)
|
|
self.assertIn("试用已到期", context.exception.detail)
|
|
|
|
def test_billing_guard_blocks_review_over_quota(self):
|
|
engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, future=True)()
|
|
try:
|
|
service = TenantService(session)
|
|
service.ensure_default_tenant()
|
|
tenant = service.repository.get_by_id("tenant-demo-001")
|
|
tenant.trial_ends_at = datetime.now(timezone.utc) + timedelta(days=7)
|
|
session.add_all(
|
|
[
|
|
ReviewTask(
|
|
id=f"quota-review-{index}",
|
|
tenant_id="tenant-demo-001",
|
|
user_id="user-admin",
|
|
title=f"审查-{index}",
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
for index in range(30)
|
|
]
|
|
)
|
|
session.commit()
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
service.assert_billing_allows({"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"}, "review")
|
|
|
|
self.assertEqual(context.exception.status_code, 402)
|
|
self.assertIn("额度已用完", context.exception.detail)
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
def test_billing_guard_counts_knowledge_document_storage(self):
|
|
engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, future=True)()
|
|
try:
|
|
service = TenantService(session)
|
|
service.ensure_default_tenant()
|
|
tenant = service.repository.get_by_id("tenant-demo-001")
|
|
tenant.trial_ends_at = datetime.now(timezone.utc) + timedelta(days=7)
|
|
tenant.storage_quota_mb = 1
|
|
session.add(
|
|
KnowledgeBaseDocument(
|
|
id="kbdoc-storage-1",
|
|
knowledge_base_id="kb-1",
|
|
tenant_id="tenant-demo-001",
|
|
file_name="rules.txt",
|
|
mime_type="text/plain",
|
|
file_size=1024 * 1024,
|
|
status="indexed",
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
with self.assertRaises(HTTPException) as context:
|
|
service.assert_billing_allows(
|
|
{"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"},
|
|
"storage",
|
|
requested_file_size_bytes=1,
|
|
)
|
|
|
|
self.assertEqual(context.exception.status_code, 402)
|
|
self.assertIn("存储容量不足", context.exception.detail)
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
def test_commercial_order_activation_updates_tenant_plan_and_quota(self):
|
|
engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, future=True)()
|
|
try:
|
|
service = TenantService(session)
|
|
service.ensure_default_tenant()
|
|
current_user = {"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"}
|
|
|
|
order = service.create_commercial_order(
|
|
current_user=current_user,
|
|
order_type="subscription",
|
|
plan_code="team",
|
|
billing_cycle="monthly",
|
|
amount_cents=199900,
|
|
payment_provider="offline",
|
|
invoice_required=True,
|
|
)
|
|
result = service.activate_commercial_order(current_user, order["id"], payment_provider="bank_transfer")
|
|
|
|
self.assertEqual(result["order"]["status"], "activated")
|
|
self.assertEqual(result["order"]["payment_provider"], "bank_transfer")
|
|
self.assertEqual(result["tenant"]["plan_code"], "team")
|
|
self.assertEqual(result["tenant"]["billing_status"], "active")
|
|
self.assertEqual(result["tenant"]["review_quota_monthly"], 800)
|
|
self.assertEqual(result["tenant"]["generation_quota_monthly"], 500)
|
|
self.assertEqual(result["tenant"]["knowledge_base_quota"], 5)
|
|
self.assertEqual(result["tenant"]["storage_quota_mb"], 10240)
|
|
self.assertEqual(session.query(CommercialOrder).count(), 1)
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
def test_commercial_invoice_request_and_issue_updates_order_invoice_status(self):
|
|
engine = create_engine("sqlite:///:memory:", future=True)
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine, future=True)()
|
|
try:
|
|
service = TenantService(session)
|
|
service.ensure_default_tenant()
|
|
current_user = {"id": "user-admin", "tenant_id": "tenant-demo-001", "role": "admin"}
|
|
order = service.create_commercial_order(
|
|
current_user=current_user,
|
|
order_type="subscription",
|
|
plan_code="team",
|
|
billing_cycle="monthly",
|
|
amount_cents=199900,
|
|
payment_provider="offline",
|
|
invoice_required=True,
|
|
)
|
|
invoice = service.request_commercial_invoice(
|
|
current_user=current_user,
|
|
order_id=order["id"],
|
|
invoice_type="vat_general",
|
|
invoice_title="示例企业",
|
|
tax_number="91310000TEST000001",
|
|
receiver_email="finance@example.com",
|
|
)
|
|
stored_order = session.query(CommercialOrder).filter(CommercialOrder.id == order["id"]).first()
|
|
self.assertEqual(invoice["status"], "requested")
|
|
self.assertEqual(stored_order.invoice_status, "requested")
|
|
|
|
service.activate_commercial_order(current_user, order["id"], payment_provider="bank_transfer")
|
|
issued = service.issue_commercial_invoice(
|
|
current_user=current_user,
|
|
invoice_id=invoice["id"],
|
|
invoice_no="INV-20260518-001",
|
|
invoice_file_id="file-invoice-1",
|
|
)
|
|
stored_order = session.query(CommercialOrder).filter(CommercialOrder.id == order["id"]).first()
|
|
|
|
self.assertEqual(issued["status"], "issued")
|
|
self.assertEqual(issued["invoice_no"], "INV-20260518-001")
|
|
self.assertEqual(stored_order.invoice_status, "issued")
|
|
self.assertEqual(session.query(CommercialInvoice).count(), 1)
|
|
finally:
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|