shanxia2.0/backend/tests/test_pl_dispatch_calendar.py

"""手工派车日历：独立记录、隔离、选择项及 HTTP 权限回归。"""
from datetime import date, datetime, timedelta
from importlib.util import find_spec

import pytest
from fastapi import HTTPException
from pydantic import ValidationError
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.core.db_auth import DbUserContext
from app.models.crm_models import CrmProduct
from app.models.pl_models import PlVehicle


def modules():
    assert find_spec('app.services.pl_dispatch_calendar_service') is not None, '手工派车日历服务尚未实现'
    from app.models.pl_dispatch_calendar import PlDispatchCalendar
    from app.schemas.pl.dispatch_calendar import CreateBody, UpdateBody
    from app.services import pl_dispatch_calendar_service as svc
    return PlDispatchCalendar, CreateBody, UpdateBody, svc


def user(uid='u1', tenant='t1', role='super_admin'):
    return DbUserContext(uid, '测试用户', 'd1', role, tenant)


def payload(**kwargs):
    return {'dispatchDate': '2026-10-26', 'vehicleId': 'v1', 'productId': 'p1',
            'origin': '宜昌', 'destination': '重庆', 'remark': None, **kwargs}


@pytest.fixture
def db():
    model, _, _, _ = modules()
    engine = create_engine('sqlite://')
    for table in (model, PlVehicle, CrmProduct):
        table.__table__.create(engine)
    with sessionmaker(bind=engine)() as session:
        session.add_all([
            PlVehicle(id='v1', license_plate='鄂E12345', vehicle_type='self', tenant_no='t1', create_by='u1'),
            PlVehicle(id='v2', license_plate='鄂E67890', vehicle_type='carrier', tenant_no='t1', create_by='u2'),
            PlVehicle(id='v3', license_plate='渝A12345', vehicle_type='self', tenant_no='t2', create_by='u1'),
            PlVehicle(id='v4', license_plate='鄂E00000', vehicle_type='self', tenant_no='t1', create_by='u1', del_flag=1),
            CrmProduct(id='p1', product_no='P001', product_name='液碱', tenant_no='t1', create_by='u1'),
            CrmProduct(id='p2', product_no='P002', product_name='盐酸', tenant_no='t1', create_by='u2'),
            CrmProduct(id='p3', product_no='P003', product_name='硫酸', tenant_no='t2', create_by='u1'),
            CrmProduct(id='p4', product_no='P004', product_name='氯酸钠', tenant_no='t1', create_by='u1', del_flag=1),
        ])
        session.commit()
        yield session
    engine.dispose()


def create(db, caller=None, **kwargs):
    _, schema, _, svc = modules()
    return svc.create_record(db, 't1', caller or user(), schema(**payload(**kwargs)))


def test_crud_duplicate_trips_snapshots_and_soft_delete(db):
    model, _, update, svc = modules()
    first = create(db)
    second = create(db)
    assert first.id != second.id
    assert first.tenant_no == 't1' and first.create_by == 'u1' and first.create_dept_id == 'd1'
    assert first.dispatch_date == date(2026, 10, 26)
    db.get(PlVehicle, 'v1').license_plate = '鄂E99999'
    db.get(CrmProduct, 'p1').product_name = '烧碱'
    db.commit()
    assert svc.get_record(db, 't1', user(), first.id).license_plate == '鄂E12345'
    updated = svc.update_record(db, 't1', user(), first.id, update(**payload(vehicleId='v2', productId='p2', origin='武汉')))
    result = svc._row_to_dict(updated)
    assert result['licensePlate'] == '鄂E67890' and result['productName'] == '盐酸'
    assert result['origin'] == '武汉'
    assert set(result) == {'id', 'dispatchDate', 'vehicleId', 'licensePlate', 'productId', 'productName', 'origin', 'destination', 'remark', 'createTime', 'updateTime'}
    assert svc.delete_record(db, 't1', user(), first.id) == 1
    assert db.get(model, first.id).del_flag == 1
    assert svc.list_records(db, 't1', user(), month='2026-10')['total'] == 1
    with pytest.raises(HTTPException):
        svc.get_record(db, 't1', user(), first.id)


def test_month_pagination_sort_and_vehicle_filter(db):
    _, _, _, svc = modules()
    first = create(db, dispatchDate='2026-10-01')
    first.create_time = datetime.now() - timedelta(days=2)
    second = create(db, dispatchDate='2026-10-31', vehicleId='v2')
    create(db, dispatchDate='2026-09-30')
    create(db, dispatchDate='2026-11-01')
    page1 = svc.list_records(db, 't1', user(), month='2026-10', page=1, page_size=1)
    page2 = svc.list_records(db, 't1', user(), month='2026-10', page=2, page_size=1)
    assert page1['total'] == 2 and page1['list'][0]['id'] == second.id
    assert page2['list'][0]['id'] == first.id
    assert svc.list_records(db, 't1', user(), month='2026-10', vehicle_id='v1')['total'] == 1


