118 lines
4.5 KiB
Python
118 lines
4.5 KiB
Python
"""Dramatiq 调度任务测试。
|
||
|
||
方式：
|
||
- TASK_BROKER=stub：进程内 StubBroker，验证 actor 注册与消息入队；
|
||
- 直接调用 actor（dramatiq actor 可调用，同步执行函数体）验证业务结果，
|
||
避免测试进程内 Worker 线程与 SQLite 事务锁冲突。
|
||
"""
|
||
|
||
import pytest
|
||
from datetime import date, timedelta
|
||
from decimal import Decimal
|
||
from model_bakery import baker
|
||
|
||
import dramatiq
|
||
from dramatiq.brokers.stub import StubBroker
|
||
from dramatiq.middleware import AgeLimit, Callbacks, Retries, TimeLimit
|
||
|
||
# 显式安装 StubBroker（覆盖 core.ready 可能已装的 RedisBroker）
|
||
broker = StubBroker(middleware=[AgeLimit(), TimeLimit(), Callbacks(), Retries()])
|
||
dramatiq.set_broker(broker)
|
||
|
||
# 在 stub broker 就绪后导入 actors（完成注册）
|
||
from apps.notify.tasks import ( # noqa: E402
|
||
run_alert_checks_for_tenant,
|
||
run_all_alert_checks_for_all_tenants,
|
||
)
|
||
from apps.channel.tasks import ( # noqa: E402
|
||
sync_orders_for_account,
|
||
sync_orders_for_all_active_accounts,
|
||
)
|
||
|
||
from apps.catalog.models import Product # noqa: E402
|
||
from apps.inventory.models import Warehouse, Stock # noqa: E402
|
||
from apps.partner.models import Customer # noqa: E402
|
||
from apps.finance.models import Receivable # noqa: E402
|
||
from apps.notify.models import Notification # noqa: E402
|
||
from apps.channel.models import ChannelAccount # noqa: E402
|
||
from apps.sales.models import SalesOrder # noqa: E402
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _flush_broker():
|
||
broker.flush_all()
|
||
yield
|
||
broker.flush_all()
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_actors_registered():
|
||
names = {a.actor_name for a in broker.actors.values()}
|
||
assert "run_alert_checks_for_tenant" in names
|
||
assert "run_all_alert_checks_for_all_tenants" in names
|
||
assert "sync_orders_for_account" in names
|
||
assert "sync_orders_for_all_active_accounts" in names
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_alert_actor_enqueues_message(tenant):
|
||
run_alert_checks_for_tenant.send(tenant.id)
|
||
assert broker.queues["default"].qsize() == 1
|
||
|
||
import json
|
||
raw = broker.queues["default"].queue[0]
|
||
msg = json.loads(raw)
|
||
assert msg["actor_name"] == "run_alert_checks_for_tenant"
|
||
assert msg["args"] == [tenant.id]
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_alert_actor_inline_execution_triggers_overdue_alert(tenant):
|
||
from apps.notify.models import AlertRule
|
||
baker.make(AlertRule, tenant=tenant, code="receivable_overdue_30d",
|
||
name="逾期催收", rule_type="receivable_overdue", is_enabled=True)
|
||
|
||
cust = baker.make(Customer, tenant=tenant, code="C_TQ", name="逾期客户")
|
||
baker.make(
|
||
Receivable,
|
||
tenant=tenant, customer=cust, bill_no="RC-TQ-01",
|
||
bill_date=date.today() - timedelta(days=60),
|
||
due_date=date.today() - timedelta(days=10),
|
||
total_amount=Decimal("500"), paid_amount=Decimal("0"), status="open",
|
||
)
|
||
|
||
result = run_alert_checks_for_tenant(tenant.id) # 直接调用 = 同步执行
|
||
assert result["receivable_alerts_count"] == 1
|
||
assert Notification.objects.filter(tenant=tenant, category="warning").count() == 1
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_all_tenants_alert_actor_inline(tenant, other_tenant):
|
||
for t in (tenant, other_tenant):
|
||
wh = baker.make(Warehouse, tenant=t, code=f"W_{t.code}")
|
||
p = baker.make(Product, tenant=t, code=f"P_{t.code}", name=f"缺货商品{t.code}")
|
||
baker.make(Stock, tenant=t, warehouse=wh, product=p, on_hand=Decimal("2"), locked=Decimal("0"))
|
||
|
||
summary = run_all_alert_checks_for_all_tenants()
|
||
by_code = {row["tenant"]: row for row in summary}
|
||
assert by_code[tenant.code]["stock_alerts_count"] == 1
|
||
assert by_code[other_tenant.code]["stock_alerts_count"] == 1
|
||
assert Notification.objects.filter(tenant=tenant, category="warning").count() == 1
|
||
assert Notification.objects.filter(tenant=other_tenant, category="warning").count() == 1
|
||
|
||
|
||
@pytest.mark.django_db
|
||
def test_channel_sync_actor_inline_with_mock_fallback(tenant):
|
||
account = baker.make(
|
||
ChannelAccount, tenant=tenant, platform="douyin",
|
||
shop_id="DY_TASK", shop_name="任务店铺",
|
||
app_key="", access_token="", # 无凭证 → Mock 回退
|
||
)
|
||
baker.make(Product, tenant=tenant, code="P_TASK", status="active", sale_price=Decimal("30.00"))
|
||
|
||
result = sync_orders_for_account(account.id)
|
||
assert result["pulled_count"] == 1
|
||
assert result["converted_count"] == 1
|
||
assert SalesOrder.objects.filter(tenant=tenant, remark__contains=result["pulled_order"] if "pulled_order" in result else "").exists() or \
|
||
SalesOrder.objects.filter(tenant=tenant).count() == 1
|