mirror of
https://git.zavage.net/Zavage-Software/wabot.git
synced 2026-07-21 13:06:08 -06:00
Mathew Sir Guest the best
154344e94a
wabot/__init__.py defines def sessions(...) and also had a sessions.py submodule, so wabot.sessions (attribute) resolved to the function, shadowing the module for import wabot.sessions as m style access. This is the same class of collision that browser.py -> _browser.py already fixed. Renaming to _sessions.py removes the ambiguity and lets docs/api.rst drop the :no-index: Sphinx workaround in favor of plain autoclass directives on the public SessionRecord/SessionStore re-exports.
81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
"""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 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)
|