1#!/usr/bin/env python3

2# Copyright 2023-2024 Canonical Ltd.

3# All rights reserved.

4#

5# Written by:

6# Patrick Liu <patrick.liu@canonical.com>

7# Pierre Equoy <pierre.equoy@canonical.com>

8#

9# Checkbox is free software: you can redistribute it and/or modify

10# it under the terms of the GNU General Public License version 3,

11# as published by the Free Software Foundation.

12#

13# Checkbox is distributed in the hope that it will be useful,

14# but WITHOUT ANY WARRANTY; without even the implied warranty of

15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

16# GNU General Public License for more details.

17#

18# You should have received a copy of the GNU General Public License

19# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.

20

21 import argparse

22 from pathlib import Path

23 import json

24 import os

25 import sys

26 import time

27

28 from checkbox_support.snap_utils.snapd import Snapd

29 from checkbox_support.snap_utils.snapd import AsyncException

30 from checkbox_support.snap_utils.snapd import SnapdRequestError

31

32

33 def guess_snaps() -> list:

34 """

35 Guess the names of the kernel, snapd and gadget snaps from installed snaps

36 on the system.

37

38 :return: a list of snap names that are either kernel, snapd or gadget snaps

39 :rtype: list

40 """

41 snaps = [

42 snap["name"]

43 for snap in Snapd().list()

44 if snap["type"] in ("kernel", "gadget", "snapd")

45 ]

46 return snaps

47

48

49 def get_snaps_base_rev() -> dict:

50 """

51 Retrieve the name and the base revision of each snap originally installed

52 on the system.

53

54 :return: a dict containing the snap names and their base revisions

55 :rtype: dict

56 """

57 seed_snaps_dir = Path("/var/lib/snapd/seed/snaps/")

58 base_snaps = seed_snaps_dir.glob("*.snap")

59 base_rev_info = {}

60 for snap_path in base_snaps:

61 snap, rev = snap_path.stem.rsplit("_", maxsplit=1)

62 base_rev_info[snap] = rev

63 return base_rev_info

64

65

66 class SnapInfo:

67 def __init__(self, name):

68 snap = Snapd().list(name)

69 self.name = snap["name"]

70 self.type = snap["type"]

71 self.tracking_channel = snap["tracking-channel"]

72 self.installed_revision = snap["revision"]

73 self.tracking_prefix = (

74 (self.tracking_channel.split("/")[0] + "/")

75 if "/" in self.tracking_channel

76 else ""

77 )

78 self.base_revision = get_snaps_base_rev().get(name, "")

79

80 revisions = {}

81 for item in Snapd().find(name, exact=True):

82 for channel, info in item["channels"].items():

83 revisions[channel] = info["revision"]

84

85 self.stable_revision = revisions.get(

86 "{}stable".format(self.tracking_prefix), ""

87 )

88 self.candidate_revision = revisions.get(

89 "{}candidate".format(self.tracking_prefix), ""

90 )

91 self.beta_revision = revisions.get(

92 "{}beta".format(self.tracking_prefix), ""

93 )

94 self.edge_revision = revisions.get(

95 "{}edge".format(self.tracking_prefix), ""

96 )

97

98 def print_as_resource(self):

99 print("name: {}".format(self.name))

100 print("type: {}".format(self.type))

101 print("tracking: {}".format(self.tracking_channel))

102 print("base_rev: {}".format(self.base_revision))

103 print("stable_rev: {}".format(self.stable_revision))

104 print("candidate_rev: {}".format(self.candidate_revision))

105 print("beta_rev: {}".format(self.beta_revision))

106 print("edge_rev: {}".format(self.edge_revision))

107 print("original_installed_rev: {}".format(self.installed_revision))

108 print()

109

110

111 def print_resource_info():

112 for snap in guess_snaps():

113 SnapInfo(snap).print_as_resource()

114

115

116 def save_change_info(path, data):

117 with open(path, "w") as file:

118 json.dump(data, file)

119

120

121 def load_change_info(path):

122 try:

123 with open(path, "r") as file:

124 data = json.load(file)

125 except FileNotFoundError:

126 error_msg = "File not found: {}. Did the previous job run as expected?"

127 raise SystemExit(error_msg.format(path))

128 return data

129

130

131 class SnapRefreshRevert:

132 def __init__(self, name, revision, info_path, timeout):

133 self.timeout = timeout

134 self.snapd = Snapd(task_timeout=self.timeout, verbose=True)

135 self.snap_info = SnapInfo(name)

136 self.path = info_path

137 self.revision = revision

138 self.name = name

139

140 def snap_refresh(self):

141 data = {}

142 original_revision = self.snap_info.installed_revision

143 if original_revision == self.revision:

144 error_msg = "Trying to refresh to the same revision ({})!"

145 raise SystemExit(error_msg.format(self.revision))

146 data["name"] = self.name

147 data["original_revision"] = original_revision

148 data["destination_revision"] = self.revision

149 print(

150 "Refreshing snap {} from revision {} to {}".format(

151 self.name, original_revision, self.revision

152 )

153 )

154 try:

155 response = self.snapd.refresh(

156 self.name,

157 channel=self.snap_info.tracking_channel,

158 revision=self.revision,

159 )

160 except (SnapdRequestError, AsyncException) as exc:

161 checkbox_session_dir = os.getenv("PLAINBOX_SESSION_SHARE")

162 if checkbox_session_dir: 162 ↛ 176 line 162 didn't jump to line 176 because the condition on line 162 was always true

