#!/usr/bin/env python3
"""Run the stateful fixture smoke test against an existing Firefox RDP."""
from __future__ import annotations
import argparse
import asyncio
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
import json
from pathlib import Path
import subprocess
import tempfile
import threading
from src.browser_session import BrowserSession
from src.config import Config
from src.errors import StaleElementError
from src.rdp_connection import RdpConnection
class QuietHandler(SimpleHTTPRequestHandler):
"""Serve fixtures without writing HTTP access logs."""
def log_message(self, format_string: str, *arguments: object) -> None:
"""Suppress fixture-server diagnostics during the smoke test."""
def findElement(
snapshot: dict[str, object],
*,
name: str | None = None,
input_type: str | None = None,
) -> str:
"""Find one snapshot reference by accessible name or input type."""
elements = snapshot.get("elements")
if not isinstance(elements, list):
raise AssertionError("Snapshot does not contain an elements list.")
for item in elements:
if not isinstance(item, dict):
continue
if name is not None and item.get("accessible_name") != name:
continue
if input_type is not None and item.get("input_type") != input_type:
continue
reference = item.get("element")
if isinstance(reference, str):
return reference
raise AssertionError("Expected fixture element was not in the snapshot.")
async def waitForTab(
rdp: RdpConnection,
url: str,
timeout: float = 10,
) -> str:
"""Wait for the newly opened fixture tab and return its actor ID."""
deadline = asyncio.get_running_loop().time() + timeout
while asyncio.get_running_loop().time() < deadline:
for tab in await rdp.listTabs():
if tab.url == url:
return tab.actor_id
await asyncio.sleep(0.1)
raise AssertionError("Firefox did not open the local fixture tab.")
async def closeFixtureTabs(rdp: RdpConnection) -> int:
"""Close only loopback tabs created from the two test fixtures."""
process = await rdp.request("root", "getProcess", {"id": 0})
descriptor = process.get("processDescriptor")
if not isinstance(descriptor, dict):
raise AssertionError("Firefox did not expose its process descriptor.")
descriptor_actor = descriptor.get("actor")
if not isinstance(descriptor_actor, str):
raise AssertionError("Firefox process descriptor has no actor.")
target = await rdp.request(descriptor_actor, "getTarget")
process_target = target.get("process")
if not isinstance(process_target, dict):
raise AssertionError("Firefox did not expose its parent target.")
console_actor = process_target.get("consoleActor")
if not isinstance(console_actor, str):
raise AssertionError("Firefox parent target has no console actor.")
expression = r"""
(() => {
const names = new Set(['Interaction fixture', 'Upload fixture']);
const tabs = Array.from(gBrowser.tabs).filter(tab =>
names.has(tab.label) &&
tab.linkedBrowser.currentURI.spec.startsWith('http://127.0.0.1:'));
for (const tab of tabs) gBrowser.removeTab(tab);
return JSON.stringify({ok:true,value:{removed:tabs.length}});
})()
"""
result = await rdp.evaluateJson(console_actor, expression)
value = result.get("value")
if not isinstance(value, dict) or not isinstance(
value.get("removed"),
int,
):
raise AssertionError("Firefox returned invalid cleanup metadata.")
return value["removed"]
async def cleanupOnly() -> dict[str, object]:
"""Connect to Firefox and remove leftover local fixture tabs."""
rdp = RdpConnection("127.0.0.1", 6000, 10_000)
await rdp.connect()
try:
return {"removed": await closeFixtureTabs(rdp)}
finally:
await rdp.close()
async def runSmoke(firefox_binary: str) -> dict[str, object]:
"""Exercise all stateful browser tools against local fixture pages."""
fixture_root = Path(__file__).parent / "fixtures"
handler = partial(QuietHandler, directory=str(fixture_root))
server = ThreadingHTTPServer(("127.0.0.1", 0), handler)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
port = server.server_address[1]
interaction_url = f"http://127.0.0.1:{port}/interaction.html"
upload_url = f"http://127.0.0.1:{port}/upload.html"
temporary = tempfile.TemporaryDirectory()
upload_root = Path(temporary.name)
pdf_path = upload_root / "résumé 'fixture'.pdf"
pdf_path.write_bytes(b"%PDF-1.4\nfixture\n%%EOF\n")
config = Config(
host="127.0.0.1",
port=6000,
max_file_bytes=1024 * 1024,
max_upload_bytes=1024 * 1024,
upload_chunk_bytes=7,
max_elements=100,
max_text_chars=10_000,
request_timeout_ms=10_000,
log_level="error",
)
rdp = RdpConnection(config.host, config.port, config.request_timeout_ms)
actor_id: str | None = None
try:
subprocess.run(
[firefox_binary, "--new-tab", interaction_url],
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
timeout=10,
)
for attempt in range(3):
try:
await rdp.connect()
break
except Exception:
if attempt == 2:
raise
await asyncio.sleep(0.5)
actor_id = await waitForTab(rdp, interaction_url)
browser = BrowserSession(config, rdp)
await browser.selectTab(tab_id=actor_id)
javascript = await browser.runJavaScript(
"return document.title;",
config.request_timeout_ms,
)
if javascript != {
"result": "Interaction fixture",
"result_type": "string",
}:
raise AssertionError("JavaScript result did not match the title.")
webdriver = await rdp.evaluateJson(
browser.selected_tab.console_actor_id,
"(() => JSON.stringify({ok:true,value:{webdriver:"
"navigator.webdriver}}))()",
)
if webdriver.get("value", {}).get("webdriver") is not False:
raise AssertionError("navigator.webdriver is unexpectedly true.")
first = await browser.snapshot(
include_all_text=True,
max_elements=100,
max_text_chars=10_000,
)
query = findElement(first, name="Search")
replace_button = findElement(first, name="Replace input")
await browser.fill(query, "replaced", False)
await browser.typeText(query, "!", 0, lambda: False)
await browser.pressKey(query, "Escape")
typed = await browser.snapshot(
include_all_text=False,
max_elements=100,
max_text_chars=10_000,
)
typed_item = next(
item for item in typed["elements"]
if item.get("accessible_name") == "Search"
)
if typed_item.get("value") != "replaced!":
raise AssertionError(
"Fill/type result did not match expected value."
)
await browser.waitFor(
"selector",
"#query",
"present",
1000,
50,
lambda: False,
)
await browser.click(replace_button, 0)
try:
await browser.fill(query, "must not apply", False)
except StaleElementError:
stale_rejected = True
else:
raise AssertionError("A stale element reference was accepted.")
await browser.navigate(
upload_url,
"dom",
10_000,
lambda: False,
)
await browser.waitFor(
"url_contains",
"upload.html",
"present",
1000,
50,
lambda: False,
)
upload_snapshot = await browser.snapshot(
include_all_text=False,
max_elements=100,
max_text_chars=10_000,
)
file_input = findElement(upload_snapshot, input_type="file")
upload = await browser.uploadFiles(
file_input,
[str(pdf_path)],
lambda: False,
)
verification = await rdp.evaluateJson(
browser.selected_tab.console_actor_id,
"(() => JSON.stringify({ok:true,value:{events:"
"document.querySelector('#events').textContent}}))()",
)
events = json.loads(str(verification["value"]["events"]))
if events.get("submitted") is not False:
raise AssertionError("File assignment submitted the fixture form.")
return {
"navigator_webdriver": False,
"ran_javascript": True,
"filled_and_typed": True,
"pressed_key": True,
"clicked": True,
"stale_reference_rejected": stale_rejected,
"navigated_and_waited": True,
"uploaded_names": events.get("names"),
"uploaded_sizes": events.get("sizes"),
"form_submitted": events.get("submitted"),
"upload_result": upload,
}
finally:
if actor_id is not None and rdp.connected:
try:
await closeFixtureTabs(rdp)
except Exception:
pass
await rdp.close()
server.shutdown()
server.server_close()
temporary.cleanup()
def main() -> int:
"""Parse the Firefox binary and print the smoke-test JSON result."""
parser = argparse.ArgumentParser()
parser.add_argument("--firefox", default="firefox-bin")
parser.add_argument("--cleanup-only", action="store_true")
arguments = parser.parse_args()
operation = cleanupOnly() if arguments.cleanup_only else runSmoke(
arguments.firefox
)
result = asyncio.run(operation)
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())