Coverage for volexport/main.py: 93%

1import os 

2import click 

3import uvicorn 

4import functools 

5from typing import Optional 

6from logging import getLogger 

7from .version import VERSION 

8 

9_log = getLogger(__name__) 

10 

11 

12@click.version_option(version=VERSION, prog_name="volexport") 

13@click.group(invoke_without_command=True) 

14@click.pass_context 

15def cli(ctx): 

16 if ctx.invoked_subcommand is None: 16 ↛ 17line 16 didn't jump to line 17 because the condition on line 16 was never true

17 click.echo(ctx.get_help()) 

18 

19 

20def set_verbose(verbose: Optional[bool]): 

21 from logging import basicConfig 

22 

23 fmt = "%(asctime)s %(levelname)s %(name)s %(message)s" 

24 level = "INFO" 

25 if verbose: 

26 level = "DEBUG" 

27 elif verbose is not None: 27 ↛ 29line 27 didn't jump to line 29 because the condition on line 27 was always true

28 level = "WARNING" 

29 basicConfig(level=level, format=fmt) 

30 

31 

32def verbose_option(func): 

33 @functools.wraps(func) 

34 def wrap(verbose, *args, **kwargs): 

35 set_verbose(verbose) 

36 return func(*args, **kwargs) 

37 

38 return click.option("--verbose/--quiet", help="log level")(wrap) 

39 

40 

41@cli.command() 

42@verbose_option 

43@click.option("--become-method", help="sudo/doas/runas, etc...") 

44@click.option("--tgtadm-bin", help="tgtadm command") 

45@click.option("--tgt-bstype", help="backing store type") 

46@click.option("--tgt-bsopts", help="bs options") 

47@click.option("--tgt-bsoflags", help="bs open flags") 

48@click.option("--lvm-bin", help="lvm command") 

49@click.option("--nics", multiple=True, help="use interfaces") 

50@click.option("--iqn-base", help="iSCSI target basename") 

51@click.option("--vg", help="LVM volume group") 

52@click.option("--host", default="127.0.0.1", help="listen host") 

53@click.option("--port", default=8080, type=int, help="listen port") 

54@click.option("--log-config", type=click.Path(), help="uvicorn log config") 

55@click.option("--cmd-timeout", type=float, help="command execution timeout") 

56def server(host, port, log_config, **kwargs): 

57 import json 

58 

59 for k, v in kwargs.items(): 

60 if k is None or v is None or (isinstance(v, tuple) and len(v) == 0): 

61 continue 

62 kk = f"VOLEXP_{k.upper()}" 

63 if isinstance(v, tuple): 

64 vv = json.dumps(list(v)) 

65 else: 

66 vv = v 

67 os.environ[kk] = vv 

68 

69 from .api import api 

70 from .config import config 

71 

72 _log.debug("config: %s", config) 

73 if log_config is None: 73 ↛ 76line 73 didn't jump to line 76 because the condition on line 73 was always true

74 getLogger("uvicorn").setLevel("INFO") 

75 

76 uvicorn.run(api, host=host, port=port, log_config=log_config) 

77 

78 

79@cli.command() 

80@click.option("--format", type=click.Choice(["yaml", "json"]), default="yaml", show_default=True) 

81def apispec(format): 

82 import sys 

83 

84 os.environ["VOLEXP_VG"] = "dummy" 

85 os.environ["VOLEXP_NICS"] = "[]" 

86 from .api import api 

87 

88 if format == "yaml": 

89 import yaml 

90 

91 yaml.dump(api.openapi(), stream=sys.stdout) 

92 elif format == "json": 92 ↛ exitline 92 didn't return from function 'apispec' because the condition on line 92 was always true

93 import json 

94 

95 json.dump(api.openapi(), fp=sys.stdout) 

96 

97 

98if __name__ == "__main__": 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 cli()