shanxia2.0/backend/tests/test_mobile_quote_contact_sync.py

"""Execute production routes/CRUD with an in-memory Session double, never an Engine."""
from __future__ import annotations

import ast
import asyncio
import builtins
from datetime import datetime, timedelta, date
from decimal import Decimal
from pathlib import Path
from types import SimpleNamespace

import pytest

from app.schemas.crm.quote import QuoteUpdateBody


ROOT = Path(__file__).resolve().parents[1]


def load_functions(path, names, bindings):
    tree = ast.parse((ROOT / path).read_text())
    nodes = [node for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names]
    module = ast.Module(body=[ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0), *nodes], type_ignores=[])
    exec(compile(ast.fix_missing_locations(module), str(path), "exec"), bindings)


class Field:
    def __init__(self, name):
        self.name = name

    def __eq__(self, value):
        return lambda row: getattr(row, self.name) == value

    def desc(self):
        return self


class Model(type):
    def __getattr__(cls, name):
        return Field(name)


class Row(SimpleNamespace):
    def __getattr__(self, name):
        return None


class Query:
    def __init__(self, rows):
        self.rows = list(rows)

    def filter(self, *predicates):
        self.rows = [row for row in self.rows if all(p(row) for p in predicates)]
        return self

    def order_by(self, *_args):
        return self

    def first(self):
        return next(iter(self.rows), None)

    def all(self):
        return self.rows


class SessionDouble:
    def __init__(self, rows):
        self.rows = rows
        self.commits = 0
        self.flushes = 0
        self.approval_invalidations = 0

    def query(self, model):
        return Query(self.rows[model.__name__])

    def flush(self):
        self.flushes += 1

    def commit(self):
        self.commits += 1


class Router:
    def __init__(self):
        self.routes = {}

    def post(self, path, **_kwargs):
        def register(fn):
            self.routes[path] = fn
            return fn
        return register


@pytest.fixture
def h(monkeypatch):
    # Deny both network access and accidental SQLAlchemy engine construction/connection.
    def forbidden(*_args, **_kwargs):
        raise AssertionError("Network/Engine access is forbidden in mobile regression tests")

    import socket
    import sqlalchemy
    monkeypatch.setattr(socket.socket, "connect", forbidden)
    monkeypatch.setattr(sqlalchemy, "create_engine", forbidden)
    monkeypatch.setattr(sqlalchemy.engine.Engine, "connect", forbidden)

    quote = Row(id="q1", tenant_no="tenant-a", del_flag=0, customer_id="customer-old", contact_id="contact-old", contact_name="旧姓名", contact_mobile="111", contact_email="old@example.test", status="draft", owner_user_id="user-a", quote_type="sales", quote_no="SXDD20261021001")
    customers = [Row(id=id, tenant_no="tenant-a", del_flag=0, owner_user_id="user-a", short_name=id) for id in ["customer-old", "customer-new"]]
    contacts = [Row(id="contact-old", customer_id="customer-old", tenant_no="tenant-a", del_flag=0, contact_name="旧姓名", mobile="111", email="old@example.test"), Row(id="contact-new", customer_id="customer-new", tenant_no="tenant-a", del_flag=0, contact_name="新姓名", mobile="222", email="new@example.test")]
    db = SessionDouble({"CrmQuote": [quote], "CrmCustomer": customers, "CrmContact": contacts, "SysUser": []})
    ns = {
        "datetime": datetime, "timedelta": timedelta, "date": date, "Decimal": Decimal,
        "_FMT": "%Y-%m-%d %H:%M:%S", "_DATE_FMT": "%Y-%m-%d",
        "QUOTE_EDITABLE_STATUSES": frozenset({"draft", "submitted", "accepted", "rejected"}),
        "CRM_OWNER_FIELD": "owner_user_id",
        "QuoteUpdateBody": QuoteUpdateBody,
        "row_in_data_scope": lambda _db, row, user, tenant, **_kw: row.tenant_no == tenant and row.owner_user_id == user["user_id"],
        "get_tenant_no": lambda _db, user: user["tenant_no"],
        "Depends": lambda dep: dep,
        "get_db": forbidden,
        "require_admin_or_super_admin": forbidden,
        "require_menu_code": lambda *_args, **_kw: forbidden,
        "success_response": lambda data, message="success": {"code": 200, "data": data, "message": message},
        "error_response": lambda code, message: {"code": code, "message": message},
        "list_quote_products": lambda *_args, **_kw: [],
        "get_supplier_info": lambda *_args, **_kw: None,
        "assert_products_allowed_for_supplier": lambda *_args, **_kw: None,
        "_resolve_quote_product_ids": lambda *_args, **_kw: None,
        "delete_products_by_quote_id": lambda *_args, **_kw: None,
        "batch_insert_quote_products": lambda *_args, **_kw: None,
        "invalidate_all_approvals": lambda *_args, **_kw: setattr(db, "approval_invalidations", db.approval_invalidations + 1),
    }
    for name in ["CrmQuote", "CrmCustomer", "CrmContact", "SysUser"]:
        ns[name] = Model(name, (), {})
    load_functions("app/crud/crm/data_scope_helpers.py", {"crm_user_info", "ensure_crm_owner_row_in_scope"}, ns)
    load_functions("app/crud/crm/customer.py", {"get_customer"}, ns)
    # The real contact helper imports get_customer lazily; supply the real extracted function.
    real_import = builtins.__import__
    def isolated_import(name, *args, **kwargs):
        if name == "app.crud.crm.customer":
            return SimpleNamespace(get_customer=ns["get_customer"])
        return real_import(name, *args, **kwargs)
    ns["__builtins__"] = {**vars(builtins), "__import__": isolated_import}
    contact_ns = dict(ns)
    load_functions("app/crud/crm/contact.py", {"_row_to_dict", "get_contact", "list_contacts_by_customer"}, contact_ns)
    ns.update({key: contact_ns[key] for key in ["get_contact", "list_contacts_by_customer"]})
    load_functions("app/crud/crm/quote.py", {"_row_to_dict", "_get_field", "_decimal_value", "_parse_optional_date", "_ensure_quote_in_scope", "_get_quote_in_scope", "get_quote", "get_quote_detail", "assert_quote_editable_for_update", "update_quote", "_enrich_contact_fields", "_enrich_supplier_name", "_get_supplier_short_name", "generate_quote_title", "calculate_quote_amount"}, ns)
    route_tree = ast.parse((ROOT / "app/api/crm_quote_routes.py").read_text())
    load_functions("app/api/crm_quote_routes.py", {node.name for node in route_tree.body if isinstance(node, ast.FunctionDef)}, ns)
    router = Router()
    ns["register_quote_routes"](router)
    user = {"user_id": "user-a", "tenant_no": "tenant-a", "role": "admin"}

    def call(payload, mobile=True):
        path = "/quote/mobile-update" if mobile else "/quote/update"
        assert path in router.routes, "The mobile contact-safe update entry is missing"
        body = QuoteUpdateBody(**{"id": "q1", "customerId": "customer-new", "contactId": "contact-new", **payload})
        return asyncio.run(router.routes[path](body, _user=user, db=db))

    return SimpleNamespace(call=call, quote=quote, customers=customers, contacts=contacts, db=db, ns=ns, router=router)


