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

1from __future__ import annotations 

2 

3import os 

4from pathlib import Path 

5 

6from arclith.domain.ports.secret_resolver import SecretResolver 

7 

8 

9def build_secret_resolver(raw_config: dict, base_path: Path | None = None) -> SecretResolver | None: 

10 """Build a SecretResolver from raw config dict (before Pydantic validation). 

11 

12 Returns None when no mappings are declared (nothing to resolve). 

13 VAULT_ADDR env var overrides secrets.vault.addr from config. 

14 """ 

15 secrets = raw_config.get("secrets") or {} 

16 mappings: dict = secrets.get("mappings") or {} 

17 if not mappings: 

18 return None 

19 

20 resolver_type: str = secrets.get("resolver", "yaml") 

21 

22 def _make(name: str) -> SecretResolver: 

23 match name: 

24 case "vault": 

25 vault_cfg: dict = secrets.get("vault") or {} 

26 addr = os.environ.get("VAULT_ADDR") or vault_cfg.get("addr", "http://127.0.0.1:8200") 

27 mount: str = vault_cfg.get("mount", "kv") 

28 from arclith.adapters.output.vault.secret_adapter import VaultSecretAdapter 

29 return VaultSecretAdapter(addr=addr, mount=mount) 

30 case "yaml": 

31 yaml_cfg: dict = secrets.get("yaml") or {} 

32 default_path = str(base_path / "secrets.yaml") if base_path else "secrets.yaml" 

33 path: str = yaml_cfg.get("path", default_path) 

34 from arclith.adapters.output.yaml.secret_adapter import YamlSecretAdapter 

35 return YamlSecretAdapter(path=path) 

36 case "env": 

37 from arclith.adapters.output.env.secret_adapter import EnvSecretAdapter 

38 return EnvSecretAdapter() 

39 case _: 

40 raise ValueError(f"Unknown secret resolver: '{name}'") 

41 

42 if resolver_type == "chain": 

43 chain_names: list[str] = secrets.get("chain") or ["yaml"] 

44 from arclith.adapters.output.chain.secret_adapter import ChainSecretAdapter 

45 return ChainSecretAdapter([_make(n) for n in chain_names]) 

46 

47 return _make(resolver_type) 

48