1#!/usr/bin/env python3

2"""

3 Reboot networking, wait for a while and check if it's possible to send

4 some pings

5"""

6 import sys

7 import os

8 import time

9 import threading

10 import logging

11 import logging.handlers

12

13 from subprocess import check_output, check_call, CalledProcessError, STDOUT

14 from argparse import ArgumentParser

15

16 try:

17 import gi

18

19 gi.require_version("GLib", "2.0")

20 gi.require_version("Gtk", "3.0")

21 from gi.repository import Gtk, GObject, GLib

22

23 GLib.threads_init()

24 GObject.threads_init()

25 gtk_found = True

26 except (ImportError, RuntimeError):

27 gtk_found = False

28

29

30 class PingError(Exception):

31 def __init__(self, address, reason):

32 self.address = address

33 self.reason = reason

34

35

36 def main():

37 args = parse_args()

38

39 # Verify that script is run as root

40 if os.getuid():

41 sys.stderr.write(

42 "This script needs superuser permissions to run correctly\n"

43 )

44 return 1

45

46 configure_logging(args.log_level, args.output)

47

48 # Select interface based on graphich capabilities available

49 if "DISPLAY" in os.environ and gtk_found:

50 factory = GtkApplication

51 else:

52 factory = CliApplication

53

54 app = factory(args.address, args.times)

55 return app.run()

56

57

58 class Application:

59 """

60 Network restart application

61 """

62

63 def __init__(self, address, times):

64 self.address = address

65 self.times = times

66 self.return_code = 0

67

68 def run(self):

69 """

70 Restart networking as many times as requested

71 and use ping to verify

72 """

73 networking = Networking(self.address)

74 logging.info("Initial connectivity check")

75 success = ping(self.address)

76 if not success:

77 raise PingError(self.address, "Some interface is down")

78

79 for i in range(self.times):

80 if self.return_code:

81 break

82 if self.progress_cb:

83 fraction = float(i) / self.times

84 self.progress_cb(fraction)

85 logging.info("Iteration {0}/{1}...".format(i + 1, self.times))

86 networking.restart()

87 else:

88 if self.progress_cb:

89 self.progress_cb(1.0)

90 logging.info("Test successful")

91

92 return self.return_code

93

94

95 class CliApplication(Application):

96 progress_cb = None

97

98

99 class GtkApplication(Application):

100 def __init__(self, address, times):

101 Application.__init__(self, address, times)

102

103 dialog = Gtk.Dialog(

104 title="Network restart",

105 buttons=(Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL),

106 )

107 dialog.set_default_size(300, 100)

108

109 alignment = Gtk.Alignment()

110 alignment.set(0.5, 0.5, 1.0, 0.1)

111 alignment.set_padding(10, 10, 10, 10)

112 progress_bar = Gtk.ProgressBar()

113 progress_bar.set_show_text(True)

114 alignment.add(progress_bar)

115

116 content_area = dialog.get_content_area()

117 content_area.pack_start(alignment, expand=True, fill=True, padding=0)

118

119 dialog.connect("response", self.response_cb)

120 dialog.show_all()

121

122 # Add new logger handler to write info logs to progress bar

123 logger = logging.getLogger()

124 stream = ProgressBarWriter(progress_bar)

125 formatter = logging.Formatter("%(message)s")

126 log_handler = logging.StreamHandler(stream)

127 log_handler.setLevel(logging.INFO)

128 log_handler.setFormatter(formatter)

129 logger.addHandler(log_handler)

130

131 self.return_code = 0

132 self.dialog = dialog

133 self.progress_bar = progress_bar

134 self.progress_log_handler = log_handler

135

136 def response_cb(self, dialog, response_id):

137 """

138 Cancel test case execution

139 when cancel or close button are closed

140 """

141 self.return_code = response_id

142 logging.info("Test cancelled")

143 Gtk.main_quit()

144

145 def progress_cb(self, fraction):

146 """

147 Update progress bar

148 """

149 GLib.idle_add(self.progress_bar.set_fraction, fraction)

150

151 def thread_target(self):

152 """

153 Run test case in a separate thread

154 """

155 try:

156 Application.run(self)

157 except PingError as exception:

158 logging.error(

159 "Failed to ping {0!r}\n{1}".format(

160 exception.address, exception.reason

161 )

162 )

163 self.return_code = -1

164 except CalledProcessError:

165 self.return_code = -1

166 finally:

167 Gtk.main_quit()

168

169 def run(self):

170 """

171 Launch test case and gtk mainloop

172 """

173 thread = threading.Thread(target=self.thread_target)

174 thread.daemon = True

175 thread.start()

176 Gtk.main()

177 return self.return_code

178

179

180 class ProgressBarWriter:

181 """

182 Write logs to a progress bar

183 """

184

185 def __init__(self, progressbar):

186 self.progressbar = progressbar

187

188 def write(self, message):

189 if message == "\n":

190 return

191 GLib.idle_add(self.progressbar.set_text, message)

192

193

194 def ping(address):

195 """

196 Send ping to a given address

197 """

198 logging.info("Pinging {0!r}...".format(address))

199 try:

200 check_call(

201 "ping -c 1 -w 5 {0}".format(address),

202 stdout=open(os.devnull, "w"),

203 stderr=STDOUT,

204 shell=True,

205 )