163 result = {

164 "outcome": "fail",

165 "comments": (

166 "Marking the test as failed because it raised"

167 " the following:"

168 )

169 + str(exc),

170 }

171 result_filename = os.path.join(

172 checkbox_session_dir, "__result"

173 )

174 with open(result_filename, "wt") as result_f:

175 json.dump(result, result_f)

176 raise

177 data["change_id"] = response["change"]

178 print(

179 "Snap operation finished. "

180 "See `snap change {}` for more info".format(response["change"])

181 )

182 save_change_info(self.path, data)

183 print("Waiting for reboot...")

184

185 def snap_revert(self):

186 data = load_change_info(self.path)

187 original_rev = data["original_revision"]

188 destination_rev = data["destination_revision"]

189 print(

190 "Reverting snap {} from revision {} to {}".format(

191 self.name, destination_rev, original_rev

192 )

193 )

194 try:

195 response = self.snapd.revert(self.name)

196 except (SnapdRequestError, AsyncException) as exc:

197 checkbox_session_dir = os.getenv("PLAINBOX_SESSION_SHARE")

198 if checkbox_session_dir: 198 ↛ 208 line 198 didn't jump to line 208 because the condition on line 198 was always true

199 result = {

200 "outcome": "fail",

201 "comments": exc.message,

202 }

203 result_filename = os.path.join(

204 checkbox_session_dir, "__result"

205 )

206 with open(result_filename, "wt") as result_f:

207 json.dump(result, result_f)

208 raise

209 data["change_id"] = response["change"]

210 print(

211 "Snap operation finished. "

212 "See `snap change {}` for more info".format(response["change"])

213 )

214 save_change_info(self.path, data)

215 print("Waiting for reboot...")

216

217 def wait_for_snap_change(self, change_id, type):

218 start_time = time.time()

219 while True:

220 result = self.snapd.change(str(change_id))

221 if result == "Done":

222 print("{} snap {} complete".format(self.name, type))

223 return

224 elif result == "Error":

225 tasks = self.snapd.tasks(str(change_id))

226 for task in tasks:

227 print(

228 "{} | {} | {}".format(

229 task["id"], task["status"], task["summary"]

230 )

231 )

232 if task.get("log"): 232 ↛ 226 line 232 didn't jump to line 226 because the condition on line 232 was always true

233 for log in task["log"]:

234 print("\t {}".format(log))

235 raise SystemExit(

236 "Error during snap {} {}.".format(self.name, type)

237 )

238

239 current_time = time.time()

240 if current_time - start_time >= self.timeout:

241 raise SystemExit(

242 "{} snap {} did not complete within {} seconds".format(

243 self.name, type, self.timeout

244 )

245 )

246 print(

247 "Waiting for {} snap {} to be done...".format(self.name, type)

248 )

249 print("Trying again in 10 seconds...")

250 time.sleep(10)

251

252 def verify(self, type):

253 print("Beginning verify...")

254 if type not in ("refresh", "revert"):

255 msg = (

256 "'{}' verification unknown. Can be either 'refresh' "

257 "or 'revert'."

258 ).format(type)

259 raise SystemExit(msg)

260 data = load_change_info(self.path)

261 id = data["change_id"]

262 self.wait_for_snap_change(id, type)

263 print("Checking {} status for snap {}...".format(type, self.name))

264

265 current_rev = self.snapd.list(self.name)["revision"]

266 if type == "refresh":

267 tested_rev = data["destination_revision"]

268 else:

269 tested_rev = data["original_revision"]

270 if current_rev != tested_rev:

271 msg = (

272 "Current revision ({}) is different from expected revision "

273 "({})"

274 ).format(current_rev, tested_rev)

275 raise SystemExit(msg)

276 else:

277 print(

278 "PASS: current revision ({}) matches the expected "

279 "revision".format(current_rev)

280 )

281

282

283 def main(args):

284 parser = argparse.ArgumentParser()

285 parser.add_argument(

286 "name", nargs="?", default="", help="Name of the snap to act upon"

287 )

288 parser.add_argument(

289 "--resource",

290 action="store_true",

291 help="Gather information about kernel, snapd and gadget snaps",

292 )

293 parser.add_argument(

294 "--refresh",

295 action="store_true",

296 help="Refresh the given snap",

297 )

298 parser.add_argument(

299 "--verify-refresh",

300 action="store_true",

301 help="Verify revision after refreshing the given snap",

302 )

303 parser.add_argument(

304 "--revert",

305 action="store_true",

306 help="Revert the given snap",

307 )

308 parser.add_argument(

309 "--verify-revert",

310 action="store_true",

311 help="Verify revision after reverting the given snap",

312 )

313 parser.add_argument(

314 "--info-path",

315 help="Path to the information file",

316 )

317 parser.add_argument(

318 "--revision",

319 help="Revision to refresh to",

320 )

321 parser.add_argument(

322 "--timeout",

323 default=600,

324 help="Timeout for each task, in seconds (default: %(default)s))",

325 )

326

327 args = parser.parse_args(args)

328

329 if args.resource:

330 print_resource_info()

331 else:

332 test = SnapRefreshRevert(

333 name=args.name,

334 info_path=args.info_path,

335 revision=args.revision,

336 timeout=args.timeout,

337 )

338 if args.refresh:

339 test.snap_refresh()

340 if args.verify_refresh:

341 test.verify("refresh")

342 if args.revert:

343 test.snap_revert()

344 if args.verify_revert:

345 test.verify("revert")

346

347

348 if __name__ == "__main__":

349 sys.exit(main(sys.argv[1:]))