107 lines
5.5 KiB
Python
107 lines
5.5 KiB
Python
"""Existing UI serialization must still clear full-row fields without replacing IDs."""
|
|
import json
|
|
import subprocess
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
|
|
import pymysql
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import pl_models as m
|
|
from app.schemas.pl.base_data import VehicleUpdate
|
|
from app.schemas.pl.plan import PlanLoadBody, PlanDeliveryBody
|
|
from app.services.pl_cert_service import save_owner_certs
|
|
from app.services.pl_plan_stop_service import replace_plan_stops
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
@pytest.fixture
|
|
def db(monkeypatch):
|
|
def forbidden(*args, **kwargs):
|
|
raise AssertionError('Business database forbidden')
|
|
monkeypatch.setattr(pymysql, 'connect', forbidden)
|
|
monkeypatch.setattr(pymysql.connections.Connection, 'connect', forbidden)
|
|
engine = create_engine('sqlite://')
|
|
for table in m.PlPlan.metadata.sorted_tables:
|
|
if table.name.startswith('pl_'):
|
|
table.create(engine, checkfirst=True)
|
|
with Session(engine, autoflush=False) as session:
|
|
yield session
|
|
engine.dispose()
|
|
|
|
|
|
def ui_payload(script, data):
|
|
"""Execute the current JS mapper, including JSON's omission of undefined."""
|
|
result = subprocess.run(['node', '-e', script], input=json.dumps(data), text=True,
|
|
cwd=ROOT, capture_output=True, check=True)
|
|
return json.loads(result.stdout)
|
|
|
|
|
|
def test_vehicle_ui_cleared_certificate_fields_update_original_row(db):
|
|
cert = m.PlCert(id='cert', tenant_no='t', owner_type='vehicle', owner_id='v',
|
|
cert_name='行驶证', cert_no='OLD', vehicle_part='head',
|
|
expire_date=date(2030, 1, 1), attachment_urls='old-proof',
|
|
status='revoked', create_by='creator', create_time=datetime(2020, 1, 1))
|
|
db.add(cert); db.flush()
|
|
payload = ui_payload(r"""
|
|
const fs = require('fs')
|
|
const src = fs.readFileSync('web-admin/src/components/business/pl-vehicle-form-dialog/index.vue', 'utf8')
|
|
const expression = src.match(/certList: (certList\.value\.map\(\(c\) => \(\{[\s\S]*?\}\)\))/)[1]
|
|
const rows = JSON.parse(fs.readFileSync(0, 'utf8'))
|
|
process.stdout.write(JSON.stringify({certList: new Function('certList', 'return ' + expression)({value: rows})}))
|
|
""", [dict(certId='cert', certName='行驶证', vehiclePart='head', certNo='',
|
|
expireDate='', attachmentUrls='', remindDays=30)])
|
|
assert 'certNo' not in payload['certList'][0]
|
|
body = VehicleUpdate.model_validate(payload)
|
|
save_owner_certs(db, 'vehicle', 'v', body.cert_list, 't', user_id='u', dept_id='d')
|
|
db.flush(); db.expire_all()
|
|
current = db.get(m.PlCert, 'cert')
|
|
assert current.cert_no is None
|
|
assert current.expire_date is None
|
|
assert current.attachment_urls is None
|
|
assert current.status == 'revoked' and current.create_by == 'creator'
|
|
assert current.create_time == datetime(2020, 1, 1)
|
|
assert db.query(m.PlCert).count() == 1 and current.del_flag == 0
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['load', 'delivery'])
|
|
def test_stop_ui_clears_parent_fields_and_preserves_unsubmitted_children(db, kind):
|
|
model = m.PlPlanLoad if kind == 'load' else m.PlPlanDelivery
|
|
detail_model = m.PlPlanLoadProduct if kind == 'load' else m.PlPlanDeliveryProduct
|
|
time_field = 'load_time' if kind == 'load' else 'delivery_time'
|
|
parent = model(id='stop', tenant_no='t', plan_id='p', customer_name='客户',
|
|
address='old-address', contact='old-contact', notes='old-notes',
|
|
create_by='creator', **{time_field: datetime(2030, 1, 1)})
|
|
detail = detail_model(id='detail', tenant_no='t', plan_id='p',
|
|
**{('load_id' if kind == 'load' else 'delivery_id'): 'stop'},
|
|
plan_product_id='pp', product_id='sku', product_no='SKU', product_name='产品',
|
|
remark='keep child', plan_qty=8)
|
|
db.add_all([parent, detail]); db.flush()
|
|
payload = ui_payload(r"""
|
|
const fs = require('fs')
|
|
const input = JSON.parse(fs.readFileSync(0, 'utf8'))
|
|
const src = fs.readFileSync('web-admin/src/utils/pl-plan-stop-form.ts', 'utf8')
|
|
const name = input.kind === 'load' ? 'buildLoadPayload' : 'buildDeliveriesPayload'
|
|
const body = src.match(new RegExp('export function ' + name + '\\([^\\n]*\\) \\{([\\s\\S]*?)\\n\\}'))[1]
|
|
const fn = new Function(input.kind === 'load' ? 'load' : 'deliveries', 'buildStopProductPayload', body)
|
|
process.stdout.write(JSON.stringify(fn(input.kind === 'load' ? input.row : [input.row], x => x)))
|
|
""", dict(kind=kind, row=dict(id='stop', customerName='客户', address='', contact='',
|
|
notes='', loadTime='', deliveryTime='', products=[])))
|
|
submitted = payload if kind == 'load' else payload[0]
|
|
assert 'contact' not in submitted and 'notes' not in submitted
|
|
# Parent submission and child collection omission are independent contracts.
|
|
submitted.pop('products')
|
|
body = (PlanLoadBody if kind == 'load' else PlanDeliveryBody).model_validate(submitted)
|
|
replace_plan_stops(db, 't', 'p', body if kind == 'load' else None,
|
|
[body] if kind == 'delivery' else None, 'u', 'd', [])
|
|
db.flush(); db.expire_all()
|
|
current = db.get(model, 'stop')
|
|
assert all(getattr(current, field) is None for field in ('contact', 'notes', 'address', time_field))
|
|
assert current.create_by == 'creator' and current.del_flag == 0
|
|
assert db.query(model).count() == 1
|
|
assert detail.remark == 'keep child' and detail.plan_qty == 8 and detail.del_flag == 0
|
|
assert db.query(detail_model).count() == 1
|