"""Run the legacy bridge against SQLite; never connect to a business database."""
from datetime import date, datetime
from decimal import Decimal

import pytest
from sqlalchemy import create_engine, update
from sqlalchemy.orm import Session

from app.models.crm_models import CrmProduct
from app.models.db_models import DbInboundOutbound, DbPurchasePlan, DbSequence, DbWeighingSlip
from app.models.pl_models import PlPlan, PlPlanProduct, PlTransport, PlTransportDbLink
from app.services import pl_dibang_bridge_service as bridge


@pytest.fixture
def db(monkeypatch):
    import pymysql
    monkeypatch.setattr(pymysql, 'connect', lambda *a, **k: pytest.fail('MySQL is forbidden'))
    engine = create_engine('sqlite://')
    for model in (CrmProduct, PlPlan, PlPlanProduct, PlTransport, PlTransportDbLink,
                  DbPurchasePlan, DbInboundOutbound, DbWeighingSlip, DbSequence):
        model.__table__.create(engine)
    with Session(engine) as session:
        session.add(DbSequence(id='seq', tenant_no='t', seq_name='inbound_outbound', next_val=1))
        session.commit()
        yield session


def seed(db):
    plan = PlPlan(id='p', tenant_no='t', plan_no='P', plan_type='outbound',
                  create_date=date(2026, 10, 1), plan_date=date(2026, 10, 1),
                  pickup_time=datetime(2026, 10, 1), belongs_company='shanxia',
                  customer_name='Customer', license_plate='Plate')
    product = PlPlanProduct(id='pp', tenant_no='t', plan_id='p', product_id='sku',
                            product_no='SKU', product_name='Product', model='Spec',
                            unit='吨', quantity=5)
    transport = PlTransport(id='tr', tenant_no='t', transport_no='TR', license_plate='Plate')
    db.add_all([plan, product, transport, CrmProduct(id='sku', tenant_no='t',
                product_no='SKU', product_name='Product', model='Spec', status='on')])
    db.commit()
    return plan, product, transport


def sync(db, rows, products=None):
    plan, product, transport = rows
    result = bridge.sync_dibang_for_transport(db, tenant_no='t', plan=plan,
                products=products if products is not None else [product], transport=transport,
                user_id='editor', dept_id='dept')
    db.flush()
    return result


def records(db):
    return (db.query(PlTransportDbLink).one(), db.query(DbInboundOutbound).one(),
            db.query(DbPurchasePlan).one())


def test_repeat_reuses_link_inventory_and_monthly_contribution(db):
    rows = seed(db)
    first = sync(db, rows)
    link, inventory, monthly = records(db)
    created = (link.id, inventory.id, link.create_time, inventory.create_time)
    inventory.site_feedback_remark = 'keep feedback'
    inventory.site_feedback_images = '["url"]'
    inventory.site_access_token = 'token'
    db.commit()
    second = sync(db, rows)
    assert db.query(PlTransportDbLink).count() == db.query(DbInboundOutbound).count() == 1
    assert second == first and monthly.planned_quantity == 5
    assert (link.id, inventory.id, link.create_time, inventory.create_time) == created
    assert (inventory.site_feedback_remark, inventory.site_feedback_images,
            inventory.site_access_token) == ('keep feedback', '["url"]', 'token')


def test_quantity_changes_only_apply_delta_and_keep_other_contributions(db):
    rows = seed(db)
    sync(db, rows)
    link, inventory, monthly = records(db)
    monthly.planned_quantity += 20
    rows[1].quantity = 8
    db.commit()
    sync(db, rows)
    assert db.query(DbInboundOutbound).count() == 1
    assert monthly.planned_quantity == 28
    assert inventory.quantity == link.contributed_qty == 8
    rows[1].quantity = 3
    db.commit()
    sync(db, rows)
    assert monthly.planned_quantity == 23 and inventory.quantity == 3


def test_month_migration_moves_only_this_contribution(db):
    rows = seed(db)
    sync(db, rows)
    link, inventory, old = records(db)
    old.planned_quantity += 20
    rows[0].pickup_time = datetime(2026, 11, 2)
    rows[1].quantity = 8
    db.commit()
    sync(db, rows)
    assert db.query(DbInboundOutbound).count() == 1
    new = db.get(DbPurchasePlan, link.db_purchase_plan_id)
    assert new.id != old.id and old.planned_quantity == 20 and new.planned_quantity == 8
    assert inventory.purchase_plan_id == new.id and inventory.date == date(2026, 11, 2)
    sync(db, rows)
    assert new.planned_quantity == 8 and db.query(DbPurchasePlan).count() == 2


