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 

12from postrfp.shared.constants import RunMode 

13 

14 

15class Mailer(str, Enum): 

16 postmark = "postmark" 

17 logfile = "logfile" 

18 

19 

20LIVE_SETTINGS_PATH = "/etc/postrfp/postrfp-settings.env" 

21 

22 

23class AppSettings(BaseSettings): 

24 model_config = SettingsConfigDict( 

25 env_prefix="postrfp_", 

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

27 env_file_encoding="utf-8", 

28 ) 

29 

30 run_mode: RunMode = RunMode.production 

31 

32 db_name: str = "postrfp_dev" 

33 db_host: str = "localhost" 

34 db_user: str = "root" 

35 db_password: str = "" 

36 db_brand: str = "mariadb" 

37 db_driver: str = "mariadbconnector" 

38 # db_driver: str = "mysqldb" 

39 crypt_key: str = "" 

40 webapp_hostname: str = "localhost" 

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

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

43 jwt_expiry_minutes: int = 20 # Default to 20 minutes 

44 jwt_refresh_token_expiry_hours: int = 168 # Default to 7 days 

45 

46 max_failed_login_attempts: int = 5 

47 lockout_duration_minutes: int = 20 

48 

49 # Argon2 password hashing parameters 

50 argon2_time_cost: int = 2 

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

52 argon2_parallelism: int = 8 

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

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

55 

56 remote_events_url: str = "DUMMY_LOG_SERVER" 

57 

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

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

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

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

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

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

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

65 app_path_internal: Optional[str] = "/internal" 

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

67 

68 mailer: Mailer = Mailer.logfile 

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

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

71 postmark_api_key: str = "POSTMARK_API_TEST" 

72 postmark_logging_key: str = "POSTMARK_API_TEST" 

73 system_name: str = "PostRFP" 

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

75 

76 # Background task / executor configuration 

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

78 task_executor: str = "inline" 

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

80 dagu_base_url: str = "http://localhost:8080/api/v2/" 

81 # Optional bearer token for authenticating with Dagu API 

82 dagu_api_token: str | None = None 

83 

84 app_base_url: str = "http://localhost:9000" 

85 

86 @property 

87 def sqlalchemy_dsn(self) -> str: 

88 return self._build_dsn(self.db_name) 

89 

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

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

92 return ( 

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

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

95 ) 

96 

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

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

99 return self._build_dsn(name) 

100 

101 @property 

102 def cache_dir(self) -> Path: 

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

104 

105 @property 

106 def attachments_dir(self) -> Path: 

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

108 

109 @model_validator(mode="after") 

110 def validate_dirs(self) -> Self: 

111 """ 

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

113 and that cache and attachment subdirectories exists 

114 """ 

115 data_directory = Path(self.data_directory) 

116 if not data_directory.is_dir(): 

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

118 sys.exit(m) 

119 

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

121 app_directory = data_directory / folder 

122 app_directory.mkdir(exist_ok=True) 

123 self.test_usable(app_directory) 

124 

125 return self 

126 

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

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

129 return fname, self.cache_dir / fname 

130 

131 @classmethod 

132 def test_usable(cls, dir_name): 

133 try: 

134 ts = str(time.time()) 

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

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

137 tw.write(ts) 

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

139 tr.read() 

140 os.remove(test_file_path) 

141 except IOError as ioe: 

142 sys.exit( 

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

144 ) 

145 

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

147 for key, value in self.model_dump().items(): 

148 if value and "password" in key or "key" in key: 

149 yield key, "-----" 

150 else: 

151 yield key, value