Coverage for src/bayernwerk_client/efix/auth.py: 22%

1"""Interactive login via a real browser (Playwright) for the e-fix service. 

2 

3Same reasoning as `map/auth.py`: e-fix's login is also a Salesforce 

4Experience Cloud (Aura) community (`login.e-fix.info`), reached by first 

5clicking through from the public `www.e-fix.info` landing page - navigating 

6straight to the login domain the way `map` does doesn't work here, the 

7entry point is the landing page's "e-fix Portal Zugang für Installateure" 

8link. 

9 

10After Salesforce authenticates, the e-fix single-page app itself exchanges 

11the returned OAuth `code` for a JWT via `POST 

12https://backend.e-fix.info/api/ciam/token` - that JWT (not anything found 

13in sessionStorage/localStorage) is what the GraphQL API expects as the 

14`Authorization: Bearer` token afterwards (see `client.py`). We capture it 

15directly from that response. 

16 

17Both login fields on this Aura login page report the same broken 

18accessible name ("Passwort *", even the email field) - `get_by_label` 

19can't tell them apart, so fields are selected by `input[type=...]` instead. 

20""" 

21 

22from __future__ import annotations 

23 

24import time 

25from typing import TYPE_CHECKING 

26 

27from bayernwerk_client.exceptions import AuthenticationError 

28from bayernwerk_client.tokens import TokenSet 

29 

30if TYPE_CHECKING: 

31 from playwright.sync_api import Page, Response 

32 

33DEFAULT_PORTAL_URL = "https://www.e-fix.info/bag/index.html" 

34INSTALLER_LOGIN_LINK_TEXT = "e-fix Portal Zugang für Installateure" 

35TOKEN_ENDPOINT_MARKER = "/api/ciam/token" 

36_COOKIE_BUTTON_LABELS = ("Ablehnen", "Alle ablehnen", "Nur essenzielle") 

37 

38 

39def login_interactive( 

40 email: str, 

41 password: str, 

42 *, 

43 portal_url: str = DEFAULT_PORTAL_URL, 

44 headless: bool = False, 

45 timeout: float = 60.0, 

46) -> TokenSet: 

47 """Log in through a real (headed by default) browser and return the resulting tokens. 

48 

49 Raises AuthenticationError on wrong credentials, an unexpected page 

50 (e.g. an MFA prompt this library doesn't handle), or a timeout. 

51 """ 

52 try: 

53 from playwright.sync_api import TimeoutError as PlaywrightTimeoutError 

54 from playwright.sync_api import sync_playwright 

55 except ImportError as exc: 

56 raise AuthenticationError( 

57 "Playwright is required for login. Install with: " 

58 "uv pip install 'bayernwerk-client[login]' && playwright install chromium" 

59 ) from exc 

60 

61 token_data: dict[str, str] = {} 

62 

63 def on_response(response: Response) -> None: 

64 if TOKEN_ENDPOINT_MARKER in response.url and response.request.method == "POST": 

65 try: 

66 body = response.json() 

67 except Exception: # noqa: BLE001 - not every response is JSON 

68 return 

69 if isinstance(body, dict) and "jwtToken" in body: 

70 token_data.update(body) 

71 

72 with sync_playwright() as playwright: 

73 browser = playwright.chromium.launch(headless=headless) 

74 try: 

75 page = browser.new_page() 

76 page.on("response", on_response) 

77 page.goto(portal_url, wait_until="domcontentloaded", timeout=timeout * 1000) 

78 

79 for label in _COOKIE_BUTTON_LABELS: 

80 try: 

81 page.get_by_role("button", name=label).click(timeout=3000) 

82 break 

83 except Exception: # noqa: S112, BLE001 - banner may not appear at all, that's fine 

84 continue 

85 

86 page.get_by_text(INSTALLER_LOGIN_LINK_TEXT, exact=False).first.click(timeout=timeout * 1000) 

87 page.locator('input[type="text"]').first.fill(email, timeout=timeout * 1000) 

88 page.locator('input[type="password"]').first.fill(password) 

89 page.get_by_role("button", name="Anmelden").click() 

90 

91 return _wait_for_token(page, token_data, timeout=timeout) 

92 except PlaywrightTimeoutError as exc: 

93 raise AuthenticationError( 

94 "Timed out during login (wrong credentials, an unhandled MFA/consent prompt, " 

95 "or the portal's login page changed)." 

96 ) from exc 

97 finally: 

98 browser.close() 

99 

100 

101def _wait_for_token(page: Page, token_data: dict[str, str], *, timeout: float) -> TokenSet: 

102 deadline = time.monotonic() + timeout 

103 while time.monotonic() < deadline: 

104 if "jwtToken" in token_data: 

105 return TokenSet.from_access_token(token_data["jwtToken"]) 

106 page.wait_for_timeout(200) 

107 raise AuthenticationError("Login did not produce a token within the timeout")