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 Tuple, Any, Generator, Self 

9 

10 

11from pydantic_settings import SettingsConfigDict, BaseSettings 

12from pydantic import EmailStr, model_validator 

13 

14 

15class RunMode(str, Enum): 

16 production = "production" 

17 development = "development" 

18 test = "test" 

19 

20 

21class Mailer(str, Enum): 

22 postmark = "postmark" 

23 logfile = "logfile" 

24 

25 

26LIVE_SETTINGS_PATH = "/etc/postrfp/rfpy-settings.env" 

27LOCAL_SETTINGS_PATH = "./rfpy-settings.env" 

28 

29 

30class AppSettings(BaseSettings): 

31 model_config = SettingsConfigDict( 

32 env_prefix="rfpy_", env_file=LIVE_SETTINGS_PATH, env_file_encoding="utf-8" 

33 ) 

34 

35 run_mode: RunMode = RunMode.production 

36 

37 db_name: str = "rfp" 

38 db_host: str = "localhost" 

39 db_user: str = "root" 

40 db_password: str = "" 

41 db_driver: str = "mysqldb" 

42 crypt_key: str = "" 

43 webapp_hostname: str = "localhost" 

44 data_directory: str = "/var/supplierselect/" 

45 # jwt_expiry_hours: int = 72 # Default to 3 days 

46 jwt_expiry_minutes: int = 20 # Default to 20 minutes 

47 jwt_refresh_token_expiry_hours:int = 168 # Default to 7 days 

48 

49 max_failed_login_attempts: int = 5 

50 lockout_duration_minutes: int = 20 

51 

52 remote_events_url: str = "DUMMY_LOG_SERVER" 

53 

54 mailer: Mailer = Mailer.logfile 

55 email_from_address: EmailStr = "[email protected]" 

56 email_to_override: EmailStr = "[email protected]" 

57 postmark_api_key: str = "POSTMARK_API_TEST" 

58 postmark_logging_key: str = "POSTMARK_API_TEST" 

59 system_name: str = "PostRFP" 

60 template_dir: str = "conf/postmarktmpls/txt/" 

61 fanout_realm_id: str | None = None 

62 fanout_realm_key: str | None = None 

63 

64 @property 

65 def sqlalchemy_dsn(self) -> str: 

66 return ( 

67 f"mysql+{self.db_driver}://{self.db_user}:{self.db_password}" 

68 f"@{self.db_host}/{self.db_name}?charset=utf8mb4" 

69 ) 

70 

71 @property 

72 def cache_dir(self) -> Path: 

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

74 

75 @property 

76 def attachments_dir(self) -> Path: 

77 return Path(self.data_directory) / "attachments" / self.db_name 

78 

79 @model_validator(mode="after") 

80 def validate_dirs(self) -> Self: 

81 """ 

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

83 and that cache and attachment subdirectories exists 

84 """ 

85 data_directory = Path(self.data_directory) 

86 if not data_directory.is_dir(): 

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

88 sys.exit(m) 

89 

90 for folder in ["cache", "attachments"]: 

91 app_directory = data_directory / folder 

92 app_directory.mkdir(exist_ok=True) 

93 self.test_usable(app_directory) 

94 

95 return self 

96 

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

98 fname = "%s.tmp" % uuid.uuid4().hex 

99 return fname, self.cache_dir / fname 

100 

101 def conn_string(self, db_name=None): 

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

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

104 

105 @classmethod 

106 def test_usable(cls, dir_name): 

107 try: 

108 ts = str(time.time()) 

109 test_file_path = os.path.join(dir_name, "test.txt") 

110 with open(test_file_path, "w") as tw: 

111 tw.write(ts) 

112 with open(test_file_path, "r") as tr: 

113 tr.read() 

114 os.remove(test_file_path) 

115 except IOError as ioe: 

116 sys.exit( 

117 "Test read/write to data directory %s failed with %s" % (dir_name, ioe) 

118 ) 

119 

120 def __iter__(self) -> Generator[Tuple[str, Any], None, None]: 

121 for key, value in self.model_config.items(): 

122 yield (key, value)