Coverage for graphica/gui/plugin_update_flow.py: 70.9%

1"""プラグインの更新: 起動時(1日1回)とヘルプメニュー・環境設定から新しい版を調べ、一覧で知らせて、選んだものを入れ替える。 

2 

3調べる・落とす・入れ替えるのは別スレッド(TaskRunner)。入れ替えたプラグインは次の起動から使われる 

4(読み込んだプラグインをその場で入れ替える仕組みは無い)。 

5""" 

6import datetime 

7import logging 

8import os 

9import zipfile 

10 

11from graphica.core.app_paths import get_user_plugins_dir 

12from graphica.core.i18n import tr 

13from graphica.core.plugin_install import PluginInstallError, install_plugin_zip 

14from graphica.core.plugin_manifest import PluginManifestError 

15from graphica.core.plugin_updates import check_plugin_updates, check_update_zip 

16from graphica.gui import app_settings, notify 

17from graphica.gui.task_runner import TaskRunner 

18from graphica.gui.updater import download_file, update_download_dir 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23def _today(): 

24 return datetime.date.today().isoformat() 

25 

26 

27def _start_check(app, on_succeeded, on_failed): 

28 if getattr(app, '_plugin_update_runner', None) is not None: 28 ↛ 29line 28 didn't jump to line 29 because the condition on line 28 was never true

29 return False 

30 runner = TaskRunner(check_plugin_updates, get_user_plugins_dir(), parent=app) 

31 runner.succeeded.connect(on_succeeded) 

32 runner.failed.connect(on_failed) 

33 app._plugin_update_runner = runner 

34 runner.start() 

35 return True 

36 

37 

38def _finish(app): 

39 app._plugin_update_runner = None 

40 

41 

42def start_startup_check(app): 

43 """1日1回だけ調べ、新しい版があるときだけ知らせる(失敗・無しなら何も出さない)。""" 

44 if app_settings.PLUGIN_UPDATE_LAST_CHECK.read(app.settings) == _today(): 

45 return 

46 

47 def succeeded(updates): 

48 _finish(app) 

49 app_settings.PLUGIN_UPDATE_LAST_CHECK.write(app.settings, _today()) 

50 if updates: 

51 offer_updates(app, updates) 

52 

53 def failed(message): 

54 _finish(app) 

55 logger.info("起動時のプラグインの更新の確認に失敗しました(オフライン等の可能性): %s", message) 

56 

57 _start_check(app, succeeded, failed) 

58 

59 

60def check_now(app): 

61 """メニュー・環境設定の「プラグインの更新を確認」。最新でも失敗でも結果を知らせる。""" 

62 def succeeded(updates): 

63 _finish(app) 

64 app_settings.PLUGIN_UPDATE_LAST_CHECK.write(app.settings, _today()) 

65 if updates: 65 ↛ 66line 65 didn't jump to line 66 because the condition on line 65 was never true

66 offer_updates(app, updates) 

67 else: 

68 notify.information(app, tr("プラグインの更新"), tr("入れているプラグインはすべて最新です。")) 

69 

70 def failed(message): 

71 _finish(app) 

72 logger.warning("プラグインの更新の確認に失敗しました: %s", message) 

73 notify.warning(app, tr("プラグインの更新"), 

74 tr("更新を確認できませんでした。インターネットに接続しているか確かめてください。") + f"\n\n{message}") 

75 

76 if not _start_check(app, succeeded, failed): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 notify.information(app, tr("プラグインの更新"), tr("確認中です。完了までお待ちください。")) 

78 

79 

80def offer_updates(app, updates): 

81 """新しい版の一覧を出し、選んだものを入れ替える。""" 

82 from graphica.gui.dialogs import PluginUpdatesDialog 

83 dialog = PluginUpdatesDialog(updates, app) 

84 if dialog.exec() != dialog.DialogCode.Accepted: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true

85 return 

86 selected = dialog.selected_updates() 

87 if selected: 87 ↛ exitline 87 didn't return from function 'offer_updates' because the condition on line 87 was always true