206 except CalledProcessError:

207 return False

208

209 return True

210

211

212 class Networking:

213 """

214 Networking abstraction to start/stop all interfaces

215 """

216

217 def __init__(self, address):

218 self.address = address

219 self.interfaces = self._get_interfaces()

220

221 def _get_interfaces(self):

222 """

223 Get all network interfaces

224 """

225 output = check_output(["/sbin/ifconfig", "-s", "-a"])

226 lines = output.splitlines()[1:]

227 interfaces = [

228 interface

229 for interface in [line.split()[0] for line in lines]

230 if interface != "lo"

231 ]

232 return interfaces

233

234 def restart(self):

235 """

236 Restart networking

237 """

238 self._stop()

239 self._start()

240

241 def _start(self):

242 """

243 Start networking

244 """

245 logging.info("Bringing all interfaces up...")

246 for interface in self.interfaces:

247 try:

248 check_output(["/sbin/ifconfig", interface, "up"])

249 except CalledProcessError:

250 logging.error(

251 "Unable to bring up interface {0!r}".format(interface)

252 )

253 raise

254

255 logging.info("Starting network manager...")

256 try:

257 check_output(["/sbin/start", "network-manager"])

258 except CalledProcessError:

259 logging.error("Unable to start network manager")

260 raise

261

262 # Verify that network interface is up

263 for timeout in [2, 4, 8, 16, 32, 64]:

264 logging.debug("Waiting ({0} seconds)...".format(timeout))

265 time.sleep(timeout)

266 success = ping(self.address)

267 if success:

268 break

269 else:

270 raise PingError(self.address, "Some interface is still down")

271

272 def _stop(self):

273 """

274 Stop network manager

275 """

276 logging.info("Stopping network manager...")

277 try:

278 check_output(["/sbin/stop", "network-manager"])

279 except CalledProcessError:

280 logging.error("Unable to stop network manager")

281 raise

282

283 logging.info("Bringing all interfaces down...")

284 for interface in self.interfaces:

285 try:

286 check_output(["/sbin/ifconfig", interface, "down"])

287 except CalledProcessError:

288 logging.error(

289 "Unable to bring down interface {0!r}".format(interface)

290 )

291 raise

292

293 # Verify that network interface is down

294 for timeout in [2, 4, 8]:

295 logging.debug("Waiting ({0} seconds)...".format(timeout))

296 time.sleep(timeout)

297 success = ping(self.address)

298 if not success:

299 break

300 else:

301 raise PingError(self.address, "Some interface is still up")

302

303

304 def parse_args():

305 """

306 Parse command line options

307 """

308 parser = ArgumentParser(

309 "Reboot networking interface and verify that is up again afterwards"

310 )

311 parser.add_argument(

312 "-a",

313 "--address",

314 default="ubuntu.com",

315 help=(

316 "Address to ping to verify that network connection is up "

317 "('%(default)s' by default)"

318 ),

319 )

320 parser.add_argument(

321 "-o",

322 "--output",

323 default="/var/log",

324 help="The path to the log directory. \

325 Default is /var/log",

326 )

327 parser.add_argument(

328 "-t",

329 "--times",

330 type=int,

331 default=1,

332 help=(

333 "Number of times that the network interface has to be restarted "

334 "(%(default)s by default)"

335 ),

336 )

337 log_levels = ["notset", "debug", "info", "warning", "error", "critical"]

338 parser.add_argument(

339 "--log-level",

340 dest="log_level_str",

341 default="notset",

342 choices=log_levels,

343 help=(

344 "Log level. "

345 "One of {0} or {1} (%(default)s by default)".format(

346 ", ".join(log_levels[:-1]), log_levels[-1]

347 )

348 ),

349 )

350 args = parser.parse_args()

351 args.log_level = getattr(logging, args.log_level_str.upper())

352 return args

353

354

355 def configure_logging(log_level, output):

356 """

357 Configure logging

358 """

359 logger = logging.getLogger()

360 logger.setLevel(logging.DEBUG)

361

362 # Log to sys.stderr using log level passed through command line

363 if log_level != logging.NOTSET:

364 log_handler = logging.StreamHandler()

365 formatter = logging.Formatter("%(levelname)-8s %(message)s")

366 log_handler.setFormatter(formatter)

367 log_handler.setLevel(log_level)

368 logger.addHandler(log_handler)

369

370 # Log to rotating file using DEBUG log level

371 log_filename = os.path.join(

372 output,

373 "{0}.log".format(os.path.splitext(os.path.basename(__file__))[0]),

374 )

375 rollover = os.path.exists(log_filename)

376 log_handler = logging.handlers.RotatingFileHandler(

377 log_filename, mode="a+", backupCount=3

378 )

379 formatter = logging.Formatter("%(asctime)s %(levelname)-8s %(message)s")

380 log_handler.setFormatter(formatter)

381 log_handler.setLevel(logging.DEBUG)

382 logger.addHandler(log_handler)

383 if rollover:

384 log_handler.doRollover()

385

386

387 if __name__ == "__main__":

388 try:

389 sys.exit(main())

390 except PingError as exception:

391 logging.error(

392 "Failed to ping {0!r}\n{1}".format(

393 exception.address, exception.reason

394 )

395 )

396 sys.exit(1)