def test_mobile_save_uses_canonical_contact_and_keeps_product_amount(h):
    result = h.call({"contactName": "伪造姓名", "tenantNo": "tenant-b", "freight": 5, "barrelDeposit": 3, "products": [{"quantity": 2, "price": 12.5}]})
    assert result["code"] == 200, result
    assert (h.quote.contact_id, h.quote.contact_name, h.quote.contact_mobile, h.quote.contact_email) == ("contact-new", "新姓名", "222", "new@example.test")
    assert result["data"]["finalAmount"] == 25
    assert h.quote.tenant_no == "tenant-a"
    assert h.db.commits == 1


@pytest.mark.parametrize("missing", ["mobile", "email", "both"])
def test_new_contact_empty_fields_do_not_retain_previous_snapshot(h, missing):
    if missing in {"mobile", "both"}:
        h.contacts[1].mobile = ""
    if missing in {"email", "both"}:
        h.contacts[1].email = None
    result = h.call({})
    assert result["code"] == 200, result
    assert h.quote.contact_mobile == ("" if missing in {"mobile", "both"} else "222")
    assert h.quote.contact_email == ("" if missing in {"email", "both"} else "new@example.test")


@pytest.mark.parametrize("contact_id", [None, "", "   "])
def test_explicit_clear_removes_the_whole_snapshot(h, contact_id):
    result = h.call({"contactId": contact_id})
    assert result["code"] == 200, result
    assert (h.quote.contact_id, h.quote.contact_name, h.quote.contact_mobile, h.quote.contact_email) == ("", "", "", "")


@pytest.mark.parametrize("customer_id, expected_id", [("customer-old", "contact-old"), ("customer-new", "")])
def test_omitted_contact_keeps_same_customer_but_clears_for_changed_customer(h, customer_id, expected_id):
    path = "/quote/mobile-update"
    assert path in h.router.routes, "The mobile contact-safe update entry is missing"
    body = QuoteUpdateBody(id="q1", customerId=customer_id)
    result = asyncio.run(h.router.routes[path](body, _user={"user_id": "user-a", "tenant_no": "tenant-a", "role": "admin"}, db=h.db))
    assert result["code"] == 200, result
    assert h.quote.contact_id == expected_id
    assert h.quote.contact_name == ("旧姓名" if expected_id else "")