@pytest.mark.parametrize('month', ['2026-13', '2026-1', 'abc', '1899-12', '2101-01'])
def test_invalid_month_rejected(db, month):
    _, _, _, svc = modules()
    with pytest.raises(HTTPException) as exc:
        svc.list_records(db, 't1', user(), month=month)
    assert exc.value.status_code == 422


def test_tenant_and_data_scope_cover_records_and_options(db, monkeypatch):
    _, _, update, svc = modules()
    first = create(db)
    other = create(db, caller=user('u2'))
    import app.core.data_scope as scope
    monkeypatch.setattr(scope, '_resolve_user_role_scopes', lambda *a: [{'data_scope': '4', 'details': []}])
    limited = user(role='admin')
    assert [r['id'] for r in svc.list_records(db, 't1', limited, month='2026-10')['list']] == [first.id]
    assert svc.list_records(db, 't2', user(tenant='t2'), month='2026-10')['total'] == 0
    for tenant, caller, record in [('t1', limited, other), ('t2', user(tenant='t2'), first)]:
        for action in [lambda: svc.get_record(db, tenant, caller, record.id),
                       lambda: svc.update_record(db, tenant, caller, record.id, update(**payload())),
                       lambda: svc.delete_record(db, tenant, caller, record.id)]:
            with pytest.raises(HTTPException) as exc:
                action()
            assert exc.value.status_code == 404
    assert svc.list_vehicles(db, 't1', limited)['list'] == [{'id': 'v1', 'licensePlate': '鄂E12345'}]
    assert svc.list_products(db, 't1', limited)['list'] == [{'id': 'p1', 'productName': '液碱'}]
    with pytest.raises(HTTPException):
        create(db, caller=limited, vehicleId='v2')
    with pytest.raises(HTTPException):
        create(db, caller=limited, productId='p2')
    monkeypatch.setattr(scope, '_resolve_user_role_scopes', lambda *a: [{'data_scope': '4', 'details': []}, {'data_scope': '1', 'details': []}])
    assert svc.list_records(db, 't1', limited, month='2026-10')['total'] == 2
    assert svc.list_vehicles(db, 't1', limited)['total'] == 2


def test_options_full_library_types_search_pagination(db):
    _, _, _, svc = modules()
    assert svc.list_vehicles(db, 't1', user())['total'] == 2
    assert svc.list_vehicles(db, 't1', user(), q='67890')['list'] == [{'id': 'v2', 'licensePlate': '鄂E67890'}]
    assert len(svc.list_vehicles(db, 't1', user(), page_size=1)['list']) == 1
    assert svc.list_products(db, 't1', user(), q='盐酸')['list'] == [{'id': 'p2', 'productName': '盐酸'}]
    assert svc.list_products(db, 't1', user(), q='%')['total'] == 0


@pytest.mark.parametrize('field,value', [('vehicleId', 'v3'), ('vehicleId', 'v4'), ('vehicleId', 'missing'), ('productId', 'p3'), ('productId', 'p4'), ('productId', 'missing')])
def test_invalid_cross_tenant_and_deleted_relations_rejected(db, field, value):
    _, _, _, svc = modules()
    with pytest.raises(HTTPException) as exc:
        create(db, **{field: value})
    assert exc.value.status_code == 422
    assert svc.list_records(db, 't1', user(), month='2026-10')['total'] == 0


@pytest.mark.parametrize('bad', [
    {'dispatchDate': '2026-02-30'}, {'dispatchDate': '1899-12-31'}, {'dispatchDate': '2101-01-01'},
    {'dispatchDate': 1792972800}, {'dispatchDate': '2026-10-26T00:00:00'}, {'origin': ' '}, {'destination': 'x' * 256},
    {'vehicleId': ''}, {'productId': ''}, {'remark': 'x' * 501}, {'tenantNo': 't2'},
    {'licensePlate': '伪造'}, {'productName': '伪造'}, {'createBy': 'other'},
])
def test_schema_rejects_bad_fields_and_client_snapshots(bad):
    _, schema, _, _ = modules()
    with pytest.raises(ValidationError):
        schema(**payload(**bad))