def add_weighing(db, inventory):
    slip = DbWeighingSlip(id='weigh', tenant_no='t', inbound_outbound_id=inventory.id,
             vehicle_no='Plate', cargo_name='Product', business_type='出库', supplier='S',
             consignee='C', gross_weight=10000, tare_weight=5200, net_weight=4800,
             gross_weight_time=datetime(2026, 10, 1), is_voided=0)
    inventory.quantity = 4.8
    db.add(slip)
    db.commit()
    return slip


def test_identical_repeat_preserves_actual_weighing_quantity(db):
    rows = seed(db)
    first = sync(db, rows)
    link, inventory, monthly = records(db)
    slip = add_weighing(db, inventory)
    before = (slip.update_time, inventory.update_time)
    assert sync(db, rows) == first
    assert inventory.quantity == 4.8 and monthly.planned_quantity == 5
    assert link.contributed_qty == 5 and slip.net_weight == 4800
    assert (slip.update_time, inventory.update_time) == before
    assert db.query(DbInboundOutbound).count() == 1


@pytest.mark.parametrize('change', ['quantity', 'month', 'customer', 'plate', 'direction'])
def test_weighed_business_changes_rejected_without_mutation(db, change):
    rows = seed(db)
    sync(db, rows)
    link, inventory, monthly = records(db)
    add_weighing(db, inventory)
    if change == 'quantity': rows[1].quantity = 8
    elif change == 'month': rows[0].pickup_time = datetime(2026, 11, 1)
    elif change == 'customer': rows[0].customer_name = 'Changed'
    elif change == 'plate': rows[2].license_plate = 'Changed'
    else:
        rows[0].plan_type = 'inbound'
        rows[0].supplier_name = 'Customer'
    db.commit()
    with pytest.raises(ValueError):
        sync(db, rows)
    assert not db.dirty and not db.new
    assert inventory.quantity == 4.8 and monthly.planned_quantity == 5


@pytest.mark.parametrize('bad', ['duplicate', 'deleted_link', 'deleted_inventory', 'missing_inventory',
                                'foreign_inventory', 'missing_month', 'deleted_month', 'mismatch_month'])
def test_invalid_history_rejected_instead_of_creating_another_link(db, bad):
    rows = seed(db)
    sync(db, rows)
    link, inventory, monthly = records(db)
    if bad == 'duplicate':
        db.add(PlTransportDbLink(id='duplicate', tenant_no='t', transport_id='tr', plan_id='p',
                plan_product_id='pp', db_inventory_id=inventory.id, db_purchase_plan_id=monthly.id))
    elif bad == 'deleted_link': link.del_flag = 1
    elif bad == 'deleted_inventory': inventory.del_flag = 1
    elif bad == 'missing_inventory': link.db_inventory_id = 'missing'
    elif bad == 'foreign_inventory': inventory.tenant_no = 'other'
    elif bad == 'missing_month': link.db_purchase_plan_id = inventory.purchase_plan_id = 'missing'
    elif bad == 'deleted_month': monthly.del_flag = 1
    else: inventory.purchase_plan_id = 'different'
    db.commit()
    with pytest.raises(ValueError):
        sync(db, rows)
    assert not db.dirty and not db.new
    assert monthly.planned_quantity == 5


def test_multiple_lines_share_one_month_plan_and_validate_all_before_mutation(db):
    rows = seed(db)
    second = PlPlanProduct(id='pp2', tenant_no='t', plan_id='p', product_id='sku',
                            product_no='SKU', product_name='Product', model='Spec', unit='吨', quantity=3)
    db.add(second)
    db.commit()
    sync(db, rows, [rows[1], second])
    assert db.query(DbPurchasePlan).count() == 1
    monthly = db.query(DbPurchasePlan).one()
    assert monthly.planned_quantity == 8
    sync(db, rows, [rows[1], second])
    assert monthly.planned_quantity == 8 and db.query(DbInboundOutbound).count() == 2
    rows[1].quantity = 9
    bad_link = db.query(PlTransportDbLink).filter_by(plan_product_id='pp2').one()
    bad_link.db_inventory_id = 'missing'
    db.commit()
    with pytest.raises(ValueError):
        sync(db, rows, [rows[1], second])
    assert monthly.planned_quantity == 8 and not db.dirty and not db.new