@pytest.mark.parametrize("contact_mode", ["null", "omitted"])
@pytest.mark.parametrize("invalid_customer", ["deleted", "wrong-tenant", "out-of-scope"])
def test_empty_contact_still_validates_customer_before_any_mutation(
    h, contact_mode, invalid_customer
):
    if invalid_customer == "deleted":
        h.customers[1].del_flag = 1
    elif invalid_customer == "wrong-tenant":
        h.customers[1].tenant_no = "tenant-b"
    else:
        h.customers[1].owner_user_id = "user-b"
    payload = {"id": "q1", "customerId": "customer-new"}
    if contact_mode == "null":
        payload["contactId"] = None
    before = dict(vars(h.quote))
    result = asyncio.run(
        h.router.routes["/quote/mobile-update"](
            QuoteUpdateBody(**payload),
            _user={"user_id": "user-a", "tenant_no": "tenant-a", "role": "admin"},
            db=h.db,
        )
    )
    assert result["code"] == 400, result
    assert vars(h.quote) == before
    assert (h.db.commits, h.db.flushes, h.db.approval_invalidations) == (0, 0, 0)


@pytest.mark.parametrize("invalid", ["missing", "deleted", "wrong-customer", "wrong-tenant", "customer-deleted", "customer-wrong-tenant", "customer-out-of-scope", "quote-out-of-scope", "quote-wrong-tenant", "quote-not-editable"])
def test_invalid_selection_rejects_before_quote_or_approval_persistence(h, invalid):
    if invalid == "missing":
        h.contacts[1].id = "different"
    elif invalid == "deleted":
        h.contacts[1].del_flag = 1
    elif invalid == "wrong-customer":
        h.contacts[1].customer_id = "customer-old"
    elif invalid == "wrong-tenant":
        h.contacts[1].tenant_no = "tenant-b"
    elif invalid == "customer-deleted":
        h.customers[1].del_flag = 1
    elif invalid == "customer-wrong-tenant":
        h.customers[1].tenant_no = "tenant-b"
    elif invalid == "customer-out-of-scope":
        h.customers[1].owner_user_id = "user-b"
    elif invalid == "quote-out-of-scope":
        h.quote.owner_user_id = "user-b"
    elif invalid == "quote-wrong-tenant":
        h.quote.tenant_no = "tenant-b"
    elif invalid == "quote-not-editable":
        h.quote.status = "invalid"
    before = dict(vars(h.quote))
    result = h.call({})
    assert result["code"] == 400, result
    assert vars(h.quote) == before
    assert (h.db.commits, h.db.flushes, h.db.approval_invalidations) == (0, 0, 0)


def test_customer_used_for_contact_validation_matches_the_persisted_id(h):
    before = dict(vars(h.quote))
    result = h.call({"customerId": " customer-new "})
    assert result["code"] == 400, result
    assert vars(h.quote) == before
    assert h.db.commits == 0


@pytest.mark.parametrize("contact_id", ["0", "42", "9007199254740993", "uuid-contact"])
def test_string_ids_keep_precision_and_resolve_the_same_contact(h, contact_id):
    h.contacts[1].id = contact_id
    result = h.call({"contactId": contact_id})
    assert result["code"] == 200, result
    assert h.quote.contact_id == contact_id
    assert h.quote.contact_name == "新姓名"


def test_mobile_and_web_routes_share_the_existing_auth_and_schema(h):
    import inspect
    mobile = inspect.signature(h.router.routes["/quote/mobile-update"])
    web = inspect.signature(h.router.routes["/quote/update"])
    assert mobile.parameters["_user"].default is web.parameters["_user"].default
    assert mobile.parameters["db"].default is web.parameters["db"].default
    assert mobile.parameters["body"].annotation == web.parameters["body"].annotation


def test_original_web_update_retains_existing_schema_and_default_behavior(h):
    result = h.call({"contactName": "客户端姓名"}, mobile=False)
    assert result["code"] == 200, result
    assert "contactName" not in QuoteUpdateBody.model_fields
    assert h.quote.contact_name == "旧姓名"
    assert h.quote.contact_mobile == "222"
    assert h.db.commits == 1


@pytest.mark.parametrize("status", ["draft", "submitted", "accepted", "rejected"])
def test_mobile_reuses_existing_reedit_to_draft_rules(h, status):
    h.quote.status = status
    result = h.call({})
    assert result["code"] == 200, result
    assert h.quote.status == "draft"
    assert h.db.approval_invalidations == 1