Coverage for rfpy/conf/settings.py: 100%

1import os 

2import re 

3import sys 

4import time 

5import uuid 

6from enum import Enum 

7from pathlib import Path 

8from typing import Iterator, Tuple 

9 

10from pydantic import BaseSettings, EmailStr, root_validator 

11 

12 

13class RunMode(str, Enum): 

14 production = 'production' 

15 development = 'development' 

16 test = 'test' 

17 

18 

19class Mailer(str, Enum): 

20 postmark = 'postmark' 

21 logfile = 'logfile' 

22 

23 

24LIVE_SETTINGS_PATH = '/etc/postrfp/rfpy-settings.env' 

25LOCAL_SETTINGS_PATH = './rfpy-settings.env' 

26 

27 

28class AppSettings(BaseSettings): 

29 

30 class Config: 

31 env_prefix = 'rfpy_' 

32 env_file = LIVE_SETTINGS_PATH 

33 env_file_encoding = 'utf-8' 

34 private_keys: set = { 

35 'db_password', 'crypt_key', 'postmark_api_key', 

36 'postmark_logging_key', 'fanout_realm_key' 

37 } 

38 

39 run_mode: RunMode = RunMode.production 

40 

41 db_name: str = 'rfp' 

42 db_host: str = 'localhost' 

43 db_user: str = 'root' 

44 db_password: str = '' 

45 db_driver: str = 'mysqldb' 

46 crypt_key: str 

47 webapp_hostname: str = 'localhost' 

48 data_directory: str = '/var/supplierselect/' 

49 

50 remote_events_url: str = 'DUMMY_LOG_SERVER' 

51 

52 mailer: Mailer = Mailer.logfile 

53 email_from_address: EmailStr = '[email protected]' 

54 email_to_override: EmailStr = '[email protected]' 

55 postmark_api_key: str = 'POSTMARK_API_TEST' 

56 postmark_logging_key: str = 'POSTMARK_API_TEST' 

57 system_name: str = 'PostRFP' 

58 template_dir: str = 'conf/postmarktmpls/txt/' 

59 fanout_realm_id: str = None 

60 fanout_realm_key: str = None 

61 

62 @property 

63 def sqlalchemy_dsn(self) -> str: 

64 return (f'mysql+{self.db_driver}://{self.db_user}:{self.db_password}' 

65 f'@{self.db_host}/{self.db_name}?charset=utf8mb4') 

66 

67 @property 

68 def cache_dir(self) -> Path: 

69 return Path(self.data_directory) / "cache" / self.db_name 

70 

71 @property 

72 def attachments_dir(self) -> Path: 

73 return Path(self.data_directory) / 'attachments' / self.db_name 

74 

75 @root_validator 

76 def validate_dirs(cls, values): 

77 ''' 

78 Check that the data directory exists, that correct permissions are set 

79 and that cache and attachment subdirectories exists 

80 ''' 

81 data_directory = Path(values['data_directory']) 

82 if not data_directory.is_dir(): 

83 m = f'\n !!! {data_directory} directory not found - required for cache/ and attachments/' 

84 sys.exit(m) 

85 

86 for folder in ['cache', 'attachments']: 

87 app_directory = data_directory / folder 

88 app_directory.mkdir(exist_ok=True) 

89 cls.test_usable(app_directory) 

90 

91 return values 

92 

93 def random_cache_file_path(self) -> Tuple[str, Path]: 

94 fname = '%s.tmp' % uuid.uuid4().hex 

95 return fname, self.cache_dir / fname 

96 

97 def conn_string(self, db_name=None): 

98 name = self.db_name if db_name is None else db_name 

99 return re.sub(r'(?<=\/)\w+(?=\?)', name, self.sqlalchemy_dsn) 

100 

101 @classmethod 

102 def test_usable(cls, dir_name): 

103 try: 

104 ts = str(time.time()) 

105 test_file_path = os.path.join(dir_name, 'test.txt') 

106 with open(test_file_path, 'w') as tw: 

107 tw.write(ts) 

108 with open(test_file_path, 'r') as tr: 

109 tr.read() 

110 os.remove(test_file_path) 

111 except IOError as ioe: 

112 sys.exit('Test read/write to data directory %s failed with %s' % (dir_name, ioe)) 

113 

114 def __iter__(self) -> Iterator[Tuple[str, str]]: 

115 for key, value in super().__iter__(): 

116 if key in self.Config.private_keys: 

117 yield (key, '------------') 

118 continue 

119 yield (key, value)