wehub-resource-sync
813 lines
28 KiB
Python
813 lines
28 KiB
Python
"""Payment suite offline judge tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import datetime
|
|
import inspect
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
import pytest
|
|
|
|
from bench_env.task.base import BaseTask
|
|
from bench_env.task.common_tasks import AnswerTask
|
|
from bench_env.task.payment import tasks as payment_tasks
|
|
from bench_env.tests.conftest import make_judge_input
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
DEFAULT_ROUTE = {"app": "launcher", "path": "/"}
|
|
TEST_TS = int(datetime.datetime(2026, 3, 27, 12, 0, 0).timestamp() * 1000)
|
|
BASE_OS_STATE = {
|
|
"time": {"timestamp": TEST_TS},
|
|
"providers": {"sms": {"conversations": [], "messagesByConversationId": {}}},
|
|
"runningApps": [],
|
|
}
|
|
|
|
|
|
def _load_json(*parts: str) -> dict[str, Any]:
|
|
return json.loads(Path(ROOT, *parts).read_text(encoding="utf-8"))
|
|
|
|
|
|
ALIPAY_DEFAULTS = _load_json("apps", "Alipay", "data", "defaults.json")
|
|
WECHAT_DEFAULTS = _load_json("apps", "Wechat", "data", "defaults.json")
|
|
BILIBILI_DEFAULTS = _load_json("apps", "Bilibili", "data", "defaults.json")
|
|
RAILWAY_DEFAULTS = _load_json("apps", "Railway12306", "data", "defaults.json")
|
|
|
|
|
|
ALL_TASK_CLASSES: list[type[BaseTask]] = [
|
|
obj
|
|
for _, obj in inspect.getmembers(payment_tasks, inspect.isclass)
|
|
if issubclass(obj, BaseTask) and obj is not BaseTask and obj.__module__ == payment_tasks.__name__
|
|
]
|
|
ALL_TASK_IDS = [cls.__name__ for cls in ALL_TASK_CLASSES]
|
|
ANSWER_TASK_CLASSES = [cls for cls in ALL_TASK_CLASSES if issubclass(cls, AnswerTask)]
|
|
|
|
|
|
def _copy(value: Any) -> Any:
|
|
return copy.deepcopy(value)
|
|
|
|
|
|
def _base_notes_state() -> dict[str, Any]:
|
|
return {
|
|
"notes": [],
|
|
"todos": [],
|
|
"folders": [{"id": "unfiled", "name": "未分类"}],
|
|
"settings": {"notesViewMode": "grid", "showWordCount": True},
|
|
}
|
|
|
|
|
|
def _today_ymd() -> str:
|
|
return datetime.datetime.fromtimestamp(TEST_TS / 1000).date().isoformat()
|
|
|
|
|
|
def _tomorrow_ymd() -> str:
|
|
return (datetime.datetime.fromtimestamp(TEST_TS / 1000).date() + datetime.timedelta(days=1)).isoformat()
|
|
|
|
|
|
def _make_input(
|
|
*,
|
|
apps_init: dict[str, Any],
|
|
apps_curr: dict[str, Any],
|
|
os_init: dict[str, Any] | None = None,
|
|
os_curr: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
return make_judge_input(
|
|
{"apps": apps_init, "os": _copy(os_init or BASE_OS_STATE)},
|
|
{"apps": apps_curr, "os": _copy(os_curr or os_init or BASE_OS_STATE)},
|
|
route=DEFAULT_ROUTE,
|
|
)
|
|
|
|
|
|
def _append_note(state: dict[str, Any], *, title: str, content: str) -> dict[str, Any]:
|
|
next_state = _copy(state)
|
|
notes = next_state.setdefault("notes", [])
|
|
notes.append(
|
|
{
|
|
"id": f"note_{len(notes) + 1}",
|
|
"title": title,
|
|
"content": content,
|
|
"updatedAt": TEST_TS + len(notes) + 1,
|
|
"folderId": "unfiled",
|
|
}
|
|
)
|
|
return next_state
|
|
|
|
|
|
def _ensure_wechat_chat(state: dict[str, Any], contact_name: str) -> tuple[dict[str, Any], str]:
|
|
contacts = state.setdefault("contacts", [])
|
|
contact = next((item for item in contacts if str(item.get("name") or "") == contact_name), None)
|
|
if contact is None:
|
|
raise ValueError(f"wechat fixture missing contact: {contact_name}")
|
|
wxid = str(contact["wxid"])
|
|
chats = state.setdefault("chats", [])
|
|
chat = next((item for item in chats if str(item.get("id") or "") == wxid), None)
|
|
if chat is None:
|
|
chat = {
|
|
"id": wxid,
|
|
"name": contact_name,
|
|
"contactName": contact_name,
|
|
"user": {"wxid": wxid, "name": contact_name},
|
|
"isMuted": False,
|
|
"isSticky": False,
|
|
"isAlert": False,
|
|
"messages": [],
|
|
}
|
|
chats.append(chat)
|
|
return chat, wxid
|
|
|
|
|
|
def _append_wechat_text(
|
|
state: dict[str, Any],
|
|
*,
|
|
contact_name: str,
|
|
content: str,
|
|
outgoing: bool,
|
|
) -> dict[str, Any]:
|
|
next_state = _copy(state)
|
|
chat, wxid = _ensure_wechat_chat(next_state, contact_name)
|
|
sender = str(next_state.get("user", {}).get("wxid") or "") if outgoing else wxid
|
|
messages = chat.setdefault("messages", [])
|
|
messages.append(
|
|
{
|
|
"id": f"msg_{len(messages) + 1}",
|
|
"type": "text",
|
|
"content": content,
|
|
"senderId": sender,
|
|
"timestamp": TEST_TS + len(messages) + 1,
|
|
}
|
|
)
|
|
return next_state
|
|
|
|
|
|
def _append_alipay_record(
|
|
state: dict[str, Any],
|
|
*,
|
|
record_id: str,
|
|
kind: str,
|
|
delta: float,
|
|
method_id: str,
|
|
counterparty: str,
|
|
timestamp: int,
|
|
transfer_note: str | None = None,
|
|
target_account: str | None = None,
|
|
order_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
next_state = _copy(state)
|
|
record: dict[str, Any] = {
|
|
"id": record_id,
|
|
"counterpartyName": counterparty,
|
|
"displayTitle": counterparty,
|
|
"delta": float(delta),
|
|
"timestamp": int(timestamp),
|
|
"methodId": method_id,
|
|
"kind": kind,
|
|
}
|
|
if transfer_note is not None:
|
|
record["transferNote"] = transfer_note
|
|
if target_account is not None:
|
|
record["targetAccount"] = target_account
|
|
if order_id is not None:
|
|
record["orderId"] = order_id
|
|
next_state.setdefault("transferRecords", [])
|
|
next_state["transferRecords"] = [record, *next_state["transferRecords"]]
|
|
return next_state
|
|
|
|
|
|
def _rail_order(
|
|
*,
|
|
order_id: str,
|
|
status: str,
|
|
date: str,
|
|
from_station: str = "上海",
|
|
to_station: str = "南京",
|
|
train_no: str = "G1001",
|
|
passenger: str = "赵宇轩",
|
|
seat_type: str = "二等座",
|
|
seat_no: str = "06车 15B号",
|
|
price: float = 120.0,
|
|
depart_time: str = "08:00",
|
|
arrive_time: str = "09:25",
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"id": order_id,
|
|
"trainNo": train_no,
|
|
"fromStation": from_station,
|
|
"toStation": to_station,
|
|
"departTime": depart_time,
|
|
"arriveTime": arrive_time,
|
|
"date": date,
|
|
"status": status,
|
|
"createTime": f"{date}T07:00:00.000Z",
|
|
"tickets": [
|
|
{
|
|
"passengerName": passenger,
|
|
"ticketType": "成人票",
|
|
"seatType": seat_type,
|
|
"seatNo": seat_no,
|
|
"price": float(price),
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
class TestTaskDefinitions:
|
|
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
|
def test_instantiation(self, cls: type[BaseTask]) -> None:
|
|
task = cls()
|
|
assert task.name == cls.__name__
|
|
assert task.templates
|
|
assert task.apps
|
|
|
|
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
|
def test_description_renders(self, cls: type[BaseTask]) -> None:
|
|
task = cls()
|
|
desc = task.description
|
|
assert desc
|
|
assert "{" not in desc
|
|
|
|
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
|
def test_task_docstring_present(self, cls: type[BaseTask]) -> None:
|
|
assert cls.__doc__ is not None
|
|
assert cls.__doc__.strip()
|
|
|
|
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
|
def test_meta_attrs(self, cls: type[BaseTask]) -> None:
|
|
assert cls.scope in ("S1", "S2", "S3")
|
|
assert cls.objective in ("operate", "query", "hybrid")
|
|
assert cls.composition in ("atomic", "sequential", "transfer", "deep_dive")
|
|
assert cls.difficulty in ("L1", "L2", "L3", "L4")
|
|
|
|
@pytest.mark.parametrize("cls", ALL_TASK_CLASSES, ids=ALL_TASK_IDS)
|
|
def test_parameters_have_defaults(self, cls: type[BaseTask]) -> None:
|
|
for key, schema in cls.parameters.items():
|
|
if key.startswith("_"):
|
|
continue
|
|
assert "default" in schema
|
|
|
|
@pytest.mark.parametrize("cls", ANSWER_TASK_CLASSES, ids=[c.__name__ for c in ANSWER_TASK_CLASSES])
|
|
def test_answer_task_has_answer(self, cls: type[BaseTask]) -> None:
|
|
# payment suite 当前无 AnswerTask，保留占位以防后续新增遗漏
|
|
pass
|
|
|
|
def test_multichannel_task_removed(self) -> None:
|
|
assert not hasattr(payment_tasks, "AlipayTransferMultiChannel")
|
|
|
|
def test_transfer_with_note_prompt_mentions_password(self) -> None:
|
|
task = payment_tasks.TransferToContactWithNote()
|
|
assert "我的密码是123456" in task.description
|
|
|
|
|
|
def _bind_multiple_cards_positive() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.AlipayBindMultipleCardsTransferAndRecordSuccessfulCards()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
init_alipay["balance"] = {"total": 200.0}
|
|
init_alipay["bankCards"] = [
|
|
{"id": "ccb", "bankName": "建设银行储蓄卡", "cardNumber": "6227000000000005445", "last4": "5445", "bound": True, "available": 20000},
|
|
{"id": "icbc", "bankName": "工商银行储蓄卡", "cardNumber": "6222000000000001234", "last4": "1234", "bound": False, "available": 500},
|
|
{"id": "abc", "bankName": "农业银行储蓄卡", "cardNumber": "6228480000000006789", "last4": "6789", "bound": False, "available": 20000},
|
|
]
|
|
init_alipay["transferRecords"] = []
|
|
init_alipay["lastPaymentHint"] = ""
|
|
curr_alipay = _copy(init_alipay)
|
|
for card in curr_alipay["bankCards"]:
|
|
card["bound"] = True
|
|
target = str(task.p.targetAccount)
|
|
curr_alipay = _append_alipay_record(
|
|
curr_alipay,
|
|
record_id="bind_tx_1",
|
|
kind="transfer",
|
|
delta=-float(task.p.amount1),
|
|
method_id="ccb",
|
|
counterparty=f"转账-{target}",
|
|
target_account=target,
|
|
transfer_note="第一笔",
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
curr_alipay = _append_alipay_record(
|
|
curr_alipay,
|
|
record_id="bind_tx_2",
|
|
kind="transfer",
|
|
delta=0.0,
|
|
method_id="icbc",
|
|
counterparty=f"转账-{target}",
|
|
target_account=target,
|
|
transfer_note="第二笔工行余额不足",
|
|
timestamp=TEST_TS + 2000,
|
|
)
|
|
curr_alipay = _append_alipay_record(
|
|
curr_alipay,
|
|
record_id="bind_tx_3",
|
|
kind="transfer",
|
|
delta=-float(task.p.amount3),
|
|
method_id="abc",
|
|
counterparty=f"转账-{target}",
|
|
target_account=target,
|
|
transfer_note="第三笔",
|
|
timestamp=TEST_TS + 3000,
|
|
)
|
|
init_notes = _base_notes_state()
|
|
curr_notes = _append_note(
|
|
init_notes,
|
|
title=str(task.p.noteTitle),
|
|
content="成功银行卡：建设银行储蓄卡（5445）、农业银行储蓄卡（6789）",
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay, "notes": init_notes},
|
|
apps_curr={"alipay": curr_alipay, "notes": curr_notes},
|
|
)
|
|
|
|
|
|
def _continuous_payments_positive() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.AlipayContinuousPaymentsToContactsRecordBalances()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
init_alipay["balance"] = {"total": 2000.0}
|
|
init_alipay["transferRecords"] = []
|
|
curr_alipay = _copy(init_alipay)
|
|
contacts = [
|
|
str(task.p.contact1),
|
|
str(task.p.contact2),
|
|
str(task.p.contact3),
|
|
str(task.p.contact4),
|
|
str(task.p.contact5),
|
|
]
|
|
amounts = [
|
|
float(task.p.amount1),
|
|
float(task.p.amount2),
|
|
float(task.p.amount3),
|
|
float(task.p.amount4),
|
|
float(task.p.amount5),
|
|
]
|
|
running = float(init_alipay["balance"]["total"])
|
|
balances: list[float] = []
|
|
for index, (contact, amount) in enumerate(zip(contacts, amounts, strict=True), start=1):
|
|
curr_alipay = _append_alipay_record(
|
|
curr_alipay,
|
|
record_id=f"cont_tx_{index}",
|
|
kind="transfer",
|
|
delta=-amount,
|
|
method_id="balance",
|
|
counterparty=contact,
|
|
target_account=contact,
|
|
transfer_note="发工资",
|
|
timestamp=TEST_TS + index * 1000,
|
|
)
|
|
running -= amount
|
|
balances.append(running)
|
|
curr_alipay["balance"]["total"] = balances[-1]
|
|
init_notes = _base_notes_state()
|
|
curr_notes = _append_note(
|
|
init_notes,
|
|
title=str(task.p.noteTitle),
|
|
content="\n".join(f"第{i}次后余额：{balance:g}" for i, balance in enumerate(balances, start=1)),
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay, "notes": init_notes},
|
|
apps_curr={"alipay": curr_alipay, "notes": curr_notes},
|
|
)
|
|
|
|
|
|
def _continuous_payments_negative() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.AlipayContinuousPaymentsToContactsRecordBalances()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
init_alipay["balance"] = {"total": 2000.0}
|
|
init_alipay["transferRecords"] = []
|
|
curr_alipay = _copy(init_alipay)
|
|
contacts = [
|
|
str(task.p.contact1),
|
|
str(task.p.contact2),
|
|
str(task.p.contact3),
|
|
str(task.p.contact4),
|
|
str(task.p.contact5),
|
|
]
|
|
amounts = [
|
|
float(task.p.amount1),
|
|
float(task.p.amount2),
|
|
float(task.p.amount3),
|
|
float(task.p.amount4),
|
|
float(task.p.amount5),
|
|
]
|
|
running = float(init_alipay["balance"]["total"])
|
|
recorded_balances: list[float] = []
|
|
for index, (contact, amount) in enumerate(zip(contacts, amounts, strict=True), start=1):
|
|
curr_alipay = _append_alipay_record(
|
|
curr_alipay,
|
|
record_id=f"cont_neg_tx_{index}",
|
|
kind="transfer",
|
|
delta=-amount,
|
|
method_id="balance",
|
|
counterparty=contact,
|
|
target_account=contact,
|
|
transfer_note="发工资",
|
|
timestamp=TEST_TS + index * 1000,
|
|
)
|
|
running -= amount
|
|
if index < 5:
|
|
recorded_balances.append(running)
|
|
curr_alipay["balance"]["total"] = running
|
|
init_notes = _base_notes_state()
|
|
curr_notes = _append_note(
|
|
init_notes,
|
|
title=str(task.p.noteTitle),
|
|
content="\n".join(f"第{i}次后余额：{balance:g}" for i, balance in enumerate(recorded_balances, start=1)),
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay, "notes": init_notes},
|
|
apps_curr={"alipay": curr_alipay, "notes": curr_notes},
|
|
)
|
|
|
|
|
|
def _bind_multiple_cards_negative() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.AlipayBindMultipleCardsTransferAndRecordSuccessfulCards()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
init_alipay["balance"] = {"total": 200.0}
|
|
init_alipay["bankCards"] = [
|
|
{"id": "ccb", "bankName": "建设银行储蓄卡", "cardNumber": "6227000000000005445", "last4": "5445", "bound": True, "available": 20000},
|
|
{"id": "icbc", "bankName": "工商银行储蓄卡", "cardNumber": "6222000000000001234", "last4": "1234", "bound": False, "available": 500},
|
|
{"id": "abc", "bankName": "农业银行储蓄卡", "cardNumber": "6228480000000006789", "last4": "6789", "bound": False, "available": 20000},
|
|
]
|
|
init_alipay["transferRecords"] = []
|
|
init_alipay["lastPaymentHint"] = ""
|
|
curr_alipay = _copy(init_alipay)
|
|
for card in curr_alipay["bankCards"]:
|
|
card["bound"] = True
|
|
target = str(task.p.targetAccount)
|
|
curr_alipay = _append_alipay_record(
|
|
curr_alipay,
|
|
record_id="bind_neg_tx_1",
|
|
kind="transfer",
|
|
delta=-float(task.p.amount1),
|
|
method_id="ccb",
|
|
counterparty=f"转账-{target}",
|
|
target_account=target,
|
|
transfer_note="第一笔",
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
curr_alipay = _append_alipay_record(
|
|
curr_alipay,
|
|
record_id="bind_neg_tx_2",
|
|
kind="transfer",
|
|
delta=-float(task.p.amount2),
|
|
method_id="icbc",
|
|
counterparty=f"转账-{target}",
|
|
target_account=target,
|
|
transfer_note="第二笔",
|
|
timestamp=TEST_TS + 2000,
|
|
)
|
|
init_notes = _base_notes_state()
|
|
curr_notes = _append_note(
|
|
init_notes,
|
|
title=str(task.p.noteTitle),
|
|
content="成功银行卡：建设银行储蓄卡（5445）、工商银行储蓄卡（1234）",
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay, "notes": init_notes},
|
|
apps_curr={"alipay": curr_alipay, "notes": curr_notes},
|
|
)
|
|
|
|
|
|
def _change_password_positive() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.AlipayChangePaymentPasswordThenPay()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
init_alipay.setdefault("userInfo", {})
|
|
init_alipay["userInfo"]["paymentPassword"] = "000000"
|
|
curr_alipay = _copy(init_alipay)
|
|
curr_alipay["userInfo"]["paymentPassword"] = str(task.p.newPassword)
|
|
curr_alipay = _append_alipay_record(
|
|
curr_alipay,
|
|
record_id="pwd_pay_1",
|
|
kind="payment",
|
|
delta=-float(task.p.amount),
|
|
method_id="balance",
|
|
counterparty=str(task.p.contact),
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay},
|
|
apps_curr={"alipay": curr_alipay},
|
|
)
|
|
|
|
|
|
def _change_password_negative() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.AlipayChangePaymentPasswordThenPay()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
init_alipay.setdefault("userInfo", {})
|
|
init_alipay["userInfo"]["paymentPassword"] = "000000"
|
|
curr_alipay = _copy(init_alipay)
|
|
curr_alipay["userInfo"]["paymentPassword"] = str(task.p.newPassword)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay},
|
|
apps_curr={"alipay": curr_alipay},
|
|
)
|
|
|
|
|
|
def _transfer_with_note_positive() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.TransferToContactWithNote()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
init_alipay["transferRecords"] = []
|
|
curr_alipay = _append_alipay_record(
|
|
init_alipay,
|
|
record_id="transfer_with_note_1",
|
|
kind="transfer",
|
|
delta=-float(task.p.amount),
|
|
method_id="balance",
|
|
counterparty=str(task.p.name),
|
|
target_account=str(task.p.name),
|
|
transfer_note=str(task.p.note),
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
curr_alipay["balance"]["total"] = round(
|
|
float(init_alipay["balance"]["total"]) - float(task.p.amount),
|
|
2,
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay},
|
|
apps_curr={"alipay": curr_alipay},
|
|
)
|
|
|
|
|
|
def _transfer_with_note_negative() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.TransferToContactWithNote()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
init_alipay["transferRecords"] = []
|
|
curr_alipay = _append_alipay_record(
|
|
init_alipay,
|
|
record_id="transfer_with_note_wrong_note",
|
|
kind="transfer",
|
|
delta=-float(task.p.amount),
|
|
method_id="balance",
|
|
counterparty=str(task.p.name),
|
|
target_account=str(task.p.name),
|
|
transfer_note="别的备注",
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
curr_alipay["balance"]["total"] = round(
|
|
float(init_alipay["balance"]["total"]) - float(task.p.amount),
|
|
2,
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay},
|
|
apps_curr={"alipay": curr_alipay},
|
|
)
|
|
|
|
|
|
def _subscribe_membership_positive() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.SubscribeMembershipAutoRenewThenCancelInWechat()
|
|
init_wechat = _copy(WECHAT_DEFAULTS)
|
|
init_wechat["subscriptions"] = []
|
|
curr_wechat = _copy(init_wechat)
|
|
curr_wechat["subscriptions"] = [
|
|
{
|
|
"id": "sub_task_1",
|
|
"membershipType": str(task.p.membershipType),
|
|
"price": float(task.p.price),
|
|
"billingCycle": str(task.p.billingCycle),
|
|
"autoRenew": False,
|
|
"createdAt": TEST_TS,
|
|
"source": "哔哩哔哩",
|
|
}
|
|
]
|
|
init_bilibili = _copy(BILIBILI_DEFAULTS)
|
|
curr_bilibili = _copy(init_bilibili)
|
|
curr_bilibili["user"]["isVip"] = True
|
|
curr_bilibili["user"]["vipExpireAt"] = TEST_TS + 30 * 24 * 60 * 60 * 1000
|
|
return task, _make_input(
|
|
apps_init={"wechat": init_wechat, "bilibili": init_bilibili},
|
|
apps_curr={"wechat": curr_wechat, "bilibili": curr_bilibili},
|
|
)
|
|
|
|
|
|
def _subscribe_membership_negative() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.SubscribeMembershipAutoRenewThenCancelInWechat()
|
|
init_wechat = _copy(WECHAT_DEFAULTS)
|
|
init_wechat["subscriptions"] = []
|
|
curr_wechat = _copy(init_wechat)
|
|
curr_wechat["subscriptions"] = [
|
|
{
|
|
"id": "sub_task_2",
|
|
"membershipType": str(task.p.membershipType),
|
|
"price": float(task.p.price),
|
|
"billingCycle": str(task.p.billingCycle),
|
|
"autoRenew": True,
|
|
"createdAt": TEST_TS,
|
|
"source": "哔哩哔哩",
|
|
}
|
|
]
|
|
init_bilibili = _copy(BILIBILI_DEFAULTS)
|
|
curr_bilibili = _copy(init_bilibili)
|
|
curr_bilibili["user"]["isVip"] = True
|
|
curr_bilibili["user"]["vipExpireAt"] = TEST_TS + 30 * 24 * 60 * 60 * 1000
|
|
return task, _make_input(
|
|
apps_init={"wechat": init_wechat, "bilibili": init_bilibili},
|
|
apps_curr={"wechat": curr_wechat, "bilibili": curr_bilibili},
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _transfer_notify_positive() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.AlipayTransferAndNotify()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
curr_alipay = _append_alipay_record(
|
|
init_alipay,
|
|
record_id="notify_tx_1",
|
|
kind="transfer",
|
|
delta=-float(task.p.amount),
|
|
method_id="balance",
|
|
counterparty=str(task.p.alipay_contact),
|
|
target_account=str(task.p.alipay_contact),
|
|
transfer_note=str(task.p.note),
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
init_wechat = _copy(WECHAT_DEFAULTS)
|
|
curr_wechat = _append_wechat_text(
|
|
init_wechat,
|
|
contact_name=str(task.p.contact),
|
|
content=f"我刚给你转账{float(task.p.amount):g}元，记得查收。",
|
|
outgoing=True,
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay, "wechat": init_wechat},
|
|
apps_curr={"alipay": curr_alipay, "wechat": curr_wechat},
|
|
)
|
|
|
|
|
|
def _transfer_notify_negative() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.AlipayTransferAndNotify()
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
curr_alipay = _append_alipay_record(
|
|
init_alipay,
|
|
record_id="notify_tx_neg_1",
|
|
kind="transfer",
|
|
delta=-float(task.p.amount),
|
|
method_id="balance",
|
|
counterparty=str(task.p.alipay_contact),
|
|
target_account=str(task.p.alipay_contact),
|
|
transfer_note=str(task.p.note),
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
init_wechat = _copy(WECHAT_DEFAULTS)
|
|
curr_wechat = _append_wechat_text(
|
|
init_wechat,
|
|
contact_name=str(task.p.contact),
|
|
content="晚点联系你。",
|
|
outgoing=True,
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"alipay": init_alipay, "wechat": init_wechat},
|
|
apps_curr={"alipay": curr_alipay, "wechat": curr_wechat},
|
|
)
|
|
|
|
|
|
def _wechat_extract_positive() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.WechatExtractAmountTransfer()
|
|
amount = float(task.p.requestAmount)
|
|
init_wechat = _copy(WECHAT_DEFAULTS)
|
|
curr_wechat = _append_wechat_text(
|
|
init_wechat,
|
|
contact_name=str(task.p.contact),
|
|
content=f"麻烦你转{amount:g}元给我。",
|
|
outgoing=False,
|
|
)
|
|
curr_wechat = _append_wechat_text(
|
|
curr_wechat,
|
|
contact_name=str(task.p.contact),
|
|
content=str(task.p.reply),
|
|
outgoing=True,
|
|
)
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
curr_alipay = _append_alipay_record(
|
|
init_alipay,
|
|
record_id="extract_tx_1",
|
|
kind="transfer",
|
|
delta=-amount,
|
|
method_id="balance",
|
|
counterparty=str(task.p.alipay_contact),
|
|
target_account=str(task.p.alipay_contact),
|
|
transfer_note="按微信请求转账",
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"wechat": init_wechat, "alipay": init_alipay},
|
|
apps_curr={"wechat": curr_wechat, "alipay": curr_alipay},
|
|
)
|
|
|
|
|
|
def _wechat_extract_negative() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.WechatExtractAmountTransfer()
|
|
amount = float(task.p.requestAmount)
|
|
init_wechat = _copy(WECHAT_DEFAULTS)
|
|
curr_wechat = _append_wechat_text(
|
|
init_wechat,
|
|
contact_name=str(task.p.contact),
|
|
content=f"麻烦你转{amount:g}元给我。",
|
|
outgoing=False,
|
|
)
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
curr_alipay = _append_alipay_record(
|
|
init_alipay,
|
|
record_id="extract_tx_neg_1",
|
|
kind="transfer",
|
|
delta=-(amount + 5.0),
|
|
method_id="balance",
|
|
counterparty=str(task.p.alipay_contact),
|
|
target_account=str(task.p.alipay_contact),
|
|
transfer_note="转错金额",
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"wechat": init_wechat, "alipay": init_alipay},
|
|
apps_curr={"wechat": curr_wechat, "alipay": curr_alipay},
|
|
)
|
|
|
|
|
|
def _wechat_extract_missing_reply() -> tuple[BaseTask, Any]:
|
|
task = payment_tasks.WechatExtractAmountTransfer()
|
|
amount = float(task.p.requestAmount)
|
|
init_wechat = _copy(WECHAT_DEFAULTS)
|
|
curr_wechat = _append_wechat_text(
|
|
init_wechat,
|
|
contact_name=str(task.p.contact),
|
|
content=f"麻烦你转{amount:g}元给我。",
|
|
outgoing=False,
|
|
)
|
|
init_alipay = _copy(ALIPAY_DEFAULTS)
|
|
curr_alipay = _append_alipay_record(
|
|
init_alipay,
|
|
record_id="extract_tx_no_reply_1",
|
|
kind="transfer",
|
|
delta=-amount,
|
|
method_id="balance",
|
|
counterparty=str(task.p.alipay_contact),
|
|
target_account=str(task.p.alipay_contact),
|
|
transfer_note="按微信请求转账",
|
|
timestamp=TEST_TS + 1000,
|
|
)
|
|
return task, _make_input(
|
|
apps_init={"wechat": init_wechat, "alipay": init_alipay},
|
|
apps_curr={"wechat": curr_wechat, "alipay": curr_alipay},
|
|
)
|
|
|
|
|
|
OFFLINE_JUDGE_TASK_NAMES = {
|
|
"AlipayContinuousPaymentsToContactsRecordBalances",
|
|
"AlipayBindMultipleCardsTransferAndRecordSuccessfulCards",
|
|
"AlipayChangePaymentPasswordThenPay",
|
|
"SubscribeMembershipAutoRenewThenCancelInWechat",
|
|
"TransferToContactWithNote",
|
|
"AlipayTransferAndNotify",
|
|
"WechatExtractAmountTransfer",
|
|
}
|
|
|
|
|
|
OFFLINE_POSITIVE_CASES: list[tuple[str, Callable[[], tuple[BaseTask, Any]]]] = [
|
|
("AlipayContinuousPaymentsToContactsRecordBalances", _continuous_payments_positive),
|
|
("AlipayBindMultipleCardsTransferAndRecordSuccessfulCards", _bind_multiple_cards_positive),
|
|
("AlipayChangePaymentPasswordThenPay", _change_password_positive),
|
|
("SubscribeMembershipAutoRenewThenCancelInWechat", _subscribe_membership_positive),
|
|
("TransferToContactWithNote", _transfer_with_note_positive),
|
|
("AlipayTransferAndNotify", _transfer_notify_positive),
|
|
("WechatExtractAmountTransfer", _wechat_extract_positive),
|
|
]
|
|
|
|
|
|
OFFLINE_NEGATIVE_CASES: list[tuple[str, Callable[[], tuple[BaseTask, Any]]]] = [
|
|
("AlipayContinuousPaymentsToContactsRecordBalances", _continuous_payments_negative),
|
|
("AlipayBindMultipleCardsTransferAndRecordSuccessfulCards", _bind_multiple_cards_negative),
|
|
("AlipayChangePaymentPasswordThenPay", _change_password_negative),
|
|
("SubscribeMembershipAutoRenewThenCancelInWechat", _subscribe_membership_negative),
|
|
("TransferToContactWithNote", _transfer_with_note_negative),
|
|
("AlipayTransferAndNotify", _transfer_notify_negative),
|
|
("WechatExtractAmountTransfer", _wechat_extract_negative),
|
|
]
|
|
|
|
|
|
class TestTaskJudgeMatrixOffline:
|
|
@pytest.mark.parametrize("name,factory", OFFLINE_POSITIVE_CASES)
|
|
def test_positive(self, name: str, factory: Callable[[], tuple[BaseTask, Any]]) -> None:
|
|
task, inp = factory()
|
|
result = task.evaluate(inp)
|
|
assert result.success, f"positive case {name} failed: {result.issues}"
|
|
|
|
@pytest.mark.parametrize("name,factory", OFFLINE_NEGATIVE_CASES)
|
|
def test_negative(self, name: str, factory: Callable[[], tuple[BaseTask, Any]]) -> None:
|
|
task, inp = factory()
|
|
result = task.evaluate(inp)
|
|
assert not result.success, f"negative case {name} unexpectedly passed"
|
|
|
|
def test_wechat_extract_requires_reply(self) -> None:
|
|
task, inp = _wechat_extract_missing_reply()
|
|
result = task.evaluate(inp)
|
|
assert not result.success
|
|
|
|
def test_matrix_complete(self) -> None:
|
|
pos_names = {name for name, _ in OFFLINE_POSITIVE_CASES}
|
|
neg_names = {name for name, _ in OFFLINE_NEGATIVE_CASES}
|
|
assert pos_names == OFFLINE_JUDGE_TASK_NAMES
|
|
assert neg_names == OFFLINE_JUDGE_TASK_NAMES
|