110 lines
6.0 KiB
Python
110 lines
6.0 KiB
Python
from datetime import date
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
@pytest.fixture
|
|
def case(monkeypatch):
|
|
from app.models.pl_models import PlPlan, PlPlanProduct, PlPlanChange, PlPlanLoad, PlPlanDelivery, PlPlanLoadProduct, PlPlanDeliveryProduct
|
|
from app.models.crm_models import SysOss, CrmContract
|
|
from app.services import pl_legal_use_document as svc
|
|
engine = create_engine('sqlite://')
|
|
for model in (PlPlan, PlPlanProduct, PlPlanChange, SysOss, PlPlanLoad, PlPlanDelivery, PlPlanLoadProduct, PlPlanDeliveryProduct, CrmContract):
|
|
model.__table__.create(engine)
|
|
db = Session(engine)
|
|
user = SimpleNamespace(id='u', name='测试', dept_id='d', tenant_no='t', role='super_admin',
|
|
to_user_info=lambda: {'role': 'super_admin', 'user_id': 'u'})
|
|
plan = PlPlan(id='p', plan_no='P', create_date=date.today(), belongs_company='shanxia',
|
|
plan_type='direct', tenant_no='t', transport_status='arranged', archive_status='archived')
|
|
db.add(plan)
|
|
for rid in ('r1', 'r2'):
|
|
db.add(PlPlanProduct(id=rid, plan_id='p', product_id='sku', product_no='SKU', product_name='产品',
|
|
is_explosive=1, tenant_no='t', actual_qty=18, legal_use_cert_url='legacy.jpg'))
|
|
db.add(SysOss(id='oss', url='legal/new.jpg', tenant_no='t', create_by='u'))
|
|
db.commit()
|
|
monkeypatch.setattr(svc, 'storage_base_url', lambda: 'https://files.example/bucket')
|
|
yield SimpleNamespace(db=db, user=user, svc=svc, plan=plan, product=PlPlanProduct, change=PlPlanChange)
|
|
db.close()
|
|
engine.dispose()
|
|
|
|
|
|
def test_archived_patch_preserves_identity_and_business_facts(case):
|
|
c = case
|
|
before = c.db.get(c.product, 'r1').create_time
|
|
result = c.svc.save_documents(c.db, c.user, 'p', {'products': [{'id': 'r1', 'legalUseCertUrl': 'legal/new.jpg'}]})
|
|
assert next(r for r in result['products'] if r['id'] == 'r1')['legalUseCertUrl'] == 'legal/new.jpg'
|
|
row = c.db.get(c.product, 'r1')
|
|
assert row.create_time == before and row.actual_qty == 18
|
|
assert c.plan.archive_status == 'archived' and c.plan.transport_status == 'arranged'
|
|
assert c.db.get(c.product, 'r2').legal_use_cert_url == 'legacy.jpg'
|
|
assert c.db.query(c.change).count() == 1
|
|
|
|
|
|
def test_omissions_and_empty_collection_do_not_clear_attachments(case):
|
|
c = case
|
|
original = c.svc.get_documents(c.db, c.user, 'p')
|
|
for body in ({}, {'products': []}, {'products': [{'id': 'r1'}]}):
|
|
assert c.svc.save_documents(c.db, c.user, 'p', body) == original
|
|
c.svc.save_documents(c.db, c.user, 'p', {'products': [{'id': 'r1', 'legalUseCertUrl': ''}]})
|
|
assert c.db.get(c.product, 'r1').legal_use_cert_url == ''
|
|
assert c.db.get(c.product, 'r2').legal_use_cert_url == 'legacy.jpg'
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['missing', 'r1'])
|
|
def test_batch_invalid_or_duplicate_id_rejected_before_write(case, bad):
|
|
c = case
|
|
with pytest.raises(HTTPException):
|
|
c.svc.save_documents(c.db, c.user, 'p', {'products': [
|
|
{'id': 'r1', 'legalUseCertUrl': 'legal/new.jpg'}, {'id': bad, 'legalUseCertUrl': ''}]})
|
|
assert c.db.get(c.product, 'r1').legal_use_cert_url == 'legacy.jpg'
|
|
assert c.db.query(c.change).count() == 0
|
|
|
|
|
|
def test_permission_tenant_and_attachment_ownership(case, monkeypatch):
|
|
c = case
|
|
other = SimpleNamespace(**{**c.user.__dict__, 'tenant_no': 'other'})
|
|
with pytest.raises(HTTPException):
|
|
c.svc.get_documents(c.db, other, 'p')
|
|
for url in ('https://evil.example/a.jpg', '../a.jpg', 'legal/unknown.jpg'):
|
|
with pytest.raises(HTTPException):
|
|
c.svc.save_documents(c.db, c.user, 'p', {'products': [{'id': 'r1', 'legalUseCertUrl': url}]})
|
|
monkeypatch.setattr(c.svc, 'user_can_upload_plan_cert', lambda *a: False)
|
|
with pytest.raises(HTTPException) as error:
|
|
c.svc.save_documents(c.db, c.user, 'p', {'products': [{'id': 'r1', 'legalUseCertUrl': ''}]})
|
|
assert error.value.status_code == 403
|
|
assert c.db.get(c.product, 'r1').legal_use_cert_url == 'legacy.jpg'
|
|
|
|
|
|
def test_existing_legacy_url_preserved_without_reupload(case):
|
|
c = case
|
|
c.svc.save_documents(c.db, c.user, 'p', {'products': [
|
|
{'id': 'r1', 'legalUseCertUrl': 'legacy.jpg,legal/new.jpg'}]})
|
|
assert c.db.get(c.product, 'r1').legal_use_cert_url == 'legacy.jpg,legal/new.jpg'
|
|
|
|
|
|
def test_own_upload_with_legacy_empty_department_is_authorized(case, monkeypatch):
|
|
from app.services import pl_quota_attachments as attachments
|
|
c = case
|
|
def deny_scope(*args):
|
|
raise HTTPException(403, '部门范围不含空部门')
|
|
monkeypatch.setattr(attachments, 'assert_pl_row_in_scope', deny_scope)
|
|
assert attachments.authorize_attachment(c.db, c.user, 'legal/new.jpg', 'https://files.example/bucket') == 'legal/new.jpg'
|
|
foreign = SimpleNamespace(**{**c.user.__dict__, 'id': 'other-u'})
|
|
with pytest.raises(HTTPException):
|
|
attachments.authorize_attachment(c.db, foreign, 'legal/new.jpg', 'https://files.example/bucket')
|
|
|
|
|
|
def test_context_uses_original_stop_product_identity_and_filters_tenant(case):
|
|
from app.models.pl_models import PlPlanLoad, PlPlanLoadProduct
|
|
c=case
|
|
c.db.add(PlPlanLoad(id='load',plan_id='p',customer_name='实际供方',address='实际装货地',tenant_no='t'))
|
|
c.db.add(PlPlanLoadProduct(id='source',load_id='load',plan_id='p',plan_product_id='r1',product_id='sku',product_no='SKU',product_name='产品',contract_id='contract',contract_product_id='cp',tenant_no='t'))
|
|
c.db.add(PlPlanLoadProduct(id='foreign',load_id='load',plan_id='p',plan_product_id='r1',product_id='sku',product_no='SKU',product_name='其他租户',contract_id='hidden',tenant_no='other'))
|
|
c.db.flush()
|
|
row=next(p for p in c.svc.get_documents(c.db,c.user,'p')['products'] if p['id']=='r1')
|
|
assert row['contexts']==[{'stopType':'load','stopProductId':'source','partyName':'实际供方','address':'实际装货地','contractId':'contract','contractNo':None,'contractProductId':'cp'}]
|