368 lines
11 KiB
Python
368 lines
11 KiB
Python
|
|
import pytest
|
|
|
|
from gyxx_flow.modules.content_marketing import (
|
|
pgy_xhs_scraper_v2,
|
|
xingtu_scraper_v2,
|
|
)
|
|
from gyxx_flow.modules.content_marketing.data.tools import (
|
|
relogin_douyin,
|
|
relogin_pgy,
|
|
relogin_xingtu,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("module", [relogin_pgy, relogin_xingtu])
|
|
def test_failed_relogin_restores_cookie_and_profile(tmp_path, monkeypatch, module):
|
|
cookie = tmp_path / "cookies.json"
|
|
profile = tmp_path / "profile"
|
|
cookie.write_text("old-cookie", encoding="utf-8")
|
|
profile.mkdir()
|
|
(profile / "state.txt").write_text("old-profile", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(module, "COOKIE_FILE", cookie)
|
|
monkeypatch.setattr(module, "PROFILE_DIR", profile)
|
|
|
|
cookie_backup, profile_backup = module.reset_state()
|
|
assert not cookie.exists()
|
|
assert not profile.exists()
|
|
|
|
cookie.write_text("partial-new-cookie", encoding="utf-8")
|
|
profile.mkdir()
|
|
(profile / "partial.txt").write_text("partial", encoding="utf-8")
|
|
|
|
module.restore_previous_state(cookie_backup, profile_backup)
|
|
|
|
assert cookie.read_text(encoding="utf-8") == "old-cookie"
|
|
assert (profile / "state.txt").read_text(encoding="utf-8") == "old-profile"
|
|
assert not (profile / "partial.txt").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("module", [relogin_pgy, relogin_xingtu])
|
|
def test_successful_relogin_keeps_new_cookie(tmp_path, monkeypatch, module):
|
|
cookie = tmp_path / "cookies.json"
|
|
profile = tmp_path / "profile"
|
|
cookie.write_text("old-cookie", encoding="utf-8")
|
|
profile.mkdir()
|
|
|
|
monkeypatch.setattr(module, "COOKIE_FILE", cookie)
|
|
monkeypatch.setattr(module, "PROFILE_DIR", profile)
|
|
|
|
cookie_backup, profile_backup = module.reset_state()
|
|
cookie.write_text("new-cookie", encoding="utf-8")
|
|
profile.mkdir()
|
|
|
|
module.finish_successful_relogin(profile_backup)
|
|
|
|
assert cookie.read_text(encoding="utf-8") == "new-cookie"
|
|
assert cookie_backup.exists()
|
|
assert not profile_backup.exists()
|
|
|
|
|
|
def test_xingtu_scan_login_opens_customer_sso_and_clicks_douyin():
|
|
class FakeLocator:
|
|
def __init__(self):
|
|
self.first = self
|
|
self.clicked = False
|
|
self.evaluated = False
|
|
|
|
def count(self):
|
|
return 1
|
|
|
|
def is_visible(self):
|
|
return True
|
|
|
|
def click(self, timeout):
|
|
self.clicked = True
|
|
|
|
def evaluate(self, script):
|
|
assert script == "e => e.click()"
|
|
self.evaluated = True
|
|
|
|
class FakePage:
|
|
url = "https://www.xingtu.cn/?redirect_uri=/ad/creator/market"
|
|
|
|
def __init__(self):
|
|
self.goto_url = None
|
|
self.selectors = []
|
|
self.douyin = FakeLocator()
|
|
|
|
def goto(self, url, **kwargs):
|
|
self.goto_url = url
|
|
self.url = url
|
|
|
|
def wait_for_timeout(self, _milliseconds):
|
|
pass
|
|
|
|
def locator(self, selector):
|
|
self.selectors.append(selector)
|
|
return self.douyin
|
|
|
|
def get_by_text(self, _pattern):
|
|
return FakeLocator()
|
|
|
|
page = FakePage()
|
|
xingtu_scraper_v2.open_scan_login(page)
|
|
|
|
assert page.goto_url == xingtu_scraper_v2.CUSTOMER_LOGIN_URL
|
|
assert 'img[src*="aweme.png"]' in page.selectors
|
|
assert page.douyin.clicked
|
|
|
|
|
|
def test_xingtu_index_page_is_recognized_as_logged_in():
|
|
class Locator:
|
|
@property
|
|
def first(self):
|
|
return self
|
|
|
|
def wait_for(self, timeout):
|
|
assert timeout == 1200
|
|
|
|
class Page:
|
|
url = "https://www.xingtu.cn/ad/creator/index"
|
|
|
|
def get_by_text(self, _text, exact=False):
|
|
return Locator()
|
|
|
|
assert xingtu_scraper_v2.is_logged_in(Page()) is True
|
|
|
|
|
|
def test_xingtu_public_homepage_is_not_logged_in_even_with_stale_cookie():
|
|
class Context:
|
|
def cookies(self):
|
|
return [{"name": "sessionid", "value": "stale"}]
|
|
|
|
class Page:
|
|
url = "https://www.xingtu.cn/?redirect_uri=/ad/creator/market"
|
|
context = Context()
|
|
|
|
def get_by_text(self, *_args, **_kwargs):
|
|
raise AssertionError("public homepage must be rejected before marker checks")
|
|
|
|
assert xingtu_scraper_v2.is_logged_in(Page()) is False
|
|
|
|
|
|
def test_xingtu_market_page_without_business_ui_rejects_stale_cookie():
|
|
class Locator:
|
|
@property
|
|
def first(self):
|
|
return self
|
|
|
|
def wait_for(self, **_kwargs):
|
|
raise xingtu_scraper_v2.PlaywrightTimeoutError("missing")
|
|
|
|
def count(self):
|
|
return 0
|
|
|
|
class Context:
|
|
def cookies(self):
|
|
return [{"name": "sessionid", "value": "stale"}]
|
|
|
|
class Page:
|
|
url = "https://www.xingtu.cn/ad/creator/market"
|
|
context = Context()
|
|
|
|
def get_by_text(self, *_args, **_kwargs):
|
|
return Locator()
|
|
|
|
def locator(self, *_args, **_kwargs):
|
|
return Locator()
|
|
|
|
assert xingtu_scraper_v2.is_logged_in(Page()) is False
|
|
|
|
|
|
def test_xingtu_cookie_validation_requires_real_session_cookie(tmp_path):
|
|
cookie = tmp_path / "xingtu.json"
|
|
cookie.write_text('[{"name":"other","value":"1"}]', encoding="utf-8")
|
|
assert xingtu_scraper_v2.has_valid_login_cookie_file(cookie) is False
|
|
|
|
cookie.write_text('[{"name":"sessionid","value":"fresh"}]', encoding="utf-8")
|
|
assert xingtu_scraper_v2.has_valid_login_cookie_file(cookie) is True
|
|
|
|
|
|
def test_xingtu_login_only_does_not_report_success_when_page_action_was_swallowed(
|
|
tmp_path, monkeypatch
|
|
):
|
|
cookie = tmp_path / "xingtu.json"
|
|
cookie.write_text('[{"name":"sessionid","value":"old"}]', encoding="utf-8")
|
|
|
|
class Session:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def fetch(self, *_args, **_kwargs):
|
|
# Scrapling can log a page_action error and still return a response.
|
|
return object()
|
|
|
|
monkeypatch.setattr(xingtu_scraper_v2, "COOKIE_FILE", cookie)
|
|
monkeypatch.setattr(xingtu_scraper_v2, "DynamicSession", lambda **_kwargs: Session())
|
|
monkeypatch.setattr(xingtu_scraper_v2, "_force_cleanup_session", lambda _session: None)
|
|
|
|
assert xingtu_scraper_v2.login_only(1) == 1
|
|
|
|
|
|
def test_xingtu_login_only_verifies_persisted_profile_after_oauth_context_closes(
|
|
tmp_path, monkeypatch
|
|
):
|
|
cookie = tmp_path / "xingtu.json"
|
|
|
|
class Session:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def fetch(self, *_args, **_kwargs):
|
|
raise RuntimeError("Target page, context or browser has been closed")
|
|
|
|
monkeypatch.setattr(xingtu_scraper_v2, "COOKIE_FILE", cookie)
|
|
monkeypatch.setattr(xingtu_scraper_v2, "DynamicSession", lambda **_kwargs: Session())
|
|
monkeypatch.setattr(xingtu_scraper_v2, "_force_cleanup_session", lambda _session: None)
|
|
|
|
def verify():
|
|
cookie.write_text('[{"name":"sessionid","value":"fresh"}]', encoding="utf-8")
|
|
return True
|
|
|
|
monkeypatch.setattr(xingtu_scraper_v2, "verify_persisted_profile_login", verify)
|
|
|
|
assert xingtu_scraper_v2.login_only(1) == 0
|
|
|
|
|
|
def test_xingtu_scan_login_follows_replacement_page_after_oauth_tab_closes(monkeypatch):
|
|
class Context:
|
|
def __init__(self):
|
|
self.pages = []
|
|
|
|
def new_page(self):
|
|
raise AssertionError("a surviving replacement page should be reused")
|
|
|
|
class Page:
|
|
def __init__(self, context, url, *, closes_on_wait=False):
|
|
self.context = context
|
|
self.url = url
|
|
self.closed = False
|
|
self.closes_on_wait = closes_on_wait
|
|
|
|
def is_closed(self):
|
|
return self.closed
|
|
|
|
def wait_for_timeout(self, _milliseconds):
|
|
if self.closes_on_wait:
|
|
self.closed = True
|
|
|
|
def goto(self, url, **_kwargs):
|
|
self.url = url
|
|
|
|
context = Context()
|
|
qr_page = Page(
|
|
context,
|
|
"https://open.douyin.com/platform/oauth/pc/auth",
|
|
closes_on_wait=True,
|
|
)
|
|
market_page = Page(context, xingtu_scraper_v2.MARKET_URL)
|
|
context.pages = [qr_page, market_page]
|
|
saved = []
|
|
|
|
monkeypatch.setattr(xingtu_scraper_v2, "open_scan_login", lambda _page: None)
|
|
monkeypatch.setattr(
|
|
xingtu_scraper_v2, "is_scan_qr_page", lambda page: page is qr_page
|
|
)
|
|
monkeypatch.setattr(
|
|
xingtu_scraper_v2, "is_logged_in", lambda page: page is market_page
|
|
)
|
|
monkeypatch.setattr(xingtu_scraper_v2, "save_state", lambda page: saved.append(page))
|
|
|
|
result = xingtu_scraper_v2.wait_for_scan_login(qr_page, 2)
|
|
|
|
assert result is market_page
|
|
assert saved == [market_page]
|
|
|
|
|
|
def test_xingtu_interrupted_transaction_is_recovered(tmp_path, monkeypatch):
|
|
cookie = tmp_path / "xingtu_cookies.json"
|
|
profile = tmp_path / "profile"
|
|
cookie.write_text("old-cookie", encoding="utf-8")
|
|
profile.mkdir()
|
|
(profile / "old.txt").write_text("old-profile", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(relogin_xingtu, "COOKIE_FILE", cookie)
|
|
monkeypatch.setattr(relogin_xingtu, "PROFILE_DIR", profile)
|
|
|
|
cookie_backup, profile_backup = relogin_xingtu.reset_state()
|
|
profile.mkdir()
|
|
(profile / "partial.txt").write_text("partial", encoding="utf-8")
|
|
|
|
assert relogin_xingtu.recover_interrupted_state() is True
|
|
assert cookie.read_text(encoding="utf-8") == "old-cookie"
|
|
assert (profile / "old.txt").read_text(encoding="utf-8") == "old-profile"
|
|
assert not (profile / "partial.txt").exists()
|
|
assert cookie_backup is not None
|
|
assert profile_backup is not None
|
|
|
|
|
|
def test_pgy_scan_login_clicks_qr_switch_image():
|
|
class FakeLocator:
|
|
def __init__(self):
|
|
self.first = self
|
|
self.clicked = False
|
|
self.evaluated = False
|
|
|
|
def count(self):
|
|
return 1
|
|
|
|
def is_visible(self):
|
|
return True
|
|
|
|
def click(self, timeout):
|
|
self.clicked = True
|
|
|
|
def evaluate(self, script):
|
|
assert script == "e => e.click()"
|
|
self.evaluated = True
|
|
|
|
class FakePage:
|
|
def __init__(self):
|
|
self.selectors = []
|
|
self.qr_switch = FakeLocator()
|
|
|
|
def wait_for_timeout(self, _milliseconds):
|
|
pass
|
|
|
|
def get_by_text(self, _pattern):
|
|
return FakeLocator()
|
|
|
|
def locator(self, selector):
|
|
self.selectors.append(selector)
|
|
return self.qr_switch
|
|
|
|
page = FakePage()
|
|
pgy_xhs_scraper_v2.open_scan_login(page)
|
|
|
|
assert 'img[src*="qr_code"]' in page.selectors
|
|
assert page.qr_switch.evaluated
|
|
|
|
|
|
def test_failed_douyin_relogin_restores_cookie_storage_and_profile(tmp_path, monkeypatch):
|
|
cookie = tmp_path / "douyin_cookies.json"
|
|
storage = tmp_path / "douyin_storage_state.json"
|
|
profile = tmp_path / "profile"
|
|
cookie.write_text('[{"name":"sessionid","value":"old"}]', encoding="utf-8")
|
|
storage.write_text("old-storage", encoding="utf-8")
|
|
profile.mkdir()
|
|
(profile / "old.txt").write_text("old-profile", encoding="utf-8")
|
|
|
|
monkeypatch.setattr(relogin_douyin, "COOKIE_FILE", cookie)
|
|
monkeypatch.setattr(relogin_douyin, "STATE_FILE", storage)
|
|
monkeypatch.setattr(relogin_douyin, "PROFILE_DIR", profile)
|
|
monkeypatch.setattr(relogin_douyin.subprocess, "call", lambda *_args, **_kwargs: 1)
|
|
monkeypatch.setattr(relogin_douyin.sys, "argv", ["relogin_douyin.py"])
|
|
|
|
assert relogin_douyin.main() == 1
|
|
assert cookie.read_text(encoding="utf-8") == '[{"name":"sessionid","value":"old"}]'
|
|
assert storage.read_text(encoding="utf-8") == "old-storage"
|
|
assert (profile / "old.txt").read_text(encoding="utf-8") == "old-profile"
|