Coverage for arclith / infrastructure / secret_loader.py: 100%

1from __future__ import annotations 

2 

3from copy import deepcopy 

4from typing import TYPE_CHECKING 

5 

6if TYPE_CHECKING: 

7 from arclith.domain.ports.secret_resolver import SecretResolver 

8 

9 

10def resolve_dict_secrets(data: dict, resolver: "SecretResolver") -> dict: 

11 """Inject resolved secrets into a raw config dict. 

12 

13 Operates on a deepcopy — the original dict is never mutated. 

14 Works with any dict structure: AppConfig, AgentConfig, or any custom config. 

15 

16 :param data: raw config dict (from yaml.safe_load) 

17 :param resolver: SecretResolver instance (Vault, Yaml, Env, Chain…) 

18 :returns: new dict with secrets injected at their dot-notation paths 

19 """ 

20 secrets: dict = data.get("secrets") or {} 

21 mappings: dict[str, str] = secrets.get("mappings") or {} 

22 if not mappings: 

23 return data 

24 

25 result = deepcopy(data) 

26 missing: list[str] = [] 

27 

28 for field_path, secret_key in mappings.items(): 

29 value = resolver.get(field_path, secret_key) 

30 if value is not None: 

31 _set_nested(result, field_path, value) 

32 else: 

33 missing.append(field_path) 

34 

35 if missing: 

36 raise RuntimeError( 

37 f"Secrets non résolus pour les champs suivants : {missing}. " 

38 "Vérifiez votre Vault, secrets.yaml ou variables d'environnement." 

39 ) 

40 

41 return result 

42 

43 

44def _set_nested(data: dict, path: str, value: str) -> None: 

45 keys = path.split(".") 

46 current = data 

47 for key in keys[:-1]: 

48 if not isinstance(current.get(key), dict): 

49 current[key] = {} 

50 current = current[key] 

51 current[keys[-1]] = value 

52