shanxia2.0/backend/tests/test_pl_cert_differential_save.py

"""Certificate reconciliation regressions, using isolated SQLite only."""
from datetime import date, datetime

import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

from app.models.pl_models import PlCert, PlDocumentReminder
from app.schemas.pl.base_data import CertRow, PersonnelUpdate, VehicleUpdate
from app.services.pl_cert_service import save_owner_certs


@pytest.fixture
def db():
    engine = create_engine('sqlite://')
    PlCert.__table__.create(engine)
    PlDocumentReminder.__table__.create(engine)
    with Session(engine) as session:
        yield session
    engine.dispose()


def seed(db, id='old', **overrides):
    values = dict(id=id, tenant_no='t', owner_type='vehicle', owner_id='v',
                  vehicle_part='head', cert_name='行驶证', cert_no='N1', status='revoked',
                  create_by='creator', create_dept_id='dept', create_time=datetime(2020, 1, 1))
    values.update(overrides)
    cert = PlCert(**values)
    db.add(cert)
    db.flush()
    return cert


def row(**overrides):
    values = dict(vehiclePart='head', certName='行驶证', certNo='N1')
    values.update(overrides)
    return values


def save(db, rows):
    save_owner_certs(db, 'vehicle', 'v', rows, 't', user_id='editor', dept_id='new-dept')
    db.flush()


def test_legacy_no_id_reuses_unique_business_key_and_preserves_status(db):
    cert = seed(db)
    save(db, [row(attachmentUrls='changed', remindDays=15)])
    assert db.query(PlCert).count() == 1
    assert cert.del_flag == 0
    assert cert.status == 'revoked'
    assert cert.create_by == 'creator'
    assert cert.create_time == datetime(2020, 1, 1)
    assert cert.attachment_urls == 'changed'
    assert cert.remind_days == 15


@pytest.mark.parametrize('foreign', [
    dict(id='missing'), dict(id='foreign', tenant_no='other'),
    dict(id='foreign', owner_id='other'), dict(id='foreign', owner_type='personnel'),
    dict(id='foreign', del_flag=1),
])
def test_invalid_id_rejected_before_any_certificate_mutation(db, foreign):
    cert = seed(db)
    if foreign['id'] != 'missing':
        seed(db, **foreign)
    before = db.query(PlCert).count()
    with pytest.raises(HTTPException) as exc:
        save(db, [row(certId='old', certName='must not change'), row(certId=foreign['id'])])
    assert exc.value.status_code == 400
    assert cert.cert_name == '行驶证'
    assert cert.del_flag == 0
    assert db.query(PlCert).count() == before


def test_duplicate_explicit_id_is_rejected_before_mutation(db):
    cert = seed(db)
    with pytest.raises(HTTPException):
        save(db, [row(certId='old', certName='first'), row(certId='old', certName='second')])
    assert cert.cert_name == '行驶证'


def test_ambiguous_legacy_key_requires_explicit_ids(db):
    seed(db)
    seed(db, 'duplicate')
    with pytest.raises(HTTPException):
        save(db, [row()])
    assert all(c.del_flag == 0 for c in db.query(PlCert).all())


def test_duplicate_legacy_rows_rejected_before_mutation(db):
    cert = seed(db)
    with pytest.raises(HTTPException):
        save(db, [row(), row()])
    assert cert.del_flag == 0
    assert db.query(PlCert).count() == 1


def test_explicit_update_add_and_remove_only_changes_requested_rows(db):
    cert = seed(db)
    removed = seed(db, 'removed', cert_no='N2')
    save(db, [row(certId='old', certNo='edited'), row(certNo='new')])
    assert cert.cert_no == 'edited'
    assert cert.status == 'revoked'
    assert cert.del_flag == 0
    assert removed.del_flag == 1
    new = db.query(PlCert).filter(PlCert.cert_no == 'new').one()
    assert new.id not in ('old', 'removed')
    assert new.status == 'valid'
    assert db.query(PlCert).count() == 3


def test_empty_collection_explicitly_clears(db):
    cert = seed(db)
    save(db, [])
    assert cert.del_flag == 1
    assert db.query(PlCert).count() == 1


def test_explicit_row_can_coexist_with_new_same_key_row(db):
    cert = seed(db)
    save(db, [row(), row(certId='old')])
    assert cert.del_flag == 0
    assert db.query(PlCert).filter_by(del_flag=0).count() == 2


@pytest.mark.parametrize('source', ['dict', 'cert_schema', 'vehicle_schema', 'personnel_schema'])
def test_submitted_certificate_row_uses_schema_defaults_for_omitted_fields(db, source):
    cert = seed(db, expire_date=date(2030, 1, 1), remind_days=7,
                attachment_urls='stored-attachment')
    submitted = {'certId': 'old', 'certName': '新名称'}
    if source == 'cert_schema':
        submitted = CertRow.model_validate(submitted)
    elif source == 'vehicle_schema':
        submitted = VehicleUpdate(certList=[submitted]).cert_list[0]
    elif source == 'personnel_schema':
        submitted = PersonnelUpdate(certList=[submitted]).cert_list[0]
    save(db, [submitted])
    assert cert.cert_name == '新名称'
    assert cert.id == 'old' and db.query(PlCert).count() == 1
    assert cert.cert_no is None
    assert cert.vehicle_part is None
    assert cert.expire_date is None
    assert cert.remind_days == 30
    assert cert.attachment_urls is None
    assert cert.status == 'revoked'


def test_explicit_null_clears_nullable_certificate_fields(db):
    cert = seed(db, expire_date=date(2030, 1, 1), attachment_urls='stored-attachment')
    save(db, [dict(certId='old', certName='行驶证', certNo=None, vehiclePart=None,
                   expireDate=None, attachmentUrls=None, remindDays=0)])
    assert cert.cert_no is None
    assert cert.vehicle_part is None
    assert cert.expire_date is None
    assert cert.attachment_urls is None
    assert cert.remind_days == 0


def test_new_certificate_uses_schema_defaults(db):
    save(db, [{'certName': '新证书'}])
    cert = db.query(PlCert).one()
    assert cert.remind_days == 30
    assert cert.cert_no is None
    assert cert.expire_date is None
    assert cert.attachment_urls is None
    assert cert.status == 'valid'