def test_http_read_only_manage_permissions_both_clients_and_routes(db, monkeypatch):
    from fastapi import FastAPI
    from fastapi.testclient import TestClient
    from app.core.auth import get_current_user
    from app.core.database import get_db
    from app.api.pl import dispatch_calendar as calendar_api
    from app.api.pl.dispatch_calendar import router
    from sqlalchemy.pool import StaticPool
    model, _, _, _ = modules()
    engine = create_engine('sqlite://', connect_args={'check_same_thread': False}, poolclass=StaticPool)
    for table in (model, PlVehicle, CrmProduct):
        table.__table__.create(engine)
    factory = sessionmaker(bind=engine)
    with factory() as session:
        session.add(PlVehicle(id='v1', license_plate='鄂E12345', vehicle_type='self', tenant_no='t1'))
        session.add(CrmProduct(id='p1', product_no='P1', product_name='液碱', tenant_no='t1'))
        session.commit()
    def database():
        with factory() as session:
            yield session
    import app.core.data_scope as scope
    monkeypatch.setattr(scope, '_resolve_user_role_scopes', lambda *a: [{'data_scope': '1', 'details': []}])
    app = FastAPI()
    app.include_router(router, prefix='/api')
    app.dependency_overrides[get_db] = database
    path = '/api/transport/dispatch-calendar'
    codes = set()
    monkeypatch.setattr(calendar_api, 'iter_user_menu_perm_codes', lambda *a: codes)
    with TestClient(app) as client:
        assert client.get(path, params={'month': '2026-10'}).status_code == 401
        app.dependency_overrides[get_current_user] = lambda: {'role': 'admin', 'user_id': 'u1', 'tenant_no': 't1'}
        assert client.get(path, params={'month': '2026-10'}).status_code == 403
        for prefix in ('pl', 'mini:pl'):
            codes.clear()
            codes.add(prefix + ':dispatch-calendar:list')
            assert client.get(path, params={'month': '2026-10'}).status_code == 200
            assert client.get(path + '/vehicles').json()['total'] == 1
            assert client.get(path + '/products').json()['total'] == 1
            assert client.post(path, json=payload()).status_code == 403
            codes.add(prefix + ':dispatch-calendar:manage')
            response = client.post(path, json=payload())
            assert response.status_code == 200, response.text
            record_id = response.json()['id']
            assert client.get(f'{path}/{record_id}').json()['dispatchDate'] == '2026-10-26'
            assert client.put(f'{path}/{record_id}', json=payload(origin='武汉')).json()['origin'] == '武汉'
            codes.remove(prefix + ':dispatch-calendar:manage')
            assert client.put(f'{path}/{record_id}', json=payload()).status_code == 403
            assert client.delete(f'{path}/{record_id}').status_code == 403
            codes.add(prefix + ':dispatch-calendar:manage')
            assert client.delete(f'{path}/{record_id}').json() == {'deleted': 1}
            assert client.get(f'{path}/{record_id}').status_code == 404
        assert client.post(path, json=payload(tenantNo='t2')).status_code == 422
        assert client.get(path, params={'month': '2026-13'}).status_code == 422
    engine.dispose()


def test_permissions_use_current_tenant_and_button_grants(db):
    from app.api.pl.dispatch_calendar import require_manage
    from app.models.models import SysRole, SysUserRole, SysMenu, SysRoleMenu
    for model in (SysRole, SysUserRole, SysMenu, SysRoleMenu):
        model.__table__.create(db.get_bind())
    db.add_all([
        SysRole(id='role1', role_name='调度员', tenant_no='t1'),
        SysUserRole(id='ur1', user_id='u1', role_id='role1', tenant_no='t1'),
        SysMenu(id='calendar', menu_name='派车日历', menu_type='C', parent_id='0',
                perms='pl:dispatch-calendar:list', tenant_no='t1'),
        SysMenu(id='manage', menu_name='维护派车日历', menu_type='F', parent_id='calendar',
                perms='pl:dispatch-calendar:manage', visible='1', tenant_no='t1'),
        SysRoleMenu(id='rm1', role_id='role1', menu_id='manage', tenant_no='t1'),
    ])
    db.commit()
    caller = {'role': 'admin', 'user_id': 'u1', 'tenant_no': 't1'}
    assert require_manage(caller, db) == caller
    with pytest.raises(HTTPException) as exc:
        require_manage({**caller, 'tenant_no': 't2'}, db)
    assert exc.value.status_code == 403


def test_month_pagination_can_fetch_more_than_500_trips(db):
    model, _, _, svc = modules()
    db.add_all([model(id=f'bulk-{number:04}', dispatch_date=date(2026, 10, 26),
                      vehicle_id='v1', license_plate='鄂E12345', product_id='p1', product_name='液碱',
                      origin='宜昌', destination='重庆', tenant_no='t1', create_by='u1')
                for number in range(503)])
    db.commit()
    first = svc.list_records(db, 't1', user(), month='2026-10', page=1, page_size=500)
    second = svc.list_records(db, 't1', user(), month='2026-10', page=2, page_size=500)
    assert first['total'] == second['total'] == 503
    assert len(first['list']) == 500 and len(second['list']) == 3
    assert len({row['id'] for row in first['list'] + second['list']}) == 503


def test_calendar_openapi_schema_can_be_generated():
    from fastapi import FastAPI
    from app.api.pl.dispatch_calendar import router
    app = FastAPI()
    app.include_router(router, prefix='/api')
    document = app.openapi()
    for schema in ('CreateBody', 'UpdateBody'):
        field = document['components']['schemas'][schema]['properties']['dispatchDate']
        assert field['type'] == 'string' and field['format'] == 'date'