diff options
Diffstat (limited to 'playwright-py/scripts')
| -rw-r--r-- | playwright-py/scripts/detect_dev_servers.py | 71 | ||||
| -rw-r--r-- | playwright-py/scripts/safe_actions.py | 100 | ||||
| -rwxr-xr-x | playwright-py/scripts/with_server.py | 106 |
3 files changed, 277 insertions, 0 deletions
diff --git a/playwright-py/scripts/detect_dev_servers.py b/playwright-py/scripts/detect_dev_servers.py new file mode 100644 index 0000000..fb8b1ea --- /dev/null +++ b/playwright-py/scripts/detect_dev_servers.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +"""Probe common localhost ports for running HTTP dev servers. + +Outputs JSON: [{"port": N, "url": "http://localhost:N", "server": "<hint>"}, ...] + +Usage: + python detect_dev_servers.py + python detect_dev_servers.py --ports 3000,5173,8000 + python detect_dev_servers.py --host localhost --ports 8080 +""" + +import argparse +import json +import socket +import sys +import urllib.request + +DEFAULT_PORTS = [3000, 3001, 4200, 5000, 5173, 5500, 8000, 8080, 8888, 9000] + + +def is_port_open(host: str, port: int, timeout: float = 0.3) -> bool: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.settimeout(timeout) + try: + s.connect((host, port)) + return True + except (socket.timeout, ConnectionRefusedError, OSError): + return False + + +def probe_http(url: str, timeout: float = 0.5) -> str | None: + """Return a short hint about the server (e.g. its Server header), or None.""" + try: + req = urllib.request.Request(url, headers={"User-Agent": "detect_dev_servers"}) + with urllib.request.urlopen(req, timeout=timeout) as resp: + server = resp.headers.get("Server", "") + return server or "http" + except Exception: + return None + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--ports", + default=",".join(str(p) for p in DEFAULT_PORTS), + help="Comma-separated port list", + ) + parser.add_argument("--host", default="localhost") + args = parser.parse_args() + + try: + ports = [int(p.strip()) for p in args.ports.split(",") if p.strip()] + except ValueError as err: + print(f"ERROR: bad --ports value: {err}", file=sys.stderr) + return 2 + + results = [] + for port in ports: + if is_port_open(args.host, port): + url = f"http://{args.host}:{port}" + hint = probe_http(url) + if hint is not None: + results.append({"port": port, "url": url, "server": hint}) + + print(json.dumps(results, indent=2)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/playwright-py/scripts/safe_actions.py b/playwright-py/scripts/safe_actions.py new file mode 100644 index 0000000..c3f72bf --- /dev/null +++ b/playwright-py/scripts/safe_actions.py @@ -0,0 +1,100 @@ +"""Retry-wrapped Playwright action helpers + common convenience utilities. + +Usage: + from scripts.safe_actions import ( + safe_click, safe_type, handle_cookie_banner, build_context_with_headers + ) +""" + +import json +import os +import time + + +def safe_click(page, selector, retries: int = 3, delay: float = 0.5, timeout: int = 5000): + """Click SELECTOR. Retry up to RETRIES times with DELAY seconds between. + + Raises the last exception if all attempts fail. + """ + last_err = None + for attempt in range(retries): + try: + page.wait_for_selector(selector, timeout=timeout) + page.click(selector) + return + except Exception as err: + last_err = err + if attempt < retries - 1: + time.sleep(delay) + raise last_err # type: ignore[misc] + + +def safe_type(page, selector, value: str, retries: int = 3, delay: float = 0.5, timeout: int = 5000): + """Fill SELECTOR with VALUE. Retry on failure.""" + last_err = None + for attempt in range(retries): + try: + page.wait_for_selector(selector, timeout=timeout) + page.fill(selector, value) + return + except Exception as err: + last_err = err + if attempt < retries - 1: + time.sleep(delay) + raise last_err # type: ignore[misc] + + +def handle_cookie_banner(page, selectors=None) -> bool: + """Try common cookie-banner accept selectors; click the first that exists. + + Returns True if a banner was found and clicked, False otherwise. + Does not raise on failure — many pages have no banner. + """ + selectors = selectors or [ + "#onetrust-accept-btn-handler", + 'button[aria-label*="ccept" i]', + 'button:has-text("Accept")', + 'button:has-text("I agree")', + 'button:has-text("Got it")', + '[data-testid="uc-accept-all-button"]', + "#cookie-accept", + ".cookie-accept", + ] + for selector in selectors: + try: + if page.locator(selector).count() > 0: + page.click(selector, timeout=1000) + return True + except Exception: + continue + return False + + +def build_context_with_headers(browser, extra_kwargs=None): + """Create a browser context with extra HTTP headers from env vars. + + Reads: + PW_HEADER_NAME / PW_HEADER_VALUE — single header + PW_EXTRA_HEADERS='{"X-A":"1","X-B":"2"}' — JSON object of headers + + Unset env vars → plain context with no extra headers. + extra_kwargs, if supplied, are passed to browser.new_context(). + """ + headers: dict[str, str] = {} + name = os.environ.get("PW_HEADER_NAME") + value = os.environ.get("PW_HEADER_VALUE") + if name and value: + headers[name] = value + extra = os.environ.get("PW_EXTRA_HEADERS") + if extra: + try: + parsed = json.loads(extra) + if isinstance(parsed, dict): + headers.update({str(k): str(v) for k, v in parsed.items()}) + except json.JSONDecodeError: + pass + + kwargs: dict = dict(extra_kwargs or {}) + if headers: + kwargs["extra_http_headers"] = headers + return browser.new_context(**kwargs) diff --git a/playwright-py/scripts/with_server.py b/playwright-py/scripts/with_server.py new file mode 100755 index 0000000..431f2eb --- /dev/null +++ b/playwright-py/scripts/with_server.py @@ -0,0 +1,106 @@ +#!/usr/bin/env python3 +""" +Start one or more servers, wait for them to be ready, run a command, then clean up. + +Usage: + # Single server + python scripts/with_server.py --server "npm run dev" --port 5173 -- python automation.py + python scripts/with_server.py --server "npm start" --port 3000 -- python test.py + + # Multiple servers + python scripts/with_server.py \ + --server "cd backend && python server.py" --port 3000 \ + --server "cd frontend && npm run dev" --port 5173 \ + -- python test.py +""" + +import subprocess +import socket +import time +import sys +import argparse + +def is_server_ready(port, timeout=30): + """Wait for server to be ready by polling the port.""" + start_time = time.time() + while time.time() - start_time < timeout: + try: + with socket.create_connection(('localhost', port), timeout=1): + return True + except (socket.error, ConnectionRefusedError): + time.sleep(0.5) + return False + + +def main(): + parser = argparse.ArgumentParser(description='Run command with one or more servers') + parser.add_argument('--server', action='append', dest='servers', required=True, help='Server command (can be repeated)') + parser.add_argument('--port', action='append', dest='ports', type=int, required=True, help='Port for each server (must match --server count)') + parser.add_argument('--timeout', type=int, default=30, help='Timeout in seconds per server (default: 30)') + parser.add_argument('command', nargs=argparse.REMAINDER, help='Command to run after server(s) ready') + + args = parser.parse_args() + + # Remove the '--' separator if present + if args.command and args.command[0] == '--': + args.command = args.command[1:] + + if not args.command: + print("Error: No command specified to run") + sys.exit(1) + + # Parse server configurations + if len(args.servers) != len(args.ports): + print("Error: Number of --server and --port arguments must match") + sys.exit(1) + + servers = [] + for cmd, port in zip(args.servers, args.ports): + servers.append({'cmd': cmd, 'port': port}) + + server_processes = [] + + try: + # Start all servers + for i, server in enumerate(servers): + print(f"Starting server {i+1}/{len(servers)}: {server['cmd']}") + + # Use shell=True to support commands with cd and && + process = subprocess.Popen( + server['cmd'], + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE + ) + server_processes.append(process) + + # Wait for this server to be ready + print(f"Waiting for server on port {server['port']}...") + if not is_server_ready(server['port'], timeout=args.timeout): + raise RuntimeError(f"Server failed to start on port {server['port']} within {args.timeout}s") + + print(f"Server ready on port {server['port']}") + + print(f"\nAll {len(servers)} server(s) ready") + + # Run the command + print(f"Running: {' '.join(args.command)}\n") + result = subprocess.run(args.command) + sys.exit(result.returncode) + + finally: + # Clean up all servers + print(f"\nStopping {len(server_processes)} server(s)...") + for i, process in enumerate(server_processes): + try: + process.terminate() + process.wait(timeout=5) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + print(f"Server {i+1} stopped") + print("All servers stopped") + + +if __name__ == '__main__': + main()
\ No newline at end of file |