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

1import os 

2import sys 

3import time 

4import uuid 

5from enum import Enum 

6from pathlib import Path 

7from typing import Tuple, Any, Generator, Self, Optional 

8 

9from pydantic_settings import SettingsConfigDict, BaseSettings 

10from pydantic import EmailStr, model_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/postrfp-settings.env" 

25 

26 

27class AppSettings(BaseSettings): 

28 model_config = SettingsConfigDict( 

29 env_prefix="postrfp_", 

30 env_file=(LIVE_SETTINGS_PATH, ".env"), 

31 env_file_encoding="utf-8", 

32 ) 

33 

34 run_mode: RunMode = RunMode.production 

35 

36 db_name: str = "postrfp_dev" 

37 db_host: str = "localhost" 

38 db_user: str = "root" 

39 db_password: str = "" 

40 db_brand: str = "mariadb" 

41 db_driver: str = "mariadbconnector" 

42 # db_driver: str = "mysqldb" 

43 crypt_key: str = "" 

44 webapp_hostname: str = "localhost" 

45 data_directory: str = "./.postrfp-data/" 

46 log_directory: str = "./.postrfp-logs/" 

47 jwt_expiry_minutes: int = 20 # Default to 20 minutes 

48 jwt_refresh_token_expiry_hours: int = 168 # Default to 7 days 

49 

50 max_failed_login_attempts: int = 5 

51 lockout_duration_minutes: int = 20 

52 

53 # Argon2 password hashing parameters 

54 argon2_time_cost: int = 2 

55 argon2_memory_cost: int = 102400 # 100 MB (in KiB) 

56 argon2_parallelism: int = 8 

57 argon2_salt_bytes: int = 16 # Optional: Override default salt length 

58 argon2_hash_bytes: int = 16 # Optional: Override default hash length 

59 

60 remote_events_url: str = "DUMMY_LOG_SERVER" 

61 

62 # WSGI Application Mount Paths - set to None to disable mounting 

63 app_path_auth: Optional[str] = "/auth" 

64 app_path_hooks: Optional[str] = "/hooks" 

65 app_path_buyer: Optional[str] = "/buyer" 

66 app_path_vendor: Optional[str] = "/vendor" 

67 app_path_fsm: Optional[str] = "/fsm" 

68 app_path_ref: Optional[str] = "/ref" 

69 app_path_openapi: Optional[str] = "/openapi" # dev only 

70 

71 mailer: Mailer = Mailer.logfile 

72 email_from_address: EmailStr = "notifications@postrp.com" 

73 email_to_override: EmailStr = "test@blackhole.postmarkapp.com" 

74 postmark_api_key: str = "POSTMARK_API_TEST" 

75 postmark_logging_key: str = "POSTMARK_API_TEST" 

76 system_name: str = "PostRFP" 

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

78 

79 # Background task / executor configuration 

80 # Accepts values: "inline" or "dagu". Defaults to inline for tests/dev. 

81 task_executor: str = "inline" 

82 # Base URL for Dagu workflow engine (only used when task_executor == "dagu") 

83 dagu_base_url: str | None = None 

84 # Optional bearer token for authenticating with Dagu API 

85 dagu_api_token: str | None = None 

86 

87 @property 

88 def sqlalchemy_dsn(self) -> str: 

89 return self._build_dsn(self.db_name) 

90 

91 def _build_dsn(self, db_name: str) -> str: 

92 # mariadb+mariadbconnector://<user>:<password>@<host>[:<port>]/<dbname> 

93 return ( 

94 f"{self.db_brand}+{self.db_driver}://{self.db_user}:{self.db_password}" 

95 f"@{self.db_host}/{db_name}" 

96 ) 

97 

98 def conn_string(self, db_name: Optional[str] = None) -> str: 

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

100 return self._build_dsn(name) 

101 

102 @property 

103 def cache_dir(self) -> Path: 

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

105 

106 @property 

107 def attachments_dir(self) -> Path: 

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

109 

110 @model_validator(mode="after") 

111 def validate_dirs(self) -> Self: 

112 """ 

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

114 and that cache and attachment subdirectories exists 

115 """ 

116 data_directory = Path(self.data_directory) 

117 if not data_directory.is_dir(): 

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

119 sys.exit(m) 

120 

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

122 app_directory = data_directory / folder 

123 app_directory.mkdir(exist_ok=True) 

124 self.test_usable(app_directory) 

125 

126 return self 

127 

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

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

130 return fname, self.cache_dir / fname 

131 

132 @classmethod 

133 def test_usable(cls, dir_name): 

134 try: 

135 ts = str(time.time()) 

136 test_file_path = os.path.join(dir_name, f"proc_{os.getpid()}_test.txt") 

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

138 tw.write(ts) 

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

140 tr.read() 

141 os.remove(test_file_path) 

142 except IOError as ioe: 

143 sys.exit( 

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

145 ) 

146 

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

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

149 yield (key, value)