Coverage for src/shellman/context.py: 95.00%

52 statements  

« prev     ^ index     » next       coverage.py v7.3.0, created at 2023-09-03 19:58 +0200

1"""Jinja-context related utilities.""" 

2 

3from __future__ import annotations 

4 

5import contextlib 

6import json 

7import os 

8from typing import TYPE_CHECKING, Any, Sequence 

9 

10if TYPE_CHECKING: 

11 import argparse 

12 

13ENV_VAR_PREFIX = "SHELLMAN_CONTEXT_" 

14DEFAULT_JSON_FILE = ".shellman.json" 

15 

16 

17def _get_cli_context(args: Sequence[str]) -> dict: 

18 context: dict[str, Any] = {} 

19 if args: 

20 for context_arg in args: 

21 if not context_arg: 

22 continue 

23 if context_arg[0] == "{": 

24 context.update(json.loads(context_arg)) 

25 elif "=" in context_arg: 

26 name, value = context_arg.split("=", 1) 

27 if "." in name: 

28 name_dict: dict[str, Any] = {} 

29 d = name_dict 

30 parts = name.split(".") 

31 for name_part in parts[1:-1]: 

32 d[name_part] = d = {} 

33 d[parts[-1]] = value 

34 context[parts[0]] = name_dict 

35 else: 

36 context[name] = value 

37 # else invalid arg 

38 return context 

39 

40 

41def _get_env_context() -> dict: 

42 context = {} 

43 for env_name, env_value in os.environ.items(): 

44 if env_name.startswith(ENV_VAR_PREFIX): 

45 context_var_name = env_name[len(ENV_VAR_PREFIX) :].lower() 

46 context[context_var_name] = env_value 

47 return context 

48 

49 

50def _get_file_context(file: str) -> dict: 

51 with open(file) as stream: 51 ↛ 52line 51 didn't jump to line 52

52 return json.load(stream) 

53 

54 

55def _get_context(args: argparse.Namespace) -> dict: 

56 context = {} 

57 

58 if args.context_file: 58 ↛ 59line 58 didn't jump to line 59, because the condition on line 58 was never true

59 context.update(_get_file_context(args.context_file)) 

60 else: 

61 with contextlib.suppress(OSError): 

62 context.update(_get_file_context(DEFAULT_JSON_FILE)) 

63 

64 _update(context, _get_env_context()) 

65 _update(context, _get_cli_context(args.context)) 

66 

67 return context 

68 

69 

70def _update(base: dict, added: dict) -> dict: 

71 for key, value in added.items(): 

72 if isinstance(value, dict): 

73 base[key] = _update(base.get(key, {}), value) 

74 else: 

75 base[key] = value 

76 return base