Coverage for frappe_manager / output_manager / flags.py: 77%

1""" 

2Feature flags for gradual output architecture refactoring rollout. 

3 

4These flags allow incremental migration and provide escape hatches if issues arise in production. 

5""" 

6 

7import os 

8from typing import Literal 

9 

10 

11class OutputRefactoringFlags: 

12 """ 

13 Feature flags for output architecture refactoring. 

14 

15 Control via environment variables to enable gradual rollout and provide 

16 rollback capability if issues arise. 

17 """ 

18 

19 @staticmethod 

20 def use_context_managers() -> bool: 

21 """ 

22 Enable context manager pattern for spinner lifecycle. 

23 

24 When enabled, new code uses `with spinner(output, "text"):` pattern 

25 instead of manual start()/stop() calls. 

26 

27 Environment Variable: FM_USE_SPINNER_CONTEXT 

28 Values: "true" to enable, "false" to disable (default: false) 

29 

30 Returns: 

31 True if context managers should be used 

32 """ 

33 return os.getenv("FM_USE_SPINNER_CONTEXT", "false").lower() == "true" 

34 

35 @staticmethod 

36 def enforce_output_handler() -> bool: 

37 """ 

38 Block direct richprint access in favor of OutputHandler. 

39 

40 When enabled, direct imports of richprint will trigger warnings or errors. 

41 

42 Environment Variable: FM_ENFORCE_OUTPUT_HANDLER 

43 Values: "true" to enable, "false" to disable (default: false) 

44 

45 Returns: 

46 True if OutputHandler should be enforced 

47 """ 

48 return os.getenv("FM_ENFORCE_OUTPUT_HANDLER", "false").lower() == "true" 

49 

50 @staticmethod 

51 def strict_mode() -> bool: 

52 """ 

53 Raise errors on deprecated output patterns. 

54 

55 When enabled, use of deprecated patterns (direct richprint, unprotected start/stop) 

56 will raise errors instead of just warnings. 

57 

58 Environment Variable: FM_STRICT_OUTPUT 

59 Values: "true" to enable, "false" to disable (default: false) 

60 

61 Returns: 

62 True if strict mode is enabled 

63 """ 

64 return os.getenv("FM_STRICT_OUTPUT", "false").lower() == "true" 

65 

66 @staticmethod 

67 def stream_separation_mode() -> Literal["legacy", "transition", "strict"]: 

68 """ 

69 Control stream separation behavior (stdout vs stderr). 

70 

71 - legacy: All output to stderr (current behavior) 

72 - transition: Gradual migration, both patterns work 

73 - strict: Enforce stdout for data, stderr for diagnostics 

74 

75 Environment Variable: FM_STREAM_SEPARATION 

76 Values: "legacy" | "transition" | "strict" (default: legacy) 

77 

78 Returns: 

79 Current stream separation mode 

80 """ 

81 mode = os.getenv("FM_STREAM_SEPARATION", "legacy").lower() 

82 if mode in ("legacy", "transition", "strict"): 

83 return mode # type: ignore 

84 return "legacy" 

85 

86 @staticmethod 

87 def skip_migrations() -> bool: 

88 """ 

89 Skip migration execution (escape hatch). 

90 

91 CRITICAL ESCAPE HATCH: If migrations break the CLI, set this to true 

92 to bypass migration execution and recover CLI functionality. 

93 

94 Environment Variable: FM_SKIP_MIGRATIONS 

95 Values: "true" to skip, "false" to run normally (default: false) 

96 

97 Returns: 

98 True if migrations should be skipped 

99 """ 

100 return os.getenv("FM_SKIP_MIGRATIONS", "false").lower() == "true" 

101 

102 @staticmethod 

103 def enable_migration_backups() -> bool: 

104 """ 

105 Enable automatic backups before migrations. 

106 

107 When enabled, creates backup before running migrations so rollback 

108 is possible if migrations fail. 

109 

110 Environment Variable: FM_MIGRATION_BACKUPS 

111 Values: "true" to enable, "false" to disable (default: true) 

112 

113 Returns: 

114 True if migration backups should be created 

115 """ 

116 return os.getenv("FM_MIGRATION_BACKUPS", "true").lower() == "true" 

117 

118 @staticmethod 

119 def log_output_calls() -> bool: 

120 """ 

121 Log all output handler calls for debugging. 

122 

123 When enabled, logs every call to output.start(), output.stop(), output.print() 

124 etc. Useful for debugging output flow and identifying issues. 

125 

126 Environment Variable: FM_LOG_OUTPUT_CALLS 

127 Values: "true" to enable, "false" to disable (default: false) 

128 

129 Returns: 

130 True if output calls should be logged 

131 """ 

132 return os.getenv("FM_LOG_OUTPUT_CALLS", "false").lower() == "true" 

133 

134 @staticmethod 

135 def get_all_flags() -> dict[str, bool | str]: 

136 """ 

137 Get current state of all feature flags. 

138 

139 Useful for diagnostics and debugging. 

140 

141 Returns: 

142 Dictionary of all flag names and their current values 

143 """ 

144 return { 

145 "use_context_managers": OutputRefactoringFlags.use_context_managers(), 

146 "enforce_output_handler": OutputRefactoringFlags.enforce_output_handler(), 

147 "strict_mode": OutputRefactoringFlags.strict_mode(), 

148 "stream_separation_mode": OutputRefactoringFlags.stream_separation_mode(), 

149 "skip_migrations": OutputRefactoringFlags.skip_migrations(), 

150 "enable_migration_backups": OutputRefactoringFlags.enable_migration_backups(), 

151 "log_output_calls": OutputRefactoringFlags.log_output_calls(), 

152 }