chengma/cmbuyer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(client): verify T-105 quantity Gate2

Browse Source
This commit is contained in:
QiuSW
2026-08-06 09:28:00 +08:00
parent 1d726dd364
commit 21628a4dc5
6 changed files with 1579 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/tests/pdd/fixtures/quantity_gate2_initial_8_17_0.xml
+30
View File
@@ -0,0 +1,30 @@
<hierarchy rotation="0">
<!-- T-105 只保留人工确认数量 1 证据中的规格、顶部金额与数量控件。 -->
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[0,474][1080,863]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node package="com.xunmeng.pinduoduo" class="android.widget.FrameLayout" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,498][1053,570]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,498][895,570]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node text="快卖完 ¥12.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,503][712,570]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
</node>
</node>
<node text="已选： 黑色 CHA (纯棉） M（建议100-115）" package="com.xunmeng.pinduoduo" class="android.widget.TextView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,654][1053,716]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" resource-id="com.xunmeng.pinduoduo:id/gnl" bounds="[396,752][645,827]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" bounds="[396,752][645,827]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node content-desc="减少数量" package="com.xunmeng.pinduoduo" class="android.widget.ImageView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,752][474,827]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
<node text="1" package="com.xunmeng.pinduoduo" class="android.widget.EditText" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[480,752][561,827]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
<node content-desc="增加数量" package="com.xunmeng.pinduoduo" class="android.widget.ImageView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[567,752][645,827]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
</node>
</node>
</node>
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[126,1000][438,1024]" clickable="true" enabled="true" visible-to-user="true" selected="true" scrollable="false">
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[126,1000][438,1024]" clickable="false" enabled="true" visible-to-user="true" selected="true" scrollable="false">
<node text="黑色 CHA (纯棉）" package="com.xunmeng.pinduoduo" class="android.widget.TextView" resource-id="com.xunmeng.pinduoduo:id/tv_content" bounds="[126,1000][438,1024]" clickable="true" enabled="true" visible-to-user="true" selected="true" scrollable="false" />
</node>
</node>
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[439,1582][831,1667]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node text="M（建议100-115）" package="com.xunmeng.pinduoduo" class="android.widget.TextView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[439,1582][831,1667]" clickable="true" enabled="true" visible-to-user="true" selected="true" scrollable="false" />
</node>
<!-- 底部金额故意保留为反例；生产读取器只认上方证据角色。 -->
<node package="com.xunmeng.pinduoduo" class="android.widget.FrameLayout" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[0,2181][1080,2328]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node text="提交订单 ¥12.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" bounds="[369,2225][710,2284]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
</node>
</hierarchy>
client/tests/pdd/fixtures/quantity_gate2_target_8_17_0.xml
+30
View File
@@ -0,0 +1,30 @@
<hierarchy rotation="0">
<!-- T-105 只保留人工确认数量 2 证据中的规格、顶部金额与数量控件。 -->
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[0,474][1080,861]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node package="com.xunmeng.pinduoduo" class="android.widget.FrameLayout" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,498][1053,570]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,498][740,570]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node text="快卖完 ¥32.76" package="com.xunmeng.pinduoduo" class="android.widget.TextView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,503][722,570]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
</node>
</node>
<node text="已选： 黑色 CHA (纯棉） M（建议100-115）" package="com.xunmeng.pinduoduo" class="android.widget.TextView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,582][1053,644]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" resource-id="com.xunmeng.pinduoduo:id/gnl" bounds="[396,750][645,825]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" bounds="[396,750][645,825]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node content-desc="减少数量" package="com.xunmeng.pinduoduo" class="android.widget.ImageView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[396,750][474,825]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
<node text="2" package="com.xunmeng.pinduoduo" class="android.widget.EditText" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[480,750][561,825]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
<node content-desc="增加数量" package="com.xunmeng.pinduoduo" class="android.widget.ImageView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[567,750][645,825]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
</node>
</node>
</node>
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[126,998][438,1024]" clickable="true" enabled="true" visible-to-user="true" selected="true" scrollable="false">
<node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[126,998][438,1024]" clickable="false" enabled="true" visible-to-user="true" selected="true" scrollable="false">
<node text="黑色 CHA (纯棉）" package="com.xunmeng.pinduoduo" class="android.widget.TextView" resource-id="com.xunmeng.pinduoduo:id/tv_content" bounds="[126,998][438,1024]" clickable="true" enabled="true" visible-to-user="true" selected="true" scrollable="false" />
</node>
</node>
<node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[439,1582][831,1667]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node text="M（建议100-115）" package="com.xunmeng.pinduoduo" class="android.widget.TextView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[439,1582][831,1667]" clickable="true" enabled="true" visible-to-user="true" selected="true" scrollable="false" />
</node>
<!-- 与 Gate2 值相同也不能成为读取候选。 -->
<node package="com.xunmeng.pinduoduo" class="android.widget.FrameLayout" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[0,2181][1080,2328]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false">
<node text="提交订单 ¥32.76" package="com.xunmeng.pinduoduo" class="android.widget.TextView" bounds="[366,2225][714,2284]" clickable="false" enabled="true" visible-to-user="true" selected="false" scrollable="false" />
</node>
</hierarchy>
client/tests/pdd/test_quantity_gate2.py
+385
View File
@@ -6,6 +6,7 @@ import argparse
import ast
import base64
from contextlib import redirect_stderr
from datetime import UTC, datetime
from functools import lru_cache
from importlib.util import module_from_spec, spec_from_file_location
from io import BytesIO, StringIO
@@ -33,6 +34,14 @@ from cmbuyer_client.pdd.quantity_gate2_spike import (
QuantityGate2ForegroundReader,
QuantityGate2ReadDevice,
)
from cmbuyer_client.pdd.quantity_gate2 import (
Gate1Observation,
QuantityGate2Device,
QuantityGate2Error,
QuantityGate2Flow,
QuantityGate2OverCapError,
)
from cmbuyer_client.pdd.quantity_gate2_runner import QuantityGate2Runner
SERIAL = "192.168.0.173:5555"
@@ -144,6 +153,15 @@ def _load_script() -> object:
return module
def _load_run_script() -> object:
script_path = CLIENT_ROOT / "scripts" / "run_t105_quantity_gate2.py"
spec = spec_from_file_location("run_t105_quantity_gate2_for_test", script_path)
assert spec is not None and spec.loader is not None
module = module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _namespace(**changes: object) -> argparse.Namespace:
values: dict[str, object] = {
"serial": SERIAL,
@@ -555,5 +573,372 @@ class QuantityGate2CliAndStaticBoundaryTests(unittest.TestCase):
self.assertNotIn(forbidden_import, source.read_text(encoding="utf-8"))
_INITIAL_FIXTURE = CLIENT_ROOT / "tests" / "pdd" / "fixtures" / "quantity_gate2_initial_8_17_0.xml"
_TARGET_FIXTURE = CLIENT_ROOT / "tests" / "pdd" / "fixtures" / "quantity_gate2_target_8_17_0.xml"
_EXIT_FIXTURE = CLIENT_ROOT / "tests" / "pdd" / "fixtures" / "product_exit_8_17_0.xml"
def _gate1(path: Path = Path("gate1.png")) -> Gate1Observation:
return Gate1Observation(
color="黑色CHA(纯棉)",
size="M(建议100-115)",
quantity=1,
gate1_unit_price="12.88",
screenshot_path=path,
captured_at=datetime(2026, 8, 6, 0, 53, tzinfo=UTC),
)
class FakeQuantityFlowDevice:
def __init__(self, initial: str | None = None, target: str | None = None) -> None:
self.initial = initial or _INITIAL_FIXTURE.read_text(encoding="utf-8")
self.target = target or _TARGET_FIXTURE.read_text(encoding="utf-8")
self.exit = _EXIT_FIXTURE.read_text(encoding="utf-8")
self.current = self.initial
self.version = "8.17.0"
self.foreground = {"package": "com.xunmeng.pinduoduo", "activity": ".activity.NewPageActivity"}
self.screen_size = (1080, 2376)
self.increment_calls: list[str] = []
self.back_calls = 0
self.dump_calls = 0
self.raise_increment: BaseException | None = None
def app_info(self, package_name: str) -> dict[str, str]:
return {"versionName": self.version}
def current_foreground(self) -> dict[str, str]:
return dict(self.foreground)
def display_size(self) -> tuple[int, int]:
return self.screen_size
def dump_window_hierarchy(self) -> str:
self.dump_calls += 1
return self.current
def increment_quantity_once(self, bounds: str) -> None:
self.increment_calls.append(bounds)
if self.raise_increment is not None:
raise self.raise_increment
self.current = self.target
def capture_screenshot(self) -> str:
return _png_base64()
def leave_sku_panel_once(self) -> None:
self.back_calls += 1
self.current = self.exit
class QuantityGate2FlowTests(unittest.TestCase):
def test_quantity_two_reads_nonlinear_panel_total_and_exits_once(self) -> None:
device = FakeQuantityFlowDevice()
flow = QuantityGate2Flow(device, wait_timeout_seconds=0.1, poll_interval_seconds=0.01)
verified = flow.set_quantity_and_verify(_gate1(), 2, "40.00")
observation = flow.build_observation(
_gate1(),
2,
"40.00",
Path("gate2.png"),
datetime(2026, 8, 6, 1, 10, tzinfo=UTC),
)
flow.exit_sku_panel_safely()
self.assertEqual(verified.quantity, 2)
self.assertEqual(observation.quantity_read, 2)
self.assertEqual(observation.gate1_unit_price, "12.88")
self.assertEqual(observation.gate2_panel_total_price, "32.76")
self.assertEqual(observation.max_total_price, "40.00")
self.assertEqual(device.increment_calls, ["[567,752][645,827]"])
self.assertEqual(device.back_calls, 1)
self.assertTrue(flow.exited)
def test_quantity_one_is_zero_click_and_still_uses_panel_amount(self) -> None:
device = FakeQuantityFlowDevice()
flow = QuantityGate2Flow(device)
verified = flow.set_quantity_and_verify(_gate1(), 1, "12.88")
self.assertEqual((verified.quantity, verified.panel_total_price), (1, "12.88"))
self.assertEqual(device.increment_calls, [])
def test_only_evidenced_quantities_and_canonical_decimal_cap_are_allowed(self) -> None:
for quantity in (0, 3, -1, True, "2"):
with self.subTest(quantity=quantity):
device = FakeQuantityFlowDevice()
with self.assertRaises(QuantityGate2Error):
QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), quantity, "40.00") # type: ignore[arg-type]
self.assertEqual(device.increment_calls, [])
for cap in ("0.00", "40", "040.00", 40.0, "NaN"):
with self.subTest(cap=cap):
device = FakeQuantityFlowDevice()
with self.assertRaises(QuantityGate2Error):
QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), 2, cap) # type: ignore[arg-type]
self.assertEqual(device.increment_calls, [])
def test_over_cap_stops_after_exact_readback_without_fabricated_math(self) -> None:
device = FakeQuantityFlowDevice()
flow = QuantityGate2Flow(device)
with self.assertRaises(QuantityGate2OverCapError):
flow.set_quantity_and_verify(_gate1(), 2, "30.00")
self.assertEqual(device.increment_calls, ["[567,752][645,827]"])
self.assertTrue(flow.can_exit_safely)
def test_increment_timeout_is_sealed_and_cannot_be_retried(self) -> None:
device = FakeQuantityFlowDevice()
device.raise_increment = TimeoutError("ambiguous delivery")
flow = QuantityGate2Flow(device, wait_timeout_seconds=0.01, poll_interval_seconds=0.005)
with self.assertRaises(TimeoutError):
flow.set_quantity_and_verify(_gate1(), 2, "40.00")
with self.assertRaises(QuantityGate2Error):
flow.set_quantity_and_verify(_gate1(), 2, "40.00")
self.assertEqual(len(device.increment_calls), 1)
self.assertFalse(flow.can_exit_safely)
def test_missing_duplicate_drift_and_overlay_fail_before_click(self) -> None:
initial = _INITIAL_FIXTURE.read_text(encoding="utf-8")
plus = '<node content-desc="增加数量" package="com.xunmeng.pinduoduo" class="android.widget.ImageView" resource-id="com.xunmeng.pinduoduo:id/pdd" bounds="[567,752][645,827]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false" />'
variants = (
initial.replace('content-desc="增加数量"', 'content-desc="数量加一"', 1),
initial.replace(plus, plus + plus),
initial.replace('text="M（建议100-115）"', 'text="S（建议80-100）"', 1),
initial.replace(
"</hierarchy>",
'<node package="com.example.overlay" class="android.view.ViewGroup" bounds="[560,740][660,840]" clickable="true" enabled="true" visible-to-user="true" selected="false" scrollable="false" /></hierarchy>',
),
)
for hierarchy in variants:
with self.subTest(marker=hierarchy[-180:]):
device = FakeQuantityFlowDevice(initial=hierarchy)
with self.assertRaises(QuantityGate2Error):
QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), 2, "40.00")
self.assertEqual(device.increment_calls, [])
def test_foreground_version_and_screen_drift_are_zero_click(self) -> None:
devices = (FakeQuantityFlowDevice(), FakeQuantityFlowDevice(), FakeQuantityFlowDevice())
devices[0].foreground = {"package": "com.android.systemui", "activity": ".Keyguard"}
devices[1].version = "8.17.1"
devices[2].screen_size = (1080, 2400)
for device in devices:
with self.subTest(device=device):
with self.assertRaises(QuantityGate2Error):
QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), 2, "40.00")
self.assertEqual(device.increment_calls, [])
def test_bottom_submit_amount_is_never_a_gate2_candidate(self) -> None:
target = _TARGET_FIXTURE.read_text(encoding="utf-8").replace("提交订单 ¥32.76", "提交订单 ¥999.99")
device = FakeQuantityFlowDevice(target=target)
result = QuantityGate2Flow(device).set_quantity_and_verify(_gate1(), 2, "40.00")
self.assertEqual(result.panel_total_price, "32.76")
def test_gate1_target_spec_and_timestamp_are_strict(self) -> None:
invalid = (
{"color": "白色"},
{"size": "L"},
{"quantity": 2},
{"gate1_unit_price": "12.89"},
{"captured_at": datetime(2026, 8, 6, 0, 53)},
)
base = _gate1().__dict__
for change in invalid:
with self.subTest(change=change), self.assertRaises(QuantityGate2Error):
Gate1Observation(**{**base, **change})
class FakeRawQuantityDevice:
def __init__(self, *, click_timeout: bool = False) -> None:
self.current = _INITIAL_FIXTURE.read_text(encoding="utf-8")
self.target = _TARGET_FIXTURE.read_text(encoding="utf-8")
self.exit = _EXIT_FIXTURE.read_text(encoding="utf-8")
self.calls: list[tuple[object, ...]] = []
self.click_timeout = click_timeout
def app_info(self, package: str) -> dict[str, str]:
self.calls.append(("app_info", package))
return {"versionName": "8.17.0"}
def window_size(self) -> tuple[int, int]:
self.calls.append(("window_size",))
return (1080, 2376)
def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> object:
self.calls.append(("jsonrpc", method, params, timeout))
if method == "dumpWindowHierarchy":
return self.current
if method == "click":
self.current = self.target
if self.click_timeout:
raise TimeoutError("may have been delivered")
return None
if method == "takeScreenshot":
return _png_base64()
if method == "pressKey":
self.current = self.exit
return None
raise AssertionError(method)
class QuantityGate2RunnerTests(unittest.TestCase):
def test_runner_publishes_gate2_evidence_then_one_safe_exit(self) -> None:
with TemporaryDirectory() as temporary:
base = Path(temporary)
gate1_path = base / "gate1.png"
gate1_path.write_bytes(base64.b64decode(_png_base64()))
target = base / "gate2"
raw = FakeRawQuantityDevice()
result = QuantityGate2Runner(
FakeAdbClient(),
lambda serial: raw,
FakeForegroundReader(),
timeout_seconds=0.1,
).run(SERIAL, EXPECTED_GOODS_ID, _gate1(gate1_path), 2, "40.00", target)
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
methods = [call[1] for call in raw.calls if call[0] == "jsonrpc"]
self.assertEqual(methods.count("click"), 1)
self.assertEqual(methods.count("pressKey"), 1)
self.assertEqual(methods.count("takeScreenshot"), 1)
self.assertEqual(result.observation.gate2_panel_total_price, "32.76")
self.assertEqual(manifest["prices"]["gate1_unit_price"], "12.88")
self.assertEqual(manifest["prices"]["gate2_panel_total_price"], "32.76")
self.assertEqual(manifest["prices"]["max_total_price"], "40.00")
self.assertEqual(manifest["action_audit"]["increment_attempts"], 1)
self.assertEqual(manifest["action_audit"]["back_attempts"], 1)
self.assertEqual(manifest["safe_exit"], "completed")
self.assertEqual(manifest["review_status"], "human_review_required")
serialized = result.manifest_path.read_text(encoding="utf-8")
self.assertNotIn(SERIAL, serialized)
self.assertNotIn(str(gate1_path), serialized)
def test_over_cap_attempts_one_safe_exit_and_publishes_nothing(self) -> None:
with TemporaryDirectory() as temporary:
base = Path(temporary)
gate1_path = base / "gate1.png"
gate1_path.write_bytes(base64.b64decode(_png_base64()))
raw = FakeRawQuantityDevice()
target = base / "gate2"
with self.assertRaises(QuantityGate2OverCapError):
QuantityGate2Runner(
FakeAdbClient(), lambda serial: raw, FakeForegroundReader(), timeout_seconds=0.1
).run(SERIAL, EXPECTED_GOODS_ID, _gate1(gate1_path), 2, "30.00", target)
methods = [call[1] for call in raw.calls if call[0] == "jsonrpc"]
self.assertEqual(methods.count("click"), 1)
self.assertEqual(methods.count("pressKey"), 1)
self.assertFalse(target.exists())
self.assertEqual(list(base.glob(".gate2.staging-*")), [])
def test_ambiguous_increment_is_never_retried_or_followed_by_back(self) -> None:
with TemporaryDirectory() as temporary:
base = Path(temporary)
gate1_path = base / "gate1.png"
gate1_path.write_bytes(base64.b64decode(_png_base64()))
raw = FakeRawQuantityDevice(click_timeout=True)
with self.assertRaises(QuantityGate2Error):
QuantityGate2Runner(
FakeAdbClient(), lambda serial: raw, FakeForegroundReader(), timeout_seconds=0.1
).run(SERIAL, EXPECTED_GOODS_ID, _gate1(gate1_path), 2, "40.00", base / "gate2")
methods = [call[1] for call in raw.calls if call[0] == "jsonrpc"]
self.assertEqual(methods.count("click"), 1)
self.assertEqual(methods.count("pressKey"), 0)
class QuantityGate2ProductionStaticBoundaryTests(unittest.TestCase):
def test_protocol_has_only_named_t105_actions(self) -> None:
public = {name for name in QuantityGate2Device.__dict__ if not name.startswith("_")}
self.assertEqual(
public,
{
"app_info",
"current_foreground",
"display_size",
"dump_window_hierarchy",
"increment_quantity_once",
"capture_screenshot",
"leave_sku_panel_once",
},
)
def test_production_sources_have_no_old_gate2_field_or_forbidden_capability(self) -> None:
paths = (
CLIENT_ROOT / "src/cmbuyer_client/pdd/quantity_gate2.py",
CLIENT_ROOT / "src/cmbuyer_client/pdd/quantity_gate2_runner.py",
CLIENT_ROOT / "scripts/run_t105_quantity_gate2.py",
)
forbidden_attributes = {
"go_to_order_confirm",
"submit_order_once",
"start_pdd_view_intent",
"pay",
"swipe",
"set_text",
"send_keys",
}
for path in paths:
source = path.read_text(encoding="utf-8")
tree = ast.parse(source)
self.assertNotIn("gate2_unit_price", source)
self.assertFalse(
{
node.attr
for node in ast.walk(tree)
if isinstance(node, ast.Attribute) and node.attr in forbidden_attributes
}
)
quantity_tree = ast.parse(paths[0].read_text(encoding="utf-8"))
self.assertFalse(any(isinstance(node, ast.Mult) for node in ast.walk(quantity_tree)))
def test_t103_and_t104_still_do_not_import_t105(self) -> None:
for relative in (
"src/cmbuyer_client/pdd/sku_selection.py",
"src/cmbuyer_client/pdd/sku_selection_runner.py",
):
self.assertNotIn("quantity_gate2", (CLIENT_ROOT / relative).read_text(encoding="utf-8"))
def test_live_cli_fixes_the_human_review_case_and_redacts_runtime_failure(self) -> None:
script = _load_run_script()
with TemporaryDirectory() as temporary:
screenshot = Path(temporary) / "gate1.png"
screenshot.write_bytes(base64.b64decode(_png_base64()))
arguments = script.parse_arguments( # type: ignore[attr-defined]
[
"--serial", SERIAL,
"--goods-id", EXPECTED_GOODS_ID,
"--color", "黑色CHA(纯棉)",
"--size", "M(建议100-115)",
"--target-quantity", "2",
"--gate1-unit-price", "12.88",
"--gate1-screenshot", str(screenshot),
"--gate1-captured-at", "2026-08-06T00:53:00.023935+00:00",
"--max-total-price", "40.00",
"--output-dir", str(Path(temporary) / "gate2"),
]
)
script.validate_arguments(arguments) # type: ignore[attr-defined]
stderr = StringIO()
fake_runner = unittest.mock.Mock()
fake_runner.run.side_effect = QuantityGate2Error(f"secret {SERIAL} {temporary}")
with (
patch.object(script, "QuantityGate2Runner", return_value=fake_runner),
patch.dict(sys.modules, {"adbutils": unittest.mock.Mock(), "uiautomator2": unittest.mock.Mock()}),
redirect_stderr(stderr),
):
result = script.main( # type: ignore[attr-defined]
[
"--serial", SERIAL,
"--goods-id", EXPECTED_GOODS_ID,
"--color", "黑色CHA(纯棉)",
"--size", "M(建议100-115)",
"--target-quantity", "2",
"--gate1-unit-price", "12.88",
"--gate1-screenshot", str(screenshot),
"--gate1-captured-at", "2026-08-06T00:53:00.023935+00:00",
"--max-total-price", "40.00",
"--output-dir", str(Path(temporary) / "gate2"),
]
)
self.assertEqual(result, 1)
self.assertNotIn(SERIAL, stderr.getvalue())
self.assertNotIn(temporary, stderr.getvalue())
if __name__ == "__main__":
unittest.main()