Coverage for flask/src/rasp_water/control/webapi/test/time.py: 20%

1#!/usr/bin/env python3 

2 

3import datetime 

4import logging 

5import os 

6 

7import my_lib.time 

8import my_lib.webapp.config 

9import time_machine 

10 

11import flask 

12 

13blueprint = flask.Blueprint("rasp-water-test-time", __name__, url_prefix=my_lib.webapp.config.URL_PREFIX) 

14 

15 

16# テスト用の時刻モック状態を保持 

17_traveler = None 

18 

19 

20@blueprint.route("/api/test/time/set/<timestamp>", methods=["POST"]) 

21def set_mock_time(timestamp): 

22 """ 

23 テスト用時刻を設定するAPI 

24 

25 Args: 

26 timestamp: Unix timestamp (秒) またはISO形式の日時文字列 

27 

28 Returns: 

29 JSON: 設定された時刻情報 

30 

31 """ 

32 global _traveler # noqa: PLW0603 

33 

34 # DUMMY_MODE でない場合は拒否 

35 if os.environ.get("DUMMY_MODE", "false") != "true": 

36 return {"error": "Test API is only available in DUMMY_MODE"}, 403 

37 

38 try: 

39 # タイムスタンプの解析 

40 if timestamp.isdigit(): 

41 mock_datetime = datetime.datetime.fromtimestamp(int(timestamp), tz=my_lib.time.get_zoneinfo()) 

42 else: 

43 # ISO形式の解析 

44 mock_datetime = datetime.datetime.fromisoformat(timestamp) 

45 if mock_datetime.tzinfo is None: 

46 mock_datetime = mock_datetime.replace(tzinfo=my_lib.time.get_zoneinfo()) 

47 

48 # 既存のtravelerを停止 

49 if _traveler: 

50 _traveler.stop() 

51 

52 # time_machineを使用して時刻を設定 

53 _traveler = time_machine.travel(mock_datetime) 

54 _traveler.start() 

55 

56 logging.info("Mock time set to: %s", mock_datetime) 

57 

58 return { 

59 "success": True, 

60 "mock_time": mock_datetime.isoformat(), 

61 "unix_timestamp": int(mock_datetime.timestamp()), 

62 } 

63 

64 except (ValueError, TypeError) as e: 

65 return {"error": f"Invalid timestamp format: {e}"}, 400 

66 

67 

68@blueprint.route("/api/test/time/advance/<int:seconds>", methods=["POST"]) 

69def advance_mock_time(seconds): 

70 """ 

71 モック時刻を指定秒数進める 

72 

73 Args: 

74 seconds: 進める秒数 

75 

76 Returns: 

77 JSON: 更新された時刻情報 

78 

79 """ 

80 global _traveler # noqa: PLW0603 

81 

82 # DUMMY_MODE でない場合は拒否 

83 if os.environ.get("DUMMY_MODE", "false") != "true": 

84 return {"error": "Test API is only available in DUMMY_MODE"}, 403 

85 

86 if _traveler is None: 

87 return {"error": "Mock time not set. Use /api/test/time/set first"}, 400 

88 

89 # 現在の時刻を取得して、新しい時刻を計算 

90 current_mock_time = my_lib.time.now() 

91 new_mock_time = current_mock_time + datetime.timedelta(seconds=seconds) 

92 

93 # 既存のtravelerを停止 

94 _traveler.stop() 

95 

96 # 新しい時刻でtravelerを再作成 

97 _traveler = time_machine.travel(new_mock_time) 

98 _traveler.start() 

99 

100 # スケジューラーに現在のスケジュールを再読み込みさせる 

101 try: 

102 import rasp_water.control.scheduler 

103 from rasp_water.control.webapi.schedule import schedule_queue 

104 

105 current_schedule = rasp_water.control.scheduler.schedule_load() 

106 schedule_queue.put(current_schedule) 

107 logging.info("Forced scheduler reload with current schedule") 

108 

109 # スケジューラーの内部時刻同期のため複数回リロード 

110 for i in range(3): 

111 schedule_queue.put(current_schedule) 

112 

113 except Exception as e: 

114 logging.warning("Failed to force scheduler reload: %s", e) 

115 

116 current_time = my_lib.time.now() 

117 logging.info("Mock time advanced to: %s", current_time) 

118 

119 return { 

120 "success": True, 

121 "mock_time": current_time.isoformat(), 

122 "unix_timestamp": int(current_time.timestamp()), 

123 "advanced_seconds": seconds, 

124 } 

125 

126 

127@blueprint.route("/api/test/time/reset", methods=["POST"]) 

128def reset_mock_time(): 

129 """ 

130 モック時刻をリセットして実際の時刻に戻す 

131 

132 Returns: 

133 JSON: リセット結果 

134 

135 """ 

136 global _traveler # noqa: PLW0603 

137 

138 # DUMMY_MODE でない場合は拒否 

139 if os.environ.get("DUMMY_MODE", "false") != "true": 

140 return {"error": "Test API is only available in DUMMY_MODE"}, 403 

141 

142 if _traveler: 

143 _traveler.stop() 

144 _traveler = None 

145 

146 logging.info("Mock time reset to real time") 

147 

148 return {"success": True, "real_time": my_lib.time.now().isoformat()} 

149 

150 

151@blueprint.route("/api/test/time/current", methods=["GET"]) 

152def get_current_time(): 

153 """ 

154 現在の時刻（モック時刻または実時刻）を取得 

155 

156 Returns: 

157 JSON: 現在時刻情報 

158 

159 """ 

160 # DUMMY_MODE でない場合は拒否 

161 if os.environ.get("DUMMY_MODE", "false") != "true": 

162 return {"error": "Test API is only available in DUMMY_MODE"}, 403 

163 

164 current_time = my_lib.time.now() 

165 

166 return { 

167 "current_time": current_time.isoformat(), 

168 "unix_timestamp": int(current_time.timestamp()), 

169 "is_mocked": _traveler is not None, 

170 "mock_time": current_time.isoformat() if _traveler else None, 

171 }