def test_duplicate_month_plan_is_rejected_before_new_inventory(db):
    rows = seed(db)
    for identity in ['month1', 'month2']:
        db.add(DbPurchasePlan(id=identity, tenant_no='t', plan_date=date(2026, 10, 1),
            start_date=date(2026, 10, 1), end_date=date(2026, 10, 31), business_type='出库',
            product_id='sku', product='Product', specification='Spec', customer_name='Customer',
            planned_quantity=0))
    db.commit()
    with pytest.raises(ValueError):
        sync(db, rows)
    assert not db.new and not db.dirty and db.query(DbInboundOutbound).count() == 0


@pytest.mark.parametrize('bad', ['duplicate_input', 'foreign_tenant', 'foreign_plan', 'negative'])
def test_invalid_product_input_rejected_before_new_inventory(db, bad):
    rows = seed(db)
    if bad == 'foreign_tenant': rows[1].tenant_no = 'other'
    elif bad == 'foreign_plan': rows[1].plan_id = 'other'
    elif bad == 'negative': rows[1].quantity = -1
    db.commit()
    with pytest.raises(ValueError):
        sync(db, rows, [rows[1], rows[1]] if bad == 'duplicate_input' else None)
    assert not db.new and not db.dirty and db.query(DbInboundOutbound).count() == 0


def test_pending_direction_change_remounts_without_replacing_inventory(db):
    rows = seed(db)
    sync(db, rows)
    link, inventory, old = records(db)
    rows[0].plan_type = 'inbound'
    rows[0].supplier_name = 'Customer'
    db.commit()
    sync(db, rows)
    new = db.get(DbPurchasePlan, link.db_purchase_plan_id)
    assert new.id != old.id and old.planned_quantity == 0 and new.planned_quantity == 5
    assert inventory.business_type == '入库' and inventory.status == '待入库'
    assert db.query(DbInboundOutbound).count() == 1


def test_unweighed_final_status_still_protects_business_fields(db):
    rows = seed(db)
    sync(db, rows)
    link, inventory, monthly = records(db)
    inventory.status = '已出库'
    db.commit()
    sync(db, rows)
    rows[1].quantity = 8
    db.commit()
    with pytest.raises(ValueError):
        sync(db, rows)
    assert monthly.planned_quantity == 5 and inventory.quantity == 5


def test_missing_contributed_quantity_is_backfilled_only_before_weighing(db):
    rows = seed(db)
    sync(db, rows)
    link, inventory, monthly = records(db)
    link.contributed_qty = None
    db.commit()
    sync(db, rows)
    assert link.contributed_qty == 5 and monthly.planned_quantity == 5
    add_weighing(db, inventory)
    link.contributed_qty = None
    db.commit()
    with pytest.raises(ValueError):
        sync(db, rows)
    assert inventory.quantity == 4.8 and not db.dirty


def test_lock_refreshes_cached_quantities_without_losing_pending_feedback(db):
    rows = seed(db)
    sync(db, rows)
    link, inventory, monthly = records(db)
    # Simulate values newer than this Session's already-loaded identity map.
    db.execute(update(DbPurchasePlan).where(DbPurchasePlan.id == monthly.id)
               .values(planned_quantity=20).execution_options(synchronize_session=False))
    db.execute(update(PlTransportDbLink).where(PlTransportDbLink.id == link.id)
               .values(contributed_qty=4).execution_options(synchronize_session=False))
    assert monthly.planned_quantity == 5 and link.contributed_qty == 5
    inventory.site_feedback_remark = 'pending feedback'
    rows[1].quantity = 6
    sync(db, rows)
    assert monthly.planned_quantity == 22 and link.contributed_qty == 6
    assert inventory.site_feedback_remark == 'pending feedback'


def test_combined_old_contributions_cannot_exceed_month_quantity(db):
    rows = seed(db)
    second = PlPlanProduct(id='pp2', tenant_no='t', plan_id='p', product_id='sku',
                            product_no='SKU', product_name='Product', model='Spec', unit='吨', quantity=3)
    db.add(second)
    db.commit()
    sync(db, rows, [rows[1], second])
    monthly = db.query(DbPurchasePlan).one()
    monthly.planned_quantity = 6  # Each old contribution fits separately; their sum does not.
    rows[0].pickup_time = datetime(2026, 11, 1)
    db.commit()
    with pytest.raises(ValueError):
        sync(db, rows, [rows[1], second])
    assert monthly.planned_quantity == 6 and db.query(DbPurchasePlan).count() == 1
    assert not db.dirty and not db.new