From f9c206c91d2c3b280e77a75baa6d844034abf2de Mon Sep 17 00:00:00 2001 From: wuzilong <704541015@qq.com> Date: Sun, 1 Nov 2026 21:37:59 +0800 Subject: [PATCH] feat(pl): Add scoped quota statistics to distinguish shared balances from contract allocations. --- backend/app/api/pl/quota_cert.py | 31 +++ backend/app/services/pl_quota_statistics.py | 197 ++++++++++++++++++ backend/tests/test_pl_quota_api.py | 23 +++ backend/tests/test_pl_quota_statistics.py | 218 ++++++++++++++++++++ 4 files changed, 469 insertions(+) create mode 100644 backend/app/services/pl_quota_statistics.py create mode 100644 backend/tests/test_pl_quota_statistics.py diff --git a/backend/app/api/pl/quota_cert.py b/backend/app/api/pl/quota_cert.py index 32191dfb0..7ada5f120 100644 --- a/backend/app/api/pl/quota_cert.py +++ b/backend/app/api/pl/quota_cert.py @@ -27,6 +27,7 @@ from app.services import pl_stop_legal_document as stop_documents from app.services import pl_quota_planning as planning from app.services import pl_quota_adjustment as adjustment from app.services import pl_quota_overview as overview +from app.services import pl_quota_statistics as statistics from app.services.pl_quota_allocation import AllocationIncomplete, InsufficientQuota from app.services.pl_quota_attachments import read_image from app.services.pl_quota_recognition import recognize_quota_cert @@ -145,6 +146,36 @@ def save_bindings(contract_id: str, body: ContractCertBindingsBody, return _write(db, lambda: scopes.save_contract_bindings(db, user, contract_id, body)) +@router.get('/statistics/contracts') +def contract_statistics_list(keyword: str = Query('', max_length=200), + page: int = Query(1, ge=1), + page_size: int = Query(20, alias='pageSize', ge=1, le=100), + db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)): + return _wire(statistics.list_contracts(db, user, keyword=keyword, page=page, page_size=page_size)) + + +@router.get('/statistics/certificates') +def certificate_statistics_list(keyword: str = Query('', max_length=200), + cert_type: Optional[CertType] = Query(None, alias='certType'), + page: int = Query(1, ge=1), + page_size: int = Query(20, alias='pageSize', ge=1, le=100), + db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)): + return _wire(statistics.list_certificates(db, user, keyword=keyword, cert_type=cert_type, + page=page, page_size=page_size)) + + +@router.get('/contracts/{contract_id}/statistics') +def contract_statistics(contract_id: str, db: Session = Depends(get_db), + user: DbUserContext = Depends(current_user)): + return _wire(statistics.contract_statistics(db, user, contract_id)) + + +@router.get('/{cert_type}/{cert_id}/statistics') +def certificate_statistics(cert_type: CertType, cert_id: str, db: Session = Depends(get_db), + user: DbUserContext = Depends(current_user)): + return _wire(statistics.certificate_statistics(db, user, cert_type, cert_id)) + + @router.get('/{cert_type}') def list_certs(cert_type: CertType, keyword: Optional[str] = None, status: Optional[str] = None, db: Session = Depends(get_db), user: DbUserContext = Depends(current_user)): diff --git a/backend/app/services/pl_quota_statistics.py b/backend/app/services/pl_quota_statistics.py new file mode 100644 index 000000000..bc2664f21 --- /dev/null +++ b/backend/app/services/pl_quota_statistics.py @@ -0,0 +1,197 @@ +"""只读额度统计：共享缓存与授权合同分配分开展示，不重建历史余额。""" +from decimal import Decimal +from sqlalchemy import and_, or_, literal, func +from fastapi import HTTPException +from app.crud.crm.data_scope_helpers import apply_crm_owner_scope +from app.models.crm_models import CrmContract, CrmContractProduct +from app.models.pl_models import PlPlan +from app.models.pl_quota_models import (PlContractCertLink as Link, PlContractCertProduct as Scope, + PlPlanCertUsage as Usage, PlPlanCertUsageItem as Item, PlCertQtyLog as Log) +from app.services.pl_data_scope import apply_pl_scope +from app.services.pl_quota_cert_service import MODELS, _get_row + +ZERO = Decimal('0') + + +def _active(db, user, model): + return db.query(model).filter(model.tenant_no == user.tenant_no, model.del_flag == 0) + + +def _contracts(db, user): + return apply_crm_owner_scope(_active(db, user, CrmContract), CrmContract, + user.to_user_info(), db, user.tenant_no) + + +def _certificates(db, user, keyword=None, cert_type=None): + queries = [] + for kind, model in MODELS.items(): + if cert_type and kind != cert_type: + continue + query = apply_pl_scope(_active(db, user, model), model, user, db, user.tenant_no) + if keyword: + query = query.filter(or_(model.cert_no.contains(keyword, autoescape=True), + model.product_name_on_cert.contains(keyword, autoescape=True))) + queries.append(query.with_entities(literal(kind).label('cert_type'), model.id.label('cert_id'), + *[getattr(model, field).label(field) for field in ('cert_no', 'product_name_on_cert', + 'quantity_unit', 'total_qty', 'reserved_qty', 'used_qty', 'status', + 'valid_from', 'valid_to', 'create_time')])) + return (queries[0].union_all(*queries[1:]) if len(queries) > 1 else queries[0]).subquery() + + +def _cert_summary(row): + return {'id': row.cert_id, 'certId': row.cert_id, 'certType': row.cert_type, 'certNo': row.cert_no, + 'productNameOnCert': row.product_name_on_cert, 'quantityUnit': row.quantity_unit, + 'totalQty': row.total_qty, 'reservedQty': row.reserved_qty, 'usedQty': row.used_qty, + 'availableQty': row.total_qty - row.reserved_qty - row.used_qty, 'status': row.status, + 'validFrom': row.valid_from, 'validTo': row.valid_to} + + +def _contract_summary(row): + return {'id': row.id, 'contractNo': row.contract_no, 'contractName': row.contract_name, + 'contractType': row.contract_type} + + +def _key_filter(model, keys): + return or_(*[and_(model.cert_type == kind, model.cert_id.in_( + [ident for k, ident in keys if k == kind])) for kind in MODELS]) + + +def _details(db, user, keys, contract_ids=None): + """有限证件/合同范围一次读取维护身份、一次读取授权使用；不连接明细造成乘积。""" + if not keys: + return {}, [] + contracts = _contracts(db, user).with_entities(CrmContract.id) + if contract_ids is not None: + contracts = contracts.filter(CrmContract.id.in_(contract_ids)) + links = _active(db, user, Link).filter(_key_filter(Link, keys), Link.contract_id.in_(contracts)).all() + groups = {} + for link in links: + groups[(link.cert_type, link.cert_id, link.contract_id)] = { + 'bindingId': link.id, 'linkStatus': link.link_status, 'scopeRevision': link.scope_revision, + 'contractReservedQty': ZERO, 'contractUsedQty': ZERO, 'products': []} + products = {} + if links: + rows = db.query(Scope, CrmContractProduct).join(CrmContractProduct, + Scope.contract_product_id == CrmContractProduct.id).filter( + Scope.tenant_no == user.tenant_no, Scope.del_flag == 0, + Scope.contract_cert_link_id.in_([l.id for l in links]), + CrmContractProduct.tenant_no == user.tenant_no, CrmContractProduct.del_flag == 0) + by_link = {l.id: l for l in links} + for scope, product in rows: + link = by_link[scope.contract_cert_link_id] + if product.contract_id != link.contract_id: + continue + key = (link.cert_type, link.cert_id, link.contract_id) + value = {'id': product.id, 'contractProductId': product.id, 'productId': product.product_id, + 'productName': product.product_name, 'model': product.model, 'unit': product.unit, + 'bindingId': link.id, 'scopeId': scope.id, 'scopeStatus': scope.scope_status, + 'linkStatus': link.link_status, 'reservedQty': ZERO, 'usedQty': ZERO} + groups[key]['products'].append(value) + products[(*key, product.id)] = value + plans = apply_pl_scope(_active(db, user, PlPlan), PlPlan, user, db, user.tenant_no).with_entities(PlPlan.id) + rows = db.query(Item, Usage, PlPlan, CrmContractProduct).join(Usage, Item.usage_id == Usage.id).join( + PlPlan, Usage.plan_id == PlPlan.id).join(CrmContractProduct, and_( + Item.contract_product_id == CrmContractProduct.id, Item.contract_id == CrmContractProduct.contract_id) + ).filter(Item.tenant_no == user.tenant_no, Item.del_flag == 0, + Usage.tenant_no == user.tenant_no, Usage.del_flag == 0, _key_filter(Usage, keys), + Item.contract_id.in_(contracts), PlPlan.id.in_(plans), + CrmContractProduct.tenant_no == user.tenant_no, CrmContractProduct.del_flag == 0 + ).order_by(Item.create_time.desc(), Item.id).all() + usage_items = [] + for item, usage, plan, product in rows: + key = (usage.cert_type, usage.cert_id, item.contract_id) + group = groups.setdefault(key, {'bindingId': None, 'linkStatus': None, 'scopeRevision': None, + 'contractReservedQty': ZERO, 'contractUsedQty': ZERO, 'products': []}) + # 证件单位不可变；异常历史单位不混入当前单位的聚合，明细仍保留追溯。 + unit = keys[key[:2]] + if usage.quantity_unit == unit: + group['contractReservedQty'] += item.reserved_qty + group['contractUsedQty'] += item.used_qty + value = products.get((*key, product.id)) + if value is None: + value = {'id': product.id, 'contractProductId': product.id, 'productId': product.product_id, + 'productName': product.product_name, 'model': product.model, 'unit': product.unit, + 'bindingId': None, 'scopeId': None, 'scopeStatus': None, 'linkStatus': None, + 'reservedQty': ZERO, 'usedQty': ZERO} + products[(*key, product.id)] = value + group['products'].append(value) + value['reservedQty'] += item.reserved_qty + value['usedQty'] += item.used_qty + usage_items.append({'id': item.id, 'usageId': usage.id, 'certType': usage.cert_type, + 'certId': usage.cert_id, 'planId': plan.id, 'planNo': plan.plan_no, + 'stopType': item.source_stop_type, 'stopId': item.source_stop_id, + 'stopProductId': item.source_stop_product_id, 'planProductId': item.plan_product_id, + 'contractId': item.contract_id, 'contractProductId': item.contract_product_id, + 'productName': product.product_name, 'quantityUnit': usage.quantity_unit, + 'reservedQty': item.reserved_qty, 'usedQty': item.used_qty, 'itemStatus': item.item_status}) + return groups, usage_items + + +def list_contracts(db, user, keyword=None, page=1, page_size=20): + certs = _certificates(db, user) + linked = _active(db, user, Link).join(certs, and_(Link.cert_type == certs.c.cert_type, + Link.cert_id == certs.c.cert_id)).with_entities(Link.contract_id) + query = _contracts(db, user).filter(CrmContract.id.in_(linked)) + if keyword: + query = query.filter(or_(CrmContract.contract_no.contains(keyword, autoescape=True), + CrmContract.contract_name.contains(keyword, autoescape=True))) + total = query.count() + rows = query.order_by(CrmContract.create_time.desc(), CrmContract.id).offset((page-1)*page_size).limit(page_size).all() + counts = dict(db.query(Link.contract_id, func.count(Link.id)).join(certs, + and_(Link.cert_type == certs.c.cert_type, Link.cert_id == certs.c.cert_id)).filter( + Link.tenant_no == user.tenant_no, Link.del_flag == 0, Link.contract_id.in_([r.id for r in rows]) + ).group_by(Link.contract_id).all()) if rows else {} + return {'list': [{**_contract_summary(r), 'certCount': counts.get(r.id, 0)} for r in rows], + 'total': total, 'page': page, 'pageSize': page_size} + + +def list_certificates(db, user, keyword=None, cert_type=None, page=1, page_size=20): + certs = _certificates(db, user, keyword, cert_type) + query = db.query(certs) + total = query.count() + rows = query.order_by(certs.c.create_time.desc(), certs.c.cert_type, certs.c.cert_id).offset( + (page-1)*page_size).limit(page_size).all() + return {'list': [_cert_summary(r) for r in rows], 'total': total, 'page': page, 'pageSize': page_size} + + +def contract_statistics(db, user, contract_id): + contract = _contracts(db, user).filter(CrmContract.id == contract_id).first() + if contract is None: + raise HTTPException(404, '合同不存在或无权访问') + certs = _certificates(db, user) + linked = _active(db, user, Link).filter(Link.contract_id == contract_id) + # 历史使用即使已停止绑定也可追溯，但证件本身仍须有效且授权。 + used = _active(db, user, Usage).join(Item, Item.usage_id == Usage.id).filter( + Item.tenant_no == user.tenant_no, Item.del_flag == 0, Item.contract_id == contract_id) + rows = db.query(certs).filter(or_(linked.filter(Link.cert_type == certs.c.cert_type, + Link.cert_id == certs.c.cert_id).exists(), used.filter(Usage.cert_type == certs.c.cert_type, + Usage.cert_id == certs.c.cert_id).exists())).order_by(certs.c.create_time.desc()).all() + keys = {(r.cert_type, r.cert_id): r.quantity_unit for r in rows} + groups, _ = _details(db, user, keys, [contract_id]) + result = [] + for row in rows: + group = groups.get((row.cert_type, row.cert_id, contract_id)) + if group is not None: + result.append({**_cert_summary(row), **group, + 'productNames': list(dict.fromkeys(p['productName'] for p in group['products']))}) + return {**_contract_summary(contract), 'certificates': result} + + +def certificate_statistics(db, user, cert_type, cert_id): + _get_row(db, user, cert_type, cert_id) + certs = _certificates(db, user, cert_type=cert_type) + row = db.query(certs).filter(certs.c.cert_id == cert_id).one() + groups, items = _details(db, user, {(cert_type, cert_id): row.quantity_unit}) + ids = [key[2] for key in groups] + contracts = _contracts(db, user).filter(CrmContract.id.in_(ids)).order_by(CrmContract.create_time.desc()).all() if ids else [] + logs = _active(db, user, Log).filter(Log.cert_type == cert_type, Log.cert_id == cert_id, + Log.plan_id.is_(None), Log.usage_id.is_(None), Log.quantity_unit == row.quantity_unit) + opening = unattributed = ZERO + for event, delta in logs.with_entities(Log.event_type, func.sum(Log.used_delta)).group_by(Log.event_type): + if event == 'opening': + opening += delta + else: + unattributed += delta + return {**_cert_summary(row), 'contracts': [{**_contract_summary(c), + **groups[(cert_type, cert_id, c.id)]} for c in contracts], 'usageItems': items, + 'openingUsedQty': opening, 'unattributedUsedQty': unattributed} diff --git a/backend/tests/test_pl_quota_api.py b/backend/tests/test_pl_quota_api.py index 42c0246e9..6b88770d2 100644 --- a/backend/tests/test_pl_quota_api.py +++ b/backend/tests/test_pl_quota_api.py @@ -148,3 +148,26 @@ def test_cert_bindings_api_accepts_collection_and_roundtrips(api): assert client.put(detail, json={'contractBindings': [], 'remark': 'edit'}).status_code == 200 assert client.put(detail, json={'contractBindings': [], 'unknownField': 'bad'}).status_code == 422 assert client.get(detail).json()['remark'] == 'edit' + + +def test_statistics_routes_are_static_paginated_and_strict(api, monkeypatch): + from app.api.pl.quota_cert import router + paths = [route.path for route in router.routes] + prefix = '/transport/quota-certs' + assert prefix + '/statistics/contracts' in paths + assert paths.index(prefix + '/statistics/contracts') < paths.index(prefix + '/{cert_type}/{cert_id}') + from app.services import pl_quota_statistics as stats + client, app, auth, _ = api + app.dependency_overrides[auth] = lambda: {'role': 'super_admin', 'user_id': 'u', 'tenant_no': 't'} + received = [] + def capture(*args, **kwargs): + received.append(kwargs) + return {'list': [], 'total': 0, 'page': 2, 'pageSize': 3} + monkeypatch.setattr(stats, 'list_certificates', capture) + result = client.get('/api' + prefix + '/statistics/certificates', params={ + 'keyword': '硫酸', 'certType': 'purchase', 'page': 2, 'pageSize': 3}) + assert result.status_code == 200 and result.json()['pageSize'] == 3 + assert received == [{'keyword': '硫酸', 'cert_type': 'purchase', 'page': 2, 'page_size': 3}] + for params in ({'page': 0}, {'pageSize': 101}, {'certType': 'bad'}, {'keyword': 'x' * 201}): + assert client.get('/api' + prefix + '/statistics/certificates', params=params).status_code == 422 + assert client.get('/api' + prefix + '/contracts/missing/statistics').status_code == 404 diff --git a/backend/tests/test_pl_quota_statistics.py b/backend/tests/test_pl_quota_statistics.py new file mode 100644 index 000000000..ef0edf626 --- /dev/null +++ b/backend/tests/test_pl_quota_statistics.py @@ -0,0 +1,218 @@ +"""统计只读、主对象分页与授权明细聚合回归。""" +from datetime import date +from types import SimpleNamespace +import importlib +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import Session +from app.models.crm_models import CrmContract, CrmContractProduct +from app.models.pl_models import PlPlan +from app.models import pl_quota_models as m + + +@pytest.fixture +def case(): + engine = create_engine('sqlite://') + models = (CrmContract, CrmContractProduct, PlPlan, m.PlPurchaseCert, m.PlTransportCert, + m.PlContractCertLink, m.PlContractCertProduct, m.PlPlanCertUsage, + m.PlPlanCertUsageItem, m.PlCertQtyLog) + for model in models: + model.__table__.create(engine) + db = Session(engine) + user = SimpleNamespace(tenant_no='t', to_user_info=lambda: {'role': 'super_admin', 'user_id': 'u'}) + for cid in ('a', 'b'): + db.add(CrmContract(id=cid, contract_no=cid, contract_name=cid, customer_id='customer', + contract_type='sale', status='approved', tenant_no='t')) + db.add(CrmContractProduct(id=cid, contract_id=cid, product_id='chemical', product_name=cid, + unit='吨', quantity=100, tenant_no='t')) + db.add(m.PlContractCertLink(id=cid, contract_id=cid, contract_no_snapshot=cid, + cert_type='purchase', cert_id='same', tenant_no='t')) + db.add(m.PlContractCertProduct(id=cid, contract_cert_link_id=cid, + contract_product_id=cid, tenant_no='t')) + db.add(m.PlPurchaseCert(id='same', cert_no='P', product_name_on_cert='硫酸', total_qty=300, + quantity_unit='吨', reserved_qty=180, used_qty=20, valid_use_text='多次有效', + buyer_name='买方', seller_name='卖方', tenant_no='t')) + db.commit() + yield SimpleNamespace(db=db, user=user) + db.close() + engine.dispose() + + +def service(): + assert importlib.util.find_spec('app.services.pl_quota_statistics'), '缺少统计服务' + return importlib.import_module('app.services.pl_quota_statistics') + + +def add_usage(c, cid, ident, reserved, used=0): + c.db.add(PlPlan(id=ident, plan_no=ident, create_date=date.today(), belongs_company='shanxia', plan_type='direct', tenant_no='t')) + c.db.add(m.PlPlanCertUsage(id=ident, plan_id=ident, plan_no_snapshot=ident, cert_type='purchase', + cert_id='same', cert_no_snapshot='P', cert_revision=1, quantity_unit='吨', tenant_no='t')) + c.db.add(m.PlPlanCertUsageItem(id=ident, usage_id=ident, contract_cert_link_id=cid, + contract_cert_product_id=cid, contract_scope_revision=1, contract_id=cid, + contract_product_id=cid, plan_product_id=ident, source_stop_type='delivery', + source_stop_id=ident, source_stop_product_id=ident, allocation_seq=1, source_unit='吨', + planned_source_qty=999, reserved_qty=reserved, used_qty=used, tenant_no='t')) + c.db.commit() + + +def test_shared_balance_and_contract_allocation_are_distinct_and_read_only(case): + s = service() + add_usage(case, 'a', 'one', 60, 20) + add_usage(case, 'a', 'two', 40) + add_usage(case, 'b', 'three', 80) + row = s.contract_statistics(case.db, case.user, 'a')['certificates'][0] + assert (row['totalQty'], row['reservedQty'], row['usedQty'], row['availableQty']) == (300, 180, 20, 100) + assert row['contractReservedQty'] == 100 and row['contractUsedQty'] == 20 + assert row['products'][0]['reservedQty'] == 100 + detail = s.certificate_statistics(case.db, case.user, 'purchase', 'same') + assert len(detail['usageItems']) == 3 + assert detail['openingUsedQty'] == 0 + assert case.db.query(m.PlContractCertLink).count() == 2 + assert not case.db.new and not case.db.dirty + + +def test_lists_paginate_main_objects_and_filter_keywords(case): + s = service() + result = s.list_contracts(case.db, case.user, page=1, page_size=1) + assert result['total'] == 2 and len(result['list']) == 1 and result['pageSize'] == 1 + assert s.list_contracts(case.db, case.user, keyword='a')['total'] == 1 + assert s.list_certificates(case.db, case.user, keyword='硫酸')['total'] == 1 + other = SimpleNamespace(**{**case.user.__dict__, 'tenant_no': 'other'}) + assert s.list_contracts(case.db, other)['total'] == 0 + assert s.list_certificates(case.db, other)['total'] == 0 + + +def test_hidden_contracts_plans_and_deleted_products_never_enter_details(case, monkeypatch): + s = service() + add_usage(case, 'a', 'visible', 30) + add_usage(case, 'a', 'hidden-plan', 70) + add_usage(case, 'b', 'hidden-contract', 80) + monkeypatch.setattr(s, 'apply_crm_owner_scope', lambda q, model, *a: q.filter(model.id == 'a')) + monkeypatch.setattr(s, 'apply_pl_scope', lambda q, model, *a: q.filter(model.id != 'hidden-plan')) + detail = s.certificate_statistics(case.db, case.user, 'purchase', 'same') + assert [x['id'] for x in detail['contracts']] == ['a'] + assert detail['contracts'][0]['contractReservedQty'] == 30 + assert [x['planId'] for x in detail['usageItems']] == ['visible'] + assert detail['openingUsedQty'] == 0 and detail['unattributedUsedQty'] == 0 + case.db.query(CrmContractProduct).filter_by(id='a').one().del_flag = 1 + case.db.commit() + assert s.certificate_statistics(case.db, case.user, 'purchase', 'same')['usageItems'] == [] + + +def test_same_id_certificate_types_units_and_real_opening_are_separate(case): + s = service() + # 按当前模型构造运输证，验证同 ID 两个额度池不会在分页或汇总中合并。 + purchase = case.db.query(m.PlPurchaseCert).one() + values = {col.name: getattr(purchase, col.name) for col in m.PlTransportCert.__table__.columns + if hasattr(purchase, col.name)} + values.update(cert_no='T', quantity_unit='kg', total_qty=500, reserved_qty=0, used_qty=0) + for col in m.PlTransportCert.__table__.columns: + if not col.nullable and col.default is None and col.name not in values: + values[col.name] = '运输主体' + case.db.add(m.PlTransportCert(**values)) + for ident, event, amount, plan, tenant, deleted in ( + ('opening', 'opening', 7, None, 't', 0), ('adjust', 'adjust', 2, None, 't', 0), + ('hidden', 'settle', 99, 'hidden-plan', 't', 0), + ('other', 'opening', 99, None, 'other', 0), ('deleted', 'opening', 99, None, 't', 1)): + case.db.add(m.PlCertQtyLog(id=ident, operation_id=ident, request_hash='hash', cert_type='purchase', + cert_id='same', plan_id=plan, event_type=event, quantity_unit='吨', used_delta=amount, + before_total_qty=300, after_total_qty=300, before_reserved_qty=0, after_reserved_qty=0, + before_used_qty=0, after_used_qty=amount, reason='凭据', tenant_no=tenant, del_flag=deleted)) + case.db.commit() + result = s.list_certificates(case.db, case.user, page_size=1) + assert result['total'] == 2 and len(result['list']) == 1 + assert s.list_certificates(case.db, case.user, cert_type='transport')['list'][0]['quantityUnit'] == 'kg' + purchase = s.certificate_statistics(case.db, case.user, 'purchase', 'same') + transport = s.certificate_statistics(case.db, case.user, 'transport', 'same') + assert purchase['openingUsedQty'] == 7 and purchase['unattributedUsedQty'] == 2 + assert transport['openingUsedQty'] == 0 and transport['contracts'] == [] + + +@pytest.mark.parametrize('model,field,value', [ + (m.PlPlanCertUsage, 'tenant_no', 'other'), (m.PlPlanCertUsage, 'del_flag', 1), + (m.PlPlanCertUsageItem, 'tenant_no', 'other'), (m.PlPlanCertUsageItem, 'del_flag', 1), + (PlPlan, 'tenant_no', 'other'), (PlPlan, 'del_flag', 1), + (CrmContract, 'del_flag', 1), (CrmContractProduct, 'tenant_no', 'other')]) +def test_invalid_ancestors_exclude_usage(case, model, field, value): + s = service() + add_usage(case, 'a', 'one', 100) + row = case.db.query(model).filter_by(id='a' if model in (CrmContract, CrmContractProduct) else 'one').one() + setattr(row, field, value) + case.db.commit() + detail = s.certificate_statistics(case.db, case.user, 'purchase', 'same') + assert detail['usageItems'] == [] + assert all(c['contractReservedQty'] == 0 for c in detail['contracts']) + + +def test_hidden_certificates_excluded_and_page_queries_do_not_grow_with_rows(case, monkeypatch): + from sqlalchemy import event + s = service() + counts = [] + for size in (1, 20): + statements = [] + def capture(*args): + statements.append(args[2]) + event.listen(case.db.bind, 'before_cursor_execute', capture) + try: + s.list_contracts(case.db, case.user, page_size=size) + finally: + event.remove(case.db.bind, 'before_cursor_execute', capture) + counts.append(len(statements)) + assert all(sql.lstrip().upper().startswith('SELECT') for sql in statements) + assert counts[0] == counts[1] + monkeypatch.setattr(s, 'apply_pl_scope', lambda q, model, *a: q.filter(model.id != 'same')) + assert s.list_contracts(case.db, case.user)['total'] == 0 + assert s.contract_statistics(case.db, case.user, 'a')['certificates'] == [] + + +def test_historical_usage_survives_removed_binding_without_recreation(case): + s = service() + add_usage(case, 'a', 'one', 0, 20) + case.db.query(m.PlContractCertLink).filter_by(id='a').one().del_flag = 1 + case.db.query(m.PlContractCertProduct).filter_by(id='a').one().del_flag = 1 + case.db.commit() + row = s.contract_statistics(case.db, case.user, 'a')['certificates'][0] + assert row['contractUsedQty'] == 20 and row['products'][0]['bindingId'] is None + assert case.db.query(m.PlContractCertLink).filter_by(id='a').one().del_flag == 1 + + +def test_role_union_uses_existing_crm_owner_and_pl_creator_helpers(case, monkeypatch): + from app.core import data_scope + s = service() + add_usage(case, 'a', 'self-plan', 30) + add_usage(case, 'a', 'custom-plan', 40) + add_usage(case, 'a', 'hidden-plan', 50) + case.db.query(m.PlPurchaseCert).one().create_by = 'u' + case.db.query(CrmContract).filter_by(id='a').one().owner_user_id = 'u' + case.db.query(CrmContract).filter_by(id='b').one().owner_user_id = 'hidden' + for ident, creator in [('self-plan', 'u'), ('custom-plan', 'colleague'), ('hidden-plan', 'hidden')]: + case.db.query(PlPlan).filter_by(id=ident).one().create_by = creator + case.db.commit() + monkeypatch.setattr(data_scope, '_resolve_user_role_scopes', lambda *a: [ + {'data_scope': '4', 'details': []}, + {'data_scope': '5', 'details': [SimpleNamespace(scope_type='user', scope_value='colleague')]}]) + user = SimpleNamespace(tenant_no='t', to_user_info=lambda: {'role': 'staff', 'user_id': 'u', 'dept_id': 'd'}) + result = s.certificate_statistics(case.db, user, 'purchase', 'same') + assert [row['id'] for row in result['contracts']] == ['a'] + assert result['contracts'][0]['contractReservedQty'] == 70 + assert {row['planId'] for row in result['usageItems']} == {'self-plan', 'custom-plan'} + + +def test_all_statistics_reads_emit_no_mutations_or_commit(case, monkeypatch): + from sqlalchemy import event + s = service() + statements = [] + def no_commit(): + pytest.fail('统计不得提交事务') + monkeypatch.setattr(case.db, 'commit', no_commit) + def capture(*args): + statements.append(args[2]) + event.listen(case.db.bind, 'before_cursor_execute', capture) + try: + s.list_contracts(case.db, case.user) + s.list_certificates(case.db, case.user) + s.contract_statistics(case.db, case.user, 'a') + s.certificate_statistics(case.db, case.user, 'purchase', 'same') + finally: + event.remove(case.db.bind, 'before_cursor_execute', capture) + assert all(sql.lstrip().upper().startswith('SELECT') for sql in statements)