Coverage for src/lilbee/core/config/load_warnings.py: 100%

1"""What a settings source reports about a value it refuses: a warning, or a variable that stops.""" 

2 

3from __future__ import annotations 

4 

5import hashlib 

6import logging 

7import os 

8import threading 

9from collections.abc import Iterator 

10from contextlib import contextmanager 

11from contextvars import ContextVar 

12 

13log = logging.getLogger(__name__) 

14 

15_MARK_LENGTH = 12 

16 

17# A child inherits the marks the startup load logged, so it does not log them again. 

18SHOWN_ENV_VAR = "LILBEE_LOAD_WARNINGS_SHOWN" 

19_MARK_SEPARATOR = "," 

20 

21# Marks of the warnings this process already logged, seeded from the parent's. 

22_shown: set[str] = { 

23 mark for mark in os.environ.get(SHOWN_ENV_VAR, "").split(_MARK_SEPARATOR) if mark 

24} 

25_shown_lock = threading.Lock() 

26 

27_collected: ContextVar[list[str] | None] = ContextVar("lilbee_load_warnings", default=None) 

28 

29 

30class RefusedVariableError(ValueError): 

31 """The environment sets a LILBEE_* variable lilbee does not run with.""" 

32 

33 

34def _mark(message: str) -> str: 

35 """A short stable name for *message* that fits an environment variable.""" 

36 return hashlib.sha256(message.encode("utf-8")).hexdigest()[:_MARK_LENGTH] 

37 

38 

39def warn_on_load(message: str, version: str = "") -> None: 

40 """Log *message* unless this process already did, and keep it for a collecting load. 

41 

42 With *version*, the state of the file the message is about, a message that 

43 was logged for another state of that file is logged again. 

44 """ 

45 found = _collected.get() 

46 if found is not None and message not in found: 

47 found.append(message) 

48 mark = _mark(message + version) 

49 with _shown_lock: 

50 if mark in _shown: 

51 return 

52 _shown.add(mark) 

53 log.warning("%s", message) 

54 

55 

56def shown_marks() -> str: 

57 """The marks logged so far, as the value a child inherits in ``SHOWN_ENV_VAR``.""" 

58 with _shown_lock: 

59 return _MARK_SEPARATOR.join(sorted(_shown)) 

60 

61 

62def forget_shown() -> None: 

63 """Forget which warnings this process logged, so the next one logs again.""" 

64 with _shown_lock: 

65 _shown.clear() 

66 

67 

68def refuse_variable(message: str) -> None: 

69 """Raise RefusedVariableError, except in a collecting load, whose entry point raises it.""" 

70 if _collected.get() is None: 

71 raise RefusedVariableError(message) 

72 

73 

74@contextmanager 

75def collecting() -> Iterator[list[str]]: 

76 """Yield the list that receives each load warning; a refused variable does not raise inside.""" 

77 found: list[str] = [] 

78 token = _collected.set(found) 

79 try: 

80 yield found 

81 finally: 

82 _collected.reset(token)