609 lines
19 KiB
Python
609 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import replace
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
import gyxx_flow.accounts as accounts
|
|
from gyxx_flow.adapters.browser import BrowserCookieStore
|
|
from gyxx_flow.adapters.integration import AccountBinding
|
|
from gyxx_flow.catalog import WorkflowCatalog
|
|
from gyxx_flow.core.config import Settings
|
|
from gyxx_flow.scheduler_service import PythonScheduler
|
|
|
|
|
|
def _account(tmp_path: Path) -> AccountBinding:
|
|
root = tmp_path / "data" / "state" / "accounts" / "demo-shop"
|
|
return AccountBinding(
|
|
account_id="demo-shop",
|
|
cdp_port=22200,
|
|
cdp_url="http://127.0.0.1:22200",
|
|
profile_dir=root / "profile",
|
|
cookie_file=root / "cookies.json",
|
|
storage_state_file=root / "storage_state.json",
|
|
required_cookie_domains=("example.test",),
|
|
required_cookie_names=("session",),
|
|
keepalive_enabled=True,
|
|
keepalive_url="https://example.test/console/home",
|
|
keepalive_interval_seconds=600,
|
|
)
|
|
|
|
|
|
class _FakeContext:
|
|
def __init__(self, cookies: list[dict[str, object]]) -> None:
|
|
self._cookies = cookies
|
|
|
|
def cookies(self) -> list[dict[str, object]]:
|
|
return list(self._cookies)
|
|
|
|
def storage_state(self) -> dict[str, object]:
|
|
return {"cookies": self.cookies(), "origins": []}
|
|
|
|
|
|
class _FakeBrowser:
|
|
final_url = "https://example.test/console/home"
|
|
cookies = [
|
|
{
|
|
"name": "session",
|
|
"value": "fresh-session",
|
|
"domain": "example.test",
|
|
"path": "/",
|
|
"expires": 4_102_444_800,
|
|
}
|
|
]
|
|
instances: list[_FakeBrowser] = []
|
|
|
|
def __init__(self, **options: object) -> None:
|
|
self.options = options
|
|
self.context = _FakeContext(self.cookies)
|
|
self.persisted = False
|
|
self.closed = False
|
|
self.instances.append(self)
|
|
|
|
def start(self) -> _FakeBrowser:
|
|
return self
|
|
|
|
def run_fetch_action(self, url: str, action, **_options: object):
|
|
page = SimpleNamespace(url=self.final_url)
|
|
assert url == "https://example.test/console/home"
|
|
return SimpleNamespace(status=200), action(page)
|
|
|
|
def persist_bound_state(self) -> None:
|
|
store = BrowserCookieStore(
|
|
self.options["cookie_file"], # type: ignore[arg-type]
|
|
self.options["storage_state_file"], # type: ignore[arg-type]
|
|
)
|
|
store.save_cookies(self.context.cookies())
|
|
store.save_storage_state(self.context.storage_state())
|
|
self.persisted = True
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class _FakeCatalog:
|
|
def __init__(self, account: AccountBinding) -> None:
|
|
self.accounts = (account,)
|
|
self.sync_calls = 0
|
|
|
|
def account_for(self, account_id: str) -> AccountBinding:
|
|
assert account_id == self.accounts[0].account_id
|
|
return self.accounts[0]
|
|
|
|
def sync_account_state(self, account_id: str) -> list[dict[str, object]]:
|
|
assert account_id == self.accounts[0].account_id
|
|
self.sync_calls += 1
|
|
return [{"script_id": "demo:first.py", "synced": True}]
|
|
|
|
|
|
def _write_old_state(account: AccountBinding) -> None:
|
|
old_cookie = {
|
|
"name": "session",
|
|
"value": "old-session",
|
|
"domain": "example.test",
|
|
"path": "/",
|
|
"expires": 4_102_444_800,
|
|
}
|
|
store = BrowserCookieStore(account.cookie_file, account.storage_state_file)
|
|
store.save_cookies([old_cookie])
|
|
store.save_storage_state({"cookies": [old_cookie], "origins": []})
|
|
|
|
|
|
def test_keepalive_publishes_only_a_verified_authenticated_snapshot(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
account = _account(tmp_path)
|
|
_write_old_state(account)
|
|
account.cookie_file.with_name("keepalive_status.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"status": "expired",
|
|
"identity_mismatch": True,
|
|
"final_url_login": False,
|
|
"error_type": "TargetClosedError",
|
|
"http_status": 502,
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
catalog = _FakeCatalog(account)
|
|
_FakeBrowser.instances = []
|
|
monkeypatch.setattr(accounts, "ScraplingBrowser", _FakeBrowser)
|
|
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=catalog, # type: ignore[arg-type]
|
|
)
|
|
|
|
assert manager.refresh_account("demo-shop") == {
|
|
"status": "success",
|
|
"cookie_count": 1,
|
|
"synced_members": 1,
|
|
}
|
|
assert json.loads(account.cookie_file.read_text(encoding="utf-8"))[0]["value"] == (
|
|
"fresh-session"
|
|
)
|
|
assert catalog.sync_calls == 1
|
|
assert _FakeBrowser.instances[0].persisted is True
|
|
assert _FakeBrowser.instances[0].closed is True
|
|
|
|
status = json.loads(
|
|
account.cookie_file.with_name("keepalive_status.json").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
)
|
|
assert status["status"] == "success"
|
|
assert "identity_mismatch" not in status
|
|
assert "final_url_login" not in status
|
|
assert "error_type" not in status
|
|
assert "http_status" not in status
|
|
assert "fresh-session" not in json.dumps(status)
|
|
|
|
|
|
def test_keepalive_does_not_overwrite_vault_after_login_redirect(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
account = _account(tmp_path)
|
|
_write_old_state(account)
|
|
catalog = _FakeCatalog(account)
|
|
_FakeBrowser.instances = []
|
|
_FakeBrowser.final_url = "https://example.test/passport/login"
|
|
monkeypatch.setattr(accounts, "ScraplingBrowser", _FakeBrowser)
|
|
|
|
try:
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=catalog, # type: ignore[arg-type]
|
|
)
|
|
assert manager.refresh_account("demo-shop")["status"] == "expired"
|
|
assert json.loads(account.cookie_file.read_text(encoding="utf-8"))[0]["value"] == (
|
|
"old-session"
|
|
)
|
|
assert catalog.sync_calls == 0
|
|
assert _FakeBrowser.instances[0].persisted is False
|
|
finally:
|
|
_FakeBrowser.final_url = "https://example.test/console/home"
|
|
|
|
|
|
def test_login_url_is_not_an_authenticated_identity_without_markers(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
account = _account(tmp_path)
|
|
|
|
assert not accounts._page_matches_account_identity(
|
|
SimpleNamespace(url="https://example.test/passport/login"), account
|
|
)
|
|
assert accounts._looks_like_login_url(
|
|
"https://loginmyseller.taobao.com/?redirect_url=https%3A%2F%2Fqn.taobao.com"
|
|
)
|
|
|
|
|
|
def test_probe_account_page_waits_for_client_redirects(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
account = _account(tmp_path)
|
|
waits: list[int] = []
|
|
page = SimpleNamespace(
|
|
url="https://example.test/console/home",
|
|
wait_for_timeout=waits.append,
|
|
)
|
|
|
|
state = accounts._probe_account_page(page, account)
|
|
|
|
assert state == {
|
|
"url": "https://example.test/console/home",
|
|
"identity_matches": True,
|
|
}
|
|
assert waits == [3_000]
|
|
|
|
|
|
def test_keepalive_rejects_a_valid_cookie_for_the_wrong_shop(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
base = _account(tmp_path)
|
|
account = AccountBinding(
|
|
account_id=base.account_id,
|
|
cdp_port=base.cdp_port,
|
|
cdp_url=base.cdp_url,
|
|
profile_dir=base.profile_dir,
|
|
cookie_file=base.cookie_file,
|
|
storage_state_file=base.storage_state_file,
|
|
required_cookie_domains=base.required_cookie_domains,
|
|
required_cookie_names=base.required_cookie_names,
|
|
login_identity_markers=("demo shop",),
|
|
keepalive_enabled=True,
|
|
keepalive_url=base.keepalive_url,
|
|
keepalive_interval_seconds=600,
|
|
)
|
|
_write_old_state(account)
|
|
catalog = _FakeCatalog(account)
|
|
_FakeBrowser.instances = []
|
|
|
|
class _WrongShopPage:
|
|
url = _FakeBrowser.final_url
|
|
|
|
def evaluate(self, _script: str, _markers: list[str]) -> bool:
|
|
return False
|
|
|
|
class _WrongShopBrowser(_FakeBrowser):
|
|
def run_fetch_action(self, url: str, action, **_options: object):
|
|
page = _WrongShopPage()
|
|
assert url == "https://example.test/console/home"
|
|
return SimpleNamespace(status=200), action(page)
|
|
|
|
monkeypatch.setattr(accounts, "ScraplingBrowser", _WrongShopBrowser)
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=catalog, # type: ignore[arg-type]
|
|
)
|
|
|
|
assert manager.refresh_account(account.account_id) == {
|
|
"status": "expired",
|
|
"final_url_login": False,
|
|
"identity_mismatch": True,
|
|
}
|
|
assert catalog.sync_calls == 0
|
|
assert _WrongShopBrowser.instances[0].persisted is False
|
|
|
|
|
|
def test_identity_mismatch_requires_login_even_with_valid_cookie() -> None:
|
|
assert accounts._keepalive_requires_relogin(
|
|
{"status": "success", "identity_mismatch": True}
|
|
)
|
|
assert accounts._keepalive_requires_relogin({"status": "expired"})
|
|
assert not accounts._keepalive_requires_relogin({"status": "success"})
|
|
|
|
|
|
def test_keepalive_skips_expired_vault_without_starting_browser(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
account = _account(tmp_path)
|
|
store = BrowserCookieStore(account.cookie_file, account.storage_state_file)
|
|
store.save_cookies(
|
|
[
|
|
{
|
|
"name": "session",
|
|
"value": "expired",
|
|
"domain": "example.test",
|
|
"path": "/",
|
|
"expires": 1,
|
|
}
|
|
]
|
|
)
|
|
monkeypatch.setattr(
|
|
accounts,
|
|
"ScraplingBrowser",
|
|
lambda **_options: pytest.fail("expired vault must not start a browser"),
|
|
)
|
|
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=_FakeCatalog(account), # type: ignore[arg-type]
|
|
)
|
|
|
|
assert manager.refresh_account("demo-shop")["status"] == "not_logged_in"
|
|
|
|
|
|
def test_keepalive_runs_configured_login_when_account_vault_is_expired(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
base = _account(tmp_path)
|
|
account = replace(
|
|
base,
|
|
login_mode="C",
|
|
login_command_id="demo.login",
|
|
credential_env_names=("DEMO_ACCOUNT", "DEMO_PASSWORD"),
|
|
)
|
|
store = BrowserCookieStore(account.cookie_file, account.storage_state_file)
|
|
store.save_cookies(
|
|
[
|
|
{
|
|
"name": "session",
|
|
"value": "expired",
|
|
"domain": "example.test",
|
|
"path": "/",
|
|
"expires": 1,
|
|
}
|
|
]
|
|
)
|
|
|
|
login_calls: list[str] = []
|
|
|
|
def login_runner(login_account: AccountBinding) -> dict[str, object]:
|
|
login_calls.append(login_account.account_id)
|
|
fresh_cookie = {
|
|
"name": "session",
|
|
"value": "recovered-session",
|
|
"domain": "example.test",
|
|
"path": "/",
|
|
"expires": 4_102_444_800,
|
|
}
|
|
store.save_cookies([fresh_cookie])
|
|
store.save_storage_state({"cookies": [fresh_cookie], "origins": []})
|
|
return {"status": "success"}
|
|
|
|
monkeypatch.setattr(
|
|
accounts,
|
|
"ScraplingBrowser",
|
|
lambda **_options: pytest.fail("login recovery should run the login flow"),
|
|
)
|
|
catalog = _FakeCatalog(account)
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=catalog, # type: ignore[arg-type]
|
|
login_runner=login_runner,
|
|
)
|
|
|
|
assert manager.refresh_account(account.account_id) == {
|
|
"status": "success",
|
|
"login_recovered": True,
|
|
"cookie_count": 1,
|
|
"synced_members": 1,
|
|
}
|
|
assert login_calls == [account.account_id]
|
|
assert json.loads(account.cookie_file.read_text(encoding="utf-8"))[0]["value"] == (
|
|
"recovered-session"
|
|
)
|
|
|
|
|
|
def test_keepalive_runs_configured_login_after_server_side_expiry(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
base = _account(tmp_path)
|
|
account = replace(
|
|
base,
|
|
login_mode="C",
|
|
login_command_id="demo.login",
|
|
credential_env_names=("DEMO_ACCOUNT", "DEMO_PASSWORD"),
|
|
)
|
|
_write_old_state(account)
|
|
account.cookie_file.with_name("keepalive_status.json").write_text(
|
|
json.dumps({"status": "expired", "final_url_login": True}),
|
|
encoding="utf-8",
|
|
)
|
|
calls: list[str] = []
|
|
|
|
def login_runner(login_account: AccountBinding) -> dict[str, object]:
|
|
calls.append(login_account.account_id)
|
|
return {"status": "success"}
|
|
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=_FakeCatalog(account), # type: ignore[arg-type]
|
|
login_runner=login_runner,
|
|
)
|
|
|
|
assert manager.refresh_account(account.account_id)["status"] == "success"
|
|
assert calls == [account.account_id]
|
|
|
|
|
|
def test_configured_login_keeps_credentials_out_of_process_arguments(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
base = _account(tmp_path)
|
|
account = replace(
|
|
base,
|
|
login_mode="C",
|
|
login_command_id="demo.login",
|
|
credential_env_names=("DEMO_ACCOUNT", "DEMO_PASSWORD"),
|
|
)
|
|
|
|
class _LoginCatalog(_FakeCatalog):
|
|
def binding_for(self, binding_id: str) -> SimpleNamespace:
|
|
assert binding_id == "demo.login"
|
|
return SimpleNamespace(
|
|
command_id="demo.login",
|
|
module="demo",
|
|
entry="login.py",
|
|
)
|
|
|
|
def environment_for(
|
|
self,
|
|
binding_id: str,
|
|
base_environment: dict[str, str],
|
|
) -> dict[str, str]:
|
|
assert binding_id == "demo.login"
|
|
return dict(base_environment)
|
|
|
|
monkeypatch.setenv("DEMO_ACCOUNT", "demo-user")
|
|
monkeypatch.setenv("DEMO_PASSWORD", "demo-password")
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
|
|
captured["command"] = command
|
|
captured["env"] = kwargs["env"]
|
|
return SimpleNamespace(returncode=0)
|
|
|
|
monkeypatch.setattr(accounts.subprocess, "run", fake_run)
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=_LoginCatalog(account), # type: ignore[arg-type]
|
|
)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_login_script_path",
|
|
lambda _binding: tmp_path / "login.py",
|
|
)
|
|
|
|
result = manager._run_configured_login(account)
|
|
|
|
assert result == {
|
|
"status": "success",
|
|
"login_attempted": True,
|
|
"login_exit_code": 0,
|
|
}
|
|
assert captured["command"][-1] == "--login-only"
|
|
assert "demo-password" not in captured["command"]
|
|
environment = captured["env"]
|
|
assert isinstance(environment, dict)
|
|
assert environment["ERP_USERNAME"] == "demo-user"
|
|
assert environment["ERP_PASSWORD"] == "demo-password"
|
|
|
|
|
|
def test_configured_login_can_use_runtime_product_config_without_env_credentials(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
base = _account(tmp_path)
|
|
account = replace(
|
|
base,
|
|
login_mode="C",
|
|
login_command_id="demo.login",
|
|
credential_env_names=("DEMO_ACCOUNT", "DEMO_PASSWORD"),
|
|
)
|
|
config_path = (
|
|
tmp_path / "data" / "state" / "product_commerce" / "auto-flow-config.json"
|
|
)
|
|
config_path.parent.mkdir(parents=True)
|
|
config_path.write_text(
|
|
json.dumps({"login": {"username": "configured-user", "password": "secret"}}),
|
|
encoding="utf-8",
|
|
)
|
|
monkeypatch.delenv("DEMO_ACCOUNT", raising=False)
|
|
monkeypatch.delenv("DEMO_PASSWORD", raising=False)
|
|
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_run(command: list[str], **kwargs: object) -> SimpleNamespace:
|
|
captured["command"] = command
|
|
captured["env"] = kwargs["env"]
|
|
return SimpleNamespace(returncode=0)
|
|
|
|
class _LoginCatalog(_FakeCatalog):
|
|
def binding_for(self, binding_id: str) -> SimpleNamespace:
|
|
assert binding_id == "demo.login"
|
|
return SimpleNamespace(
|
|
command_id="demo.login",
|
|
module="demo",
|
|
entry="login.py",
|
|
)
|
|
|
|
def environment_for(
|
|
self,
|
|
binding_id: str,
|
|
base_environment: dict[str, str],
|
|
) -> dict[str, str]:
|
|
assert binding_id == "demo.login"
|
|
return dict(base_environment)
|
|
|
|
monkeypatch.setattr(accounts.subprocess, "run", fake_run)
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=_LoginCatalog(account), # type: ignore[arg-type]
|
|
)
|
|
monkeypatch.setattr(
|
|
manager,
|
|
"_login_script_path",
|
|
lambda _binding: tmp_path / "login.py",
|
|
)
|
|
|
|
result = manager._run_configured_login(account)
|
|
|
|
assert result == {
|
|
"status": "success",
|
|
"login_attempted": True,
|
|
"login_exit_code": 0,
|
|
}
|
|
assert captured["command"][-1] == "--login-only"
|
|
environment = captured["env"]
|
|
assert isinstance(environment, dict)
|
|
assert environment["GYXX_PRODUCT_CONFIG"] == str(config_path)
|
|
assert "secret" not in captured["command"]
|
|
|
|
|
|
def test_keepalive_initial_delay_staggers_first_refresh(tmp_path: Path) -> None:
|
|
base = _account(tmp_path)
|
|
account = AccountBinding(
|
|
account_id=base.account_id,
|
|
cdp_port=base.cdp_port,
|
|
cdp_url=base.cdp_url,
|
|
profile_dir=base.profile_dir,
|
|
cookie_file=base.cookie_file,
|
|
storage_state_file=base.storage_state_file,
|
|
required_cookie_domains=base.required_cookie_domains,
|
|
required_cookie_names=base.required_cookie_names,
|
|
keepalive_enabled=True,
|
|
keepalive_url=base.keepalive_url,
|
|
keepalive_initial_delay_seconds=60,
|
|
keepalive_interval_seconds=600,
|
|
)
|
|
clock = [100.0]
|
|
manager = accounts.AccountKeepaliveManager(
|
|
Settings(project_root=tmp_path, data_root=tmp_path / "data"),
|
|
catalog=_FakeCatalog(account), # type: ignore[arg-type]
|
|
clock=lambda: clock[0],
|
|
)
|
|
calls: list[str] = []
|
|
manager.refresh_account = lambda account_id: ( # type: ignore[method-assign]
|
|
calls.append(account_id) or {"status": "success"}
|
|
)
|
|
|
|
assert manager.tick() == []
|
|
clock[0] = 159.0
|
|
assert manager.tick() == []
|
|
clock[0] = 160.0
|
|
assert manager.tick() == [{"status": "success"}]
|
|
assert calls == ["demo-shop"]
|
|
|
|
|
|
def test_scheduler_maintenance_runs_in_execute_mode_but_not_dry_run(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
catalog = WorkflowCatalog(
|
|
timezone="Asia/Shanghai",
|
|
workflows=(),
|
|
schedules=(),
|
|
)
|
|
calls: list[str] = []
|
|
now = datetime(2026, 8, 20, 10, 0, tzinfo=ZoneInfo("Asia/Shanghai"))
|
|
|
|
scheduler = PythonScheduler(
|
|
catalog,
|
|
tmp_path / "execute-data",
|
|
launcher=object(), # no workflows are present in this boundary test
|
|
maintenance=lambda: calls.append("execute"),
|
|
)
|
|
scheduler.tick(now)
|
|
|
|
dry_run_scheduler = PythonScheduler(
|
|
catalog,
|
|
tmp_path / "dry-run-data",
|
|
launcher=object(),
|
|
dry_run=True,
|
|
maintenance=lambda: calls.append("dry-run"),
|
|
)
|
|
dry_run_scheduler.tick(now)
|
|
|
|
assert calls == ["execute"]
|