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

1#!/usr/bin/env python3 

2 

3import datetime 

4import logging 

5 

6import flask 

7import my_lib.time 

8import my_lib.webapp.config 

9import time_machine 

10 

11import rasp_shutter.util 

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"]) 

21@rasp_shutter.util.require_dummy_mode 

22def set_mock_time(timestamp): 

23 """ 

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

25 

26 Args: 

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

28 

29 Returns: 

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

31 

32 """ 

33 global _traveler 

34 

35 try: 

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

37 if timestamp.isdigit(): 

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

39 else: 

40 # ISO形式の解析 

41 mock_datetime = datetime.datetime.fromisoformat(timestamp) 

42 if mock_datetime.tzinfo is None: 

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

44 

45 # 既存のtravelerを停止 

46 if _traveler: 

47 _traveler.stop() 

48 

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

50 _traveler = time_machine.travel(mock_datetime) 

51 _traveler.start() 

52 

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

54 

55 return { 

56 "success": True, 

57 "mock_time": mock_datetime.isoformat(), 

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

59 } 

60 

61 except (ValueError, TypeError) as e: 

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

63 

64 

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

66@rasp_shutter.util.require_dummy_mode 

67def advance_mock_time(seconds): 

68 """ 

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

70 

71 Args: 

72 seconds: 進める秒数 

73 

74 Returns: 

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

76 

77 """ 

78 global _traveler 

79 

80 if _traveler is None: 

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

82 

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

84 current_mock_time = my_lib.time.now() 

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

86 

87 # 既存のtravelerを停止 

88 _traveler.stop() 

89 

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

91 _traveler = time_machine.travel(new_mock_time) 

92 _traveler.start() 

93 

94 # NOTE: スケジュールの強制リロードは行わない。 

95 # 時刻を進めた後にスケジュールをリロードすると、Python schedule ライブラリが 

96 # 既に過ぎた時刻のジョブを「明日」にスケジュールしてしまう。 

97 # 既存のジョブはそのまま維持し、scheduler.run_pending() で評価させる。 

98 

99 current_time = my_lib.time.now() 

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

101 

102 return { 

103 "success": True, 

104 "mock_time": current_time.isoformat(), 

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

106 "advanced_seconds": seconds, 

107 } 

108 

109 

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

111@rasp_shutter.util.require_dummy_mode 

112def reset_mock_time(): 

113 """ 

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

115 

116 Returns: 

117 JSON: リセット結果 

118 

119 """ 

120 global _traveler 

121 

122 if _traveler: 

123 _traveler.stop() 

124 _traveler = None 

125 

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

127 

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

129 

130 

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

132@rasp_shutter.util.require_dummy_mode 

133def get_current_time(): 

134 """ 

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

136 

137 Returns: 

138 JSON: 現在時刻情報 

139 

140 """ 

141 current_time = my_lib.time.now() 

142 

143 return { 

144 "current_time": current_time.isoformat(), 

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

146 "is_mocked": _traveler is not None, 

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

148 }