Coverage for src/lilbee/crawler/stored_login.py: 100%

22 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-11-06 08:17 +0000

1"""The login the crawl settings store, and the one login a crawl sends.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import asdict 

6 

7from lilbee.core.config import cfg 

8from lilbee.core.config.login_host import login_host 

9from lilbee.crawler.models import NO_LOGIN, CrawlLogin 

10from lilbee.crawler.url_filter import parsed_url 

11 

12ONE_HOST_MESSAGE = ( 

13 "a crawl login goes to one host, and these URLs name {count}; " 

14 "add the URLs of each host in their own command" 

15) 

16STORED_LOGIN_NOT_SENT = ( 

17 "The stored crawl login was not sent: crawl_login_host " 

18 "(LILBEE_CRAWL_LOGIN_HOST) does not name the one host of these URLs." 

19) 

20 

21 

22def url_hosts(urls: list[str]) -> set[str]: 

23 """The hosts of *urls* as the URL standard reads them; a URL that does not parse is its own.""" 

24 return {parsed.hostname if (parsed := parsed_url(url)) else url for url in urls} 

25 

26 

27def _stored_fields() -> dict[str, str]: 

28 """The login fields the settings hold, whatever host they name.""" 

29 stored = { 

30 "authorization": cfg.crawl_authorization, 

31 "username": cfg.crawl_username, 

32 "password": cfg.crawl_password, 

33 "cookie": cfg.crawl_cookie, 

34 } 

35 return {name: value for name, value in stored.items() if value} 

36 

37 

38def stored_login_withheld(urls: list[str]) -> bool: 

39 """Whether the settings hold a login that `crawl_login_host` does not send to *urls*.""" 

40 return bool(_stored_fields()) and url_hosts(urls) != {login_host(cfg.crawl_login_host)} 

41 

42 

43def crawl_login(urls: list[str], typed: CrawlLogin = NO_LOGIN) -> CrawlLogin: 

44 """The login for the crawl of *urls*: *typed* over the stored one, all for one host. 

45 

46 Raises ``ValueError`` when a typed login meets URLs on more than one host. 

47 """ 

48 hosts = url_hosts(urls) 

49 typed_fields = {name: value for name, value in asdict(typed).items() if value} 

50 if typed_fields and len(hosts) > 1: 

51 raise ValueError(ONE_HOST_MESSAGE.format(count=len(hosts))) 

52 stored = {} if stored_login_withheld(urls) else _stored_fields() 

53 return CrawlLogin(**{**stored, **typed_fields})