88 apply_updates(app, selected) 

89 

90 

91def download_and_install(updates, plugins_dir, download_dir, report_progress=None, is_cancelled=None): 

92 """1件ずつ落として確かめ、入れ替える。結果は [(更新の情報, エラーの文 or None)]。TaskRunner 用。""" 

93 results = [] 

94 for number, update in enumerate(updates, start=1): 

95 if is_cancelled is not None and is_cancelled(): 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 break 

97 if report_progress is not None: 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true

98 report_progress(number - 1, len(updates), update['name']) 

99 asset = update['zip'] 

100 path = os.path.join(download_dir, f"{update['folder']}-{update['latest']}.zip") 

101 try: 

102 if download_file(asset['url'], path, asset.get('size', 0), asset.get('sha256', ''), 102 ↛ 104line 102 didn't jump to line 104 because the condition on line 102 was never true

103 is_cancelled=is_cancelled) is None: 

104 break 

105 check_update_zip(path, update['folder']) 

106 install_plugin_zip(path, plugins_dir) 

107 results.append((update, None)) 

108 except (OSError, ValueError, zipfile.BadZipFile, PluginManifestError, PluginInstallError) as e: 

109 # 1件の失敗でほかを止めない(理由は結果に出す) 

110 logger.warning("プラグイン %s の更新に失敗しました", update['name'], exc_info=True) 

111 results.append((update, str(e))) 

112 finally: 

113 if os.path.exists(path): 113 ↛ 94line 113 didn't jump to line 94 because the condition on line 113 was always true

114 try: 

115 os.remove(path) 

116 except OSError: 

117 pass 

118 return results 

119 

120 

121def apply_updates(app, updates): 

122 runner = TaskRunner(download_and_install, updates, get_user_plugins_dir(), update_download_dir(), parent=app) 

123 runner.succeeded.connect(lambda results: _on_installed(app, results)) 

124 runner.failed.connect(lambda message: _on_install_failed(app, message)) 

125 app._plugin_update_runner = runner 

126 app.statusBar().showMessage(tr("プラグインを更新しています...")) 

127 runner.start() 

128 

129 

130def _on_installed(app, results): 

131 _finish(app) 

132 app.statusBar().clearMessage() 

133 done = [update for update, error in results if error is None] 

134 failed = [(update, error) for update, error in results if error is not None] 

135 lines = [] 

136 if done: 136 ↛ 139line 136 didn't jump to line 139 because the condition on line 136 was always true

137 lines.append(tr("次のプラグインを更新しました。Graphica を起動し直すと新しい版が使われます。")) 

138 lines += [f" • {u['name']} {u['current']} → {u['latest']}" for u in done] 

139 if failed: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true

140 if lines: 

141 lines.append("") 

142 lines.append(tr("次のプラグインは更新できませんでした。")) 

143 lines += [f" • {u['name']}: {error}" for u, error in failed] 

144 if not lines: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 return 

146 (notify.warning if failed else notify.information)(app, tr("プラグインの更新"), "\n".join(lines)) 

147 

148 

149def _on_install_failed(app, message): 

150 _finish(app) 

151 app.statusBar().clearMessage() 

152 notify.warning(app, tr("プラグインの更新"), tr("プラグインを更新できませんでした。") + f"\n\n{message}") 

153 

154 

155def stop(app): 

156 """タブを閉じるとき: 動いている確認・更新を止めて待つ(閉じたタブに結果を返さない)。""" 

157 runner = getattr(app, '_plugin_update_runner', None) 

158 if runner is None: 158 ↛ 160line 158 didn't jump to line 160 because the condition on line 158 was always true

159 return 

160 try: 

161 runner.succeeded.disconnect() 

162 runner.failed.disconnect() 

163 except (RuntimeError, TypeError): 

164 pass 

165 runner.requestInterruption() 

166 runner.wait() 

167 runner.deleteLater() 

168 app._plugin_update_runner = None