1be28f10e9
The patrol-delete flow now waits for country and listing-status controls before acting, uses the dedicated PatrolDeleteTask entry point, and includes the browser helper scripts required by the bulk-delete path. Tests cover delayed dropdown readiness, payload rejection/retry behavior, complete-draft normalization, filtered bulk deletion, and asset fallback lookup. Constraint: Amazon listing UI exposes status controls through dynamic KAT components and delayed option rendering Rejected: Keep row-by-row delete flow | bulk selection is the current implemented path and is covered by the added helper scripts Confidence: high Scope-risk: moderate Directive: Do not remove the patrol_delete JS helper files without checking app/amazon/patrol_delete.py script loading Tested: uv run --group dev pytest tests\\test_amazon_base.py tests\\test_patrol_delete.py Tested: uv run python -m py_compile amazon\\base.py amazon\\main.py blueprints\\main.py Not-tested: Real Amazon Seller Central browser session
386 lines
13 KiB
Python
386 lines
13 KiB
Python
import json
|
|
|
|
import pytest
|
|
from amazon import base
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, payload):
|
|
self.payload = payload
|
|
|
|
def json(self):
|
|
return self.payload
|
|
|
|
|
|
class FakeWait:
|
|
def __init__(self, *, raise_on_doc_loaded=False):
|
|
self.doc_loaded_calls = []
|
|
self.displayed_calls = []
|
|
self.raise_on_doc_loaded = raise_on_doc_loaded
|
|
|
|
def doc_loaded(self, **kwargs):
|
|
if self.raise_on_doc_loaded:
|
|
raise RuntimeError("doc load failed")
|
|
self.doc_loaded_calls.append(kwargs)
|
|
|
|
def displayed(self, **kwargs):
|
|
self.displayed_calls.append(kwargs)
|
|
return True
|
|
|
|
|
|
class FakeElement:
|
|
def __init__(self, *, text="", value=None):
|
|
self.text = text
|
|
self.value = value
|
|
self.clicks = 0
|
|
self.inputs = []
|
|
self.wait = FakeWait()
|
|
|
|
def click(self, *args, **kwargs):
|
|
self.clicks += 1
|
|
|
|
def input(self, value, clear=False):
|
|
self.inputs.append((value, clear))
|
|
|
|
|
|
class FakeTab:
|
|
def __init__(self, *, ele_returns=None, eles_returns=None, raise_on_doc_loaded=False):
|
|
self.ele_returns = {key: list(values) for key, values in (ele_returns or {}).items()}
|
|
self.eles_returns = {key: list(values) for key, values in (eles_returns or {}).items()}
|
|
self.wait = FakeWait(raise_on_doc_loaded=raise_on_doc_loaded)
|
|
self.refreshes = 0
|
|
|
|
def ele(self, locator, timeout=None):
|
|
values = self.ele_returns.get(locator)
|
|
if not values:
|
|
return None
|
|
return values.pop(0)
|
|
|
|
def eles(self, locator, timeout=None):
|
|
values = self.eles_returns.get(locator)
|
|
if not values:
|
|
return []
|
|
return values.pop(0)
|
|
|
|
def refresh(self):
|
|
self.refreshes += 1
|
|
|
|
|
|
def test_ziniao_driver_initializes_browser_state():
|
|
user_info = {
|
|
"company": "test-company",
|
|
"username": "test-user",
|
|
"password": "test-password",
|
|
}
|
|
|
|
driver = base.ZiniaoDriver(user_info, socket_port=19999)
|
|
|
|
assert driver.user_info is user_info
|
|
assert driver.socket_port == 19999
|
|
assert driver.client_path is None
|
|
assert driver.browser is None
|
|
assert driver.tab is None
|
|
assert driver.store_id is None
|
|
|
|
|
|
def test_amazon_base_inherits_ziniao_driver():
|
|
driver = base.AmamzonBase({})
|
|
|
|
assert isinstance(driver, base.ZiniaoDriver)
|
|
|
|
|
|
def test_kill_process_uses_ziniao_driver(monkeypatch):
|
|
calls = []
|
|
|
|
class FakeDriver:
|
|
def __init__(self, user_info):
|
|
calls.append(("init", user_info))
|
|
|
|
def kill_process(self, version):
|
|
calls.append(("kill_process", version))
|
|
|
|
monkeypatch.setattr(base, "ZiniaoDriver", FakeDriver)
|
|
|
|
base.kill_process("v6")
|
|
|
|
assert calls == [("init", {}), ("kill_process", "v6")]
|
|
|
|
|
|
def test_get_zinaio_exe_returns_none_when_winreg_unavailable(monkeypatch):
|
|
driver = base.ZiniaoDriver({})
|
|
monkeypatch.setattr(base, "winreg", None)
|
|
messages = []
|
|
sink_id = base.logger.add(messages.append, level="ERROR", format="{message}")
|
|
|
|
try:
|
|
result = driver.get_zinaio_exe()
|
|
finally:
|
|
base.logger.remove(sink_id)
|
|
|
|
assert result is None
|
|
assert any("winreg" in message for message in messages)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("store_info", "expected_key"),
|
|
[
|
|
("12345", "browserId"),
|
|
("oauth-token", "browserOauth"),
|
|
],
|
|
)
|
|
def test_open_store_payload_uses_id_or_oauth(monkeypatch, store_info, expected_key):
|
|
payloads = []
|
|
|
|
def fake_post(url, data, timeout):
|
|
payloads.append(json.loads(data.decode("utf-8")))
|
|
return FakeResponse({"statusCode": 0, "debuggingPort": 9222})
|
|
|
|
monkeypatch.setattr(base.requests, "post", fake_post)
|
|
|
|
result = base.ZiniaoDriver({"company": "c"}).open_store(store_info)
|
|
|
|
assert result["debuggingPort"] == 9222
|
|
assert payloads[0]["action"] == "startBrowser"
|
|
assert payloads[0][expected_key] == store_info
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("payload", "expected"),
|
|
[
|
|
({"statusCode": 0, "browserList": [{"browserName": "shop"}]}, [{"browserName": "shop"}]),
|
|
({"statusCode": -10003}, None),
|
|
({"statusCode": 500, "message": "failed"}, None),
|
|
],
|
|
)
|
|
def test_get_browser_list_status_handling(monkeypatch, payload, expected):
|
|
monkeypatch.setattr(base.requests, "post", lambda *args, **kwargs: FakeResponse(payload))
|
|
|
|
assert base.ZiniaoDriver({}).get_browser_list() == expected
|
|
|
|
|
|
def test_update_core_retries_none_then_returns_on_success(monkeypatch):
|
|
calls = []
|
|
responses = [FakeResponse(None), FakeResponse({"statusCode": 0})]
|
|
|
|
def fake_post(url, data, timeout):
|
|
calls.append(json.loads(data.decode("utf-8")))
|
|
return responses.pop(0)
|
|
|
|
monkeypatch.setattr(base.requests, "post", fake_post)
|
|
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
|
|
|
|
base.ZiniaoDriver({}).update_core()
|
|
|
|
assert [payload["action"] for payload in calls] == ["updateCore", "updateCore"]
|
|
|
|
|
|
def test_wait_until_client_ready_polls_until_available(monkeypatch):
|
|
readiness = [False, False, True]
|
|
sleeps = []
|
|
driver = base.ZiniaoDriver({})
|
|
monkeypatch.setattr(driver, "_is_client_ready", lambda: readiness.pop(0))
|
|
monkeypatch.setattr(base.time, "sleep", lambda seconds: sleeps.append(seconds))
|
|
|
|
assert driver._wait_until_client_ready(timeout=10) is True
|
|
assert sleeps == [base.CLIENT_READY_INTERVAL, base.CLIENT_READY_INTERVAL]
|
|
|
|
|
|
def test_close_store_success_uses_current_store(monkeypatch):
|
|
payloads = []
|
|
|
|
def fake_post(url, data, timeout):
|
|
payloads.append(json.loads(data.decode("utf-8")))
|
|
return FakeResponse({"statusCode": 0, "ok": True})
|
|
|
|
monkeypatch.setattr(base.requests, "post", fake_post)
|
|
driver = base.ZiniaoDriver({})
|
|
driver.store_id = "oauth-1"
|
|
|
|
result = driver.close_store()
|
|
|
|
assert result["ok"] is True
|
|
assert payloads[0]["action"] == "stopBrowser"
|
|
assert payloads[0]["browserOauth"] == "oauth-1"
|
|
|
|
|
|
def test_close_store_login_failure_raises(monkeypatch):
|
|
monkeypatch.setattr(base.requests, "post", lambda *args, **kwargs: FakeResponse({"statusCode": -10003}))
|
|
|
|
with pytest.raises(RuntimeError, match="【close_store】登录失败"):
|
|
base.ZiniaoDriver({}).close_store("oauth-1")
|
|
|
|
|
|
def test_open_shop_returns_message_when_shop_missing(monkeypatch):
|
|
driver = base.ZiniaoDriver({})
|
|
monkeypatch.setattr(driver, "start_client", lambda: None)
|
|
monkeypatch.setattr(driver, "get_browser_list", lambda: [{"browserName": "other"}])
|
|
|
|
assert driver.open_shop("target") == "店铺不存在"
|
|
assert driver.store_id is None
|
|
|
|
|
|
def test_open_shop_opens_matching_store(monkeypatch):
|
|
opened_pages = []
|
|
fake_browser = object()
|
|
driver = base.ZiniaoDriver({})
|
|
monkeypatch.setattr(driver, "start_client", lambda: None)
|
|
monkeypatch.setattr(driver, "get_browser_list", lambda: [{"browserName": "target", "browserOauth": "oauth-1"}])
|
|
monkeypatch.setattr(
|
|
driver,
|
|
"open_store",
|
|
lambda store_id: {
|
|
"browserOauth": "oauth-2",
|
|
"debuggingPort": 9222,
|
|
"ipDetectionPage": "https://ip-check",
|
|
"launcherPage": "https://launcher",
|
|
},
|
|
)
|
|
monkeypatch.setattr(driver, "get_browser", lambda port: fake_browser)
|
|
monkeypatch.setattr(driver, "open_ip_check", lambda browser, url: True)
|
|
monkeypatch.setattr(driver, "open_launcher_page", lambda page, browser: opened_pages.append((page, browser)))
|
|
|
|
assert driver.open_shop("target") is fake_browser
|
|
assert driver.store_id == "oauth-2"
|
|
assert opened_pages == [("https://launcher", fake_browser)]
|
|
|
|
|
|
def test_switching_countries_returns_true_when_already_current():
|
|
current_country = FakeElement(text="西班牙")
|
|
driver = base.AmamzonBase({})
|
|
driver.browser = object()
|
|
driver.tab = FakeTab(
|
|
ele_returns={
|
|
'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]': [current_country],
|
|
}
|
|
)
|
|
|
|
assert driver.switch_to_country("西班牙") is True
|
|
|
|
|
|
def test_switching_countries_returns_false_when_target_missing(monkeypatch):
|
|
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
|
|
driver = base.AmamzonBase({})
|
|
driver.browser = object()
|
|
driver.tab = FakeTab(
|
|
ele_returns={
|
|
'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]': [FakeElement(text="英国")],
|
|
'xpath://div[@class="dropdown-account-switcher-header-label"]': [FakeElement()],
|
|
'xpath://div[@class="dropdown-account-switcher-list-item"]': [FakeElement()],
|
|
'xpath://div[@class="dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" and @title="西班牙"]': [
|
|
None
|
|
],
|
|
}
|
|
)
|
|
|
|
assert driver.switch_to_country("西班牙") is False
|
|
|
|
|
|
def test_switching_countries_clicks_and_verifies_target(monkeypatch):
|
|
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
|
|
dropdown = FakeElement()
|
|
first_item = FakeElement()
|
|
target = FakeElement()
|
|
driver = base.AmamzonBase({})
|
|
driver.browser = object()
|
|
driver.tab = FakeTab(
|
|
ele_returns={
|
|
'xpath://div[@class="dropdown-account-switcher-header-label"]/span[last()]': [
|
|
FakeElement(text="英国"),
|
|
FakeElement(text="西班牙"),
|
|
],
|
|
'xpath://div[@class="dropdown-account-switcher-header-label"]': [dropdown],
|
|
'xpath://div[@class="dropdown-account-switcher-list-item"]': [first_item],
|
|
'xpath://div[@class="dropdown-account-switcher-list-item dropdown-account-switcher-list-item-indented" and @title="西班牙"]': [
|
|
target
|
|
],
|
|
}
|
|
)
|
|
|
|
assert driver.switch_to_country("西班牙") is True
|
|
assert dropdown.clicks == 1
|
|
assert first_item.clicks == 1
|
|
assert target.clicks == 1
|
|
|
|
|
|
def test_open_country_dropdown_waits_for_delayed_country_list(monkeypatch):
|
|
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
|
|
dropdown = FakeElement()
|
|
first_item = FakeElement()
|
|
driver = base.AmamzonBase({})
|
|
driver.tab = FakeTab(
|
|
ele_returns={
|
|
'xpath://div[@class="dropdown-account-switcher-header-label"]': [dropdown],
|
|
'xpath://div[@class="dropdown-account-switcher-list-item"]': [None, first_item],
|
|
}
|
|
)
|
|
|
|
assert driver._open_country_dropdown() is True
|
|
assert dropdown.clicks == 2
|
|
assert first_item.clicks == 1
|
|
|
|
|
|
def test_login_returns_true_when_otp_submit_succeeds(monkeypatch):
|
|
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
|
|
password_input = FakeElement()
|
|
password_submit = FakeElement()
|
|
otp_input = FakeElement(value="123456")
|
|
otp_submit = FakeElement()
|
|
driver = base.AmamzonBase({})
|
|
driver.tab = FakeTab(
|
|
eles_returns={
|
|
'xpath://input[@type="password"]': [[password_input]],
|
|
'xpath://input[@id="signInSubmit"]': [[password_submit]],
|
|
'xpath://span[@id="auth-send-code" and contains(string(.),"发送一次性密码")]': [[]],
|
|
'xpath://input[@name="otpCode"]': [[otp_input]],
|
|
'xpath://input[@id="auth-signin-button"]': [[otp_submit]],
|
|
'xpath://div[@id="auth-error-message-box"]': [[]],
|
|
}
|
|
)
|
|
|
|
assert driver.login("secret") is True
|
|
assert password_input.inputs == [("secret", True)]
|
|
assert password_submit.clicks == 1
|
|
assert otp_submit.clicks == 1
|
|
|
|
|
|
def test_login_refreshes_on_otp_error(monkeypatch):
|
|
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
|
|
otp_input = FakeElement(value="123456")
|
|
driver = base.AmamzonBase({})
|
|
driver.tab = FakeTab(
|
|
eles_returns={
|
|
'xpath://input[@type="password"]': [[]],
|
|
'xpath://input[@id="signInSubmit"]': [[]],
|
|
'xpath://span[@id="auth-send-code" and contains(string(.),"发送一次性密码")]': [[]],
|
|
'xpath://input[@name="otpCode"]': [[otp_input]],
|
|
'xpath://input[@id="auth-signin-button"]': [[FakeElement()]],
|
|
'xpath://div[@id="auth-error-message-box"]': [[FakeElement(text="验证码错误")]],
|
|
}
|
|
)
|
|
|
|
assert driver.login("secret") is False
|
|
assert driver.tab.refreshes == 1
|
|
|
|
|
|
def test_login_returns_false_when_no_otp_input(monkeypatch):
|
|
monkeypatch.setattr(base.time, "sleep", lambda seconds: None)
|
|
driver = base.AmamzonBase({})
|
|
driver.tab = FakeTab(
|
|
eles_returns={
|
|
'xpath://input[@type="password"]': [[]],
|
|
'xpath://input[@id="signInSubmit"]': [[]],
|
|
'xpath://span[@id="auth-send-code" and contains(string(.),"发送一次性密码")]': [[]],
|
|
'xpath://input[@name="otpCode"]': [[]],
|
|
'xpath://input[@id="auth-signin-button"]': [[]],
|
|
}
|
|
)
|
|
|
|
assert driver.login("secret") is False
|
|
|
|
|
|
def test_login_returns_false_on_exception():
|
|
driver = base.AmamzonBase({})
|
|
driver.tab = FakeTab(raise_on_doc_loaded=True)
|
|
|
|
assert driver.login("secret") is False
|