301 lines
13 KiB
Python
301 lines
13 KiB
Python
"""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,
|
|
"prepare_quote_products": lambda *_args, **_kw: None,
|
|
"sync_quote_products": lambda *_args, **_kw: None,
|
|
"sync_approval_notifications": 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
|