je_web_runner.api.reliability

Façade: retries / locator strength / smart wait / throttler / pool / supervisor.

exception je_web_runner.api.reliability.AdaptiveRetryError

Bases: WebRunnerException

Raised when a non-retryable failure is classified as real.

class je_web_runner.api.reliability.BrowserPool(factory: Callable[[], Any], destructor: Callable[[Any], None] | None = None, health_check: Callable[[Any], bool] | None = None, size: int = 2, max_uses: int = 50)

Bases: object

Thread-safe browser instance pool.

checkin(session: PooledSession) → None

checkout(timeout: float = 30.0) → PooledSession

close() → None

session(timeout: float = 30.0) → Iterator[PooledSession]

property tracked_count: int

warm() → None: Pre-launch size instances eagerly.

exception je_web_runner.api.reliability.BrowserPoolError

Bases: WebRunnerException

Raised when checkout times out or factory fails.

class je_web_runner.api.reliability.FileSemaphore(name: str, capacity: int, base_dir: str, stale_after: float = 600.0)

Bases: object

File-based counting semaphore on a shared directory.

acquire(timeout: float = 30.0, poll: float = 0.1) → Path

base_dir: str

capacity: int

name: str

release(slot: Path) → None

stale_after: float = 600.0

class je_web_runner.api.reliability.LocatorScore(strategy: 'str', value: 'str', score: 'int', reasons: 'List[str]')

Bases: object

reasons: List[str]

score: int

strategy: str

value: str

exception je_web_runner.api.reliability.LocatorStrengthError

Bases: WebRunnerException

Raised when a locator definition can’t be evaluated.

class je_web_runner.api.reliability.OrphanFinding(pid: 'int', name: 'str', command_line: 'str' = '')

Bases: object

command_line: str = ''

name: str

pid: int

class je_web_runner.api.reliability.PooledSession(session_id: int, instance: Any, uses: int = 0, created_at: float = <factory>)

Bases: object

Single browser session managed by the pool.

created_at: float

instance: Any

session_id: int

uses: int = 0

class je_web_runner.api.reliability.ProcessSupervisor(lister: ~typing.Callable[[], ~typing.List[~je_web_runner.utils.process_supervisor.supervisor.OrphanFinding]] = <function default_lister>, killer: ~typing.Callable[[int], bool] = <function default_killer>)

Bases: object

List + kill orphan webdriver processes.

kill_orphans(names: Iterable[str] = ('chromedriver', 'chromedriver.exe', 'geckodriver', 'geckodriver.exe', 'msedgedriver', 'msedgedriver.exe', 'iedriver', 'IEDriverServer.exe'), protected_pids: Iterable[int] | None = None) → Dict[int, bool]

killer() → bool

list_orphans(names: Iterable[str] = ('chromedriver', 'chromedriver.exe', 'geckodriver', 'geckodriver.exe', 'msedgedriver', 'msedgedriver.exe', 'iedriver', 'IEDriverServer.exe')) → List[OrphanFinding]

lister() → List[OrphanFinding]

exception je_web_runner.api.reliability.ProcessSupervisorError

Bases: WebRunnerException

Raised when the process listing call fails.

class je_web_runner.api.reliability.RetryDecision(attempt: int, category: str, sleep_seconds: float, will_retry: bool, error_repr: str)

Bases: object

Outcome of a single retry attempt.

attempt: int

category: str

error_repr: str

sleep_seconds: float

will_retry: bool

class je_web_runner.api.reliability.RetryPolicy(transient_max: int = 3, flaky_max: int = 2, environment_max: int = 1, real_max: int = 0, base_backoff: float = 0.25, max_backoff: float = 4.0, history: List[RetryDecision] = <factory>)

Bases: object

Per-category retry budget; flat backoff per attempt.

backoff_for(attempt: int) → float

base_backoff: float = 0.25

budget_for(category: str) → int

environment_max: int = 1

flaky_max: int = 2

history: List[RetryDecision]

max_backoff: float = 4.0

real_max: int = 0

transient_max: int = 3

class je_web_runner.api.reliability.ServiceThrottler(base_dir: str = '.webrunner/throttle')

Bases: object

Manage multiple named semaphores under a single base directory.

configure(name: str, capacity: int, stale_after: float = 600.0) → None

get(name: str) → FileSemaphore

exception je_web_runner.api.reliability.SmartWaitError

Bases: WebRunnerException

Raised when a smart wait condition does not stabilise in time.

exception je_web_runner.api.reliability.ThrottlerError

Bases: WebRunnerException

Raised when a slot cannot be acquired in time.

je_web_runner.api.reliability.assert_strength(findings: Iterable[Dict[str, Any]], minimum: int = 50, raise_on_fail: bool = True) → List[Dict[str, Any]] | None: Raise (or return) entries below minimum.

je_web_runner.api.reliability.run_with_retry(func: ~typing.Callable[[...], ~typing.Any], *args: ~typing.Any, policy: ~je_web_runner.utils.adaptive_retry.policy.RetryPolicy | None = None, ledger_path: str | None = None, file_path: str | None = None, sleep: ~typing.Callable[[float], None] = <built-in function sleep>, **kwargs: ~typing.Any) → Any: 執行 func，依分類器結果決定是否重試 Call func(*args, **kwargs) and retry only when the classifier labels the exception as transient / flaky / environment. real failures raise AdaptiveRetryError immediately.

je_web_runner.api.reliability.score_action_locators(actions: Iterable[Any]) → List[Dict[str, Any]]: 從 action JSON 中抽取 {test_object_name, object_type} 評分 Walk an action list and score every locator definition encountered.

je_web_runner.api.reliability.score_locator(strategy: str, value: str) → LocatorScore: 對單一 locator 評分 Score a single (strategy, value) pair.

je_web_runner.api.reliability.throttle(name: str, timeout: float = 30.0) → Iterator[Path]: Context-manager wrapper around acquire/release.

je_web_runner.api.reliability.wait_for_fetch_idle(driver: ~typing.Any, quiet_for: float = 0.5, timeout: float = 10.0, poll: float = 0.1, sleep: ~typing.Callable[[float], None] = <built-in function sleep>) → None: 等到 window.fetch 在 quiet_for 秒內都沒有 in-flight 請求 Wait until the fetch hook reports zero in-flight requests for at least quiet_for seconds.

je_web_runner.api.reliability.wait_for_spa_route_stable(driver: ~typing.Any, quiet_for: float = 0.4, timeout: float = 10.0, poll: float = 0.1, sleep: ~typing.Callable[[float], None] = <built-in function sleep>) → None: 等到 history.pushState / replaceState / popstate Wait until no history mutation has fired for at least quiet_fors.

je_web_runner.api.reliability.wait_until(predicate: ~typing.Callable[[], bool], timeout: float = 10.0, poll: float = 0.1, sleep: ~typing.Callable[[float], None] = <built-in function sleep>, description: str = 'predicate') → None: Generic poll-until-true with timeout.

je_web_runner.api.reliability.with_watchdog(callable_obj: Callable[[], Any], timeout_seconds: float) → Any

Run callable_obj() on a daemon thread and raise after timeout_seconds.

The original callable keeps running on its thread; the caller is expected to react to the watchdog raise and clean up the underlying browser via ProcessSupervisor if needed.