1 import os, sys, logging 2 from autotest_lib.client.bin import test 3 from autotest_lib.client.common_lib import error 4 5 6 class selftest(test.test): 7 version = 1 8 9 def setup(self): 10 name = self.job.resultdir + '/sequence' 11 if (not os.path.exists(name)): 12 fd = file(name, 'w') 13 fd.write('0') 14 fd.close() 15 16 def __mark(self, checkpoint): 17 name = self.job.resultdir + '/sequence' 18 fd = file(name, 'r') 19 current = int(fd.readline()) 20 fd.close() 21 22 current += 1 23 fd = file(name + '.new', 'w') 24 fd.write('%d' % current) 25 fd.close() 26 27 os.rename(name + '.new', name) 28 29 logging.debug("checkpoint %d %d", current, checkpoint) 30 31 if (current != checkpoint): 32 raise error.JobError("selftest: sequence was " + 33 "%d when %d expected" % (current, checkpoint)) 34 35 def __throw(self): 36 __does_not_exist = __does_not_exist_either 37 38 def __print(self, msg): 39 sys.stdout.write(msg) 40 41 def __warn(self, msg): 42 sys.stderr.write(msg) 43 44 def execute(self, cmd, *args): 45 if cmd == 'mark': 46 self.__mark(*args) 47 elif cmd == 'throw': 48 self.__throw(*args) 49 elif cmd == 'print': 50 self.__print(*args) 51 elif cmd == 'warn': 52 self.__warn(*args) 53