Coverage for flask/src/rasp_shutter/control/webapi/test/time.py: 21%

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-shutter-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_shutter.control.scheduler 

103 from rasp_shutter.control.webapi.schedule import schedule_queue 

104 

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

106 schedule_queue.put(current_schedule) 

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

108 except Exception as e: 

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

110 

111 current_time = my_lib.time.now() 

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

113 

114 return { 

115 "success": True, 

116 "mock_time": current_time.isoformat(), 

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

118 "advanced_seconds": seconds, 

119 } 

120 

121 

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

123def reset_mock_time(): 

124 """ 

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

126 

127 Returns: 

128 JSON: リセット結果 

129 

130 """ 

131 global _traveler # noqa: PLW0603 

132 

133 # DUMMY_MODE でない場合は拒否 

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

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

136 

137 if _traveler: 

138 _traveler.stop() 

139 _traveler = None 

140 

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

142 

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

144 

145 

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

147def get_current_time(): 

148 """ 

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

150 

151 Returns: 

152 JSON: 現在時刻情報 

153 

154 """ 

155 # DUMMY_MODE でない場合は拒否 

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

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

158 

159 current_time = my_lib.time.now() 

160 

161 return { 

162 "current_time": current_time.isoformat(), 

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

164 "is_mocked": _traveler is not None, 

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

166 }