diff --git a/tests/integration/test_persistence.py b/tests/integration/test_persistence.py new file mode 100644 index 0000000..b913b32 --- /dev/null +++ b/tests/integration/test_persistence.py @@ -0,0 +1,71 @@ +"""The flagship feature: a second Python PROCESS picks up a saved browser.""" + +import json +import subprocess +import sys +import textwrap + +import pytest + +import wabot + +from .conftest import BROWSERS + +REATTACH_SCRIPT = textwrap.dedent( + """ + import json, sys + import wabot + from wabot.sessions import SessionStore + + store = SessionStore(path=sys.argv[1]) + bot = wabot.browser(session=sys.argv[2], pacing=wabot.NoPacing(), store=store) + value = bot.driver.find_element("id", "username").get_attribute("value") + print(json.dumps({"url": bot.driver.current_url, "username_value": value})) + """ +) + + +@pytest.mark.parametrize("browser_name", BROWSERS) +def test_second_process_picks_up_saved_browser(browser_name, site_url, store, tmp_path): + bot = wabot.browser( + session="flagship", browser=browser_name, headless=True, + pacing=wabot.NoPacing(), store=store, + ) + try: + bot.driver.get(f"{site_url}/login.html") + bot.driver.find_element("id", "username").send_keys("persisted-value") + + # simulate "the process exits": drop our handle without quitting + record = store.get("flagship") + assert record is not None and record.service_pid is not None + + result = subprocess.run( + [sys.executable, "-c", REATTACH_SCRIPT, str(store.path), "flagship"], + capture_output=True, text=True, timeout=120, check=True, + ) + payload = json.loads(result.stdout.strip().splitlines()[-1]) + assert payload["url"] == f"{site_url}/login.html" + assert payload["username_value"] == "persisted-value" + finally: + assert wabot.destroy("flagship", store=store) is True + + +def test_destroyed_session_is_really_gone(site_url, store): + from wabot.hosts import service_alive + + bot = wabot.browser( + session="doomed", browser="chromium", headless=True, + pacing=wabot.NoPacing(), store=store, + ) + bot.driver.get(f"{site_url}/login.html") + record = store.get("doomed") + assert wabot.destroy("doomed", store=store) is True + assert store.get("doomed") is None + assert wabot.destroy("doomed", store=store) is False + # the managed driver service was SIGTERMed + import time + + deadline = time.monotonic() + 10 + while time.monotonic() < deadline and service_alive(record.executor_url): + time.sleep(0.2) + assert not service_alive(record.executor_url)