xstc.py

xml开源解析代码.版本为libxml2-2.6.29,可支持GB3212.网络消息发送XML时很有用.

		schema = None
		filePath = self.fileName
		# os.path.join(options.baseDir, self.fileName)
		valid = 0
		try:
			#
			# Parse the schema.
			#
			self.debugMsg("loading schema: %s" % filePath)
			schema = parseSchema(filePath)
			self.debugMsg("after loading schema")						
			if schema is None:
				self.debugMsg("schema is None")
				self.debugMsg("checking for IO errors...")
				if self.isIOError(file, "schema"):
					return
			self.debugMsg("checking schema result")
			if (schema is None and self.val) or (schema is not None and self.val == 0):
				self.debugMsg("schema result is BAD")
				if (schema == None):
					self.fail(msgSchemaNotValidButShould)
				else:
					self.fail(msgSchemaValidButShouldNot)
			else:
				self.debugMsg("schema result is OK")
		finally:
			self.group.setSchema(self.fileName, schema is not None)
			del schema

class XSTCInstanceTest(XSTCTestCase):

	def __init__(self, groupName, name, accepted, file, val, descr):
		XSTCTestCase.__init__(self, 0, groupName, name, accepted, file, val, descr)

	def validate(self):
		instance = None
		schema = None
		filePath = self.fileName
		# os.path.join(options.baseDir, self.fileName)

		if not self.group.schemaParsed and self.group.schemaTried:
			self.failNoSchema()
			return
					
		self.debugMsg("loading instance: %s" % filePath)
		parserCtxt = libxml2.newParserCtxt()
		if (parserCtxt is None):
			# TODO: Is this one necessary, or will an exception
			# be already raised?
			raise Exception("Could not create the instance parser context.")
		if not options.validationSAX:
			try:
				try:
					instance = parserCtxt.ctxtReadFile(filePath, None, libxml2.XML_PARSE_NOWARNING)
				except:
					# Suppress exceptions.
					pass
			finally:
				del parserCtxt
			self.debugMsg("after loading instance")
			if instance is None:
				self.debugMsg("instance is None")
				self.failCritical("Failed to parse the instance for unknown reasons.")
				return		
		try:
			#
			# Validate the instance.
			#
			self.debugMsg("loading schema: %s" % self.group.schemaFileName)
			schema = parseSchema(self.group.schemaFileName)
			try:
				validationCtxt = schema.schemaNewValidCtxt()
				#validationCtxt = libxml2.schemaNewValidCtxt(None)
				if (validationCtxt is None):
					self.failCritical("Could not create the validation context.")
					return
				try:
					self.debugMsg("validating instance")
					if options.validationSAX:
						instance_Err = validationCtxt.schemaValidateFile(filePath, 0)
					else:
						instance_Err = validationCtxt.schemaValidateDoc(instance)
					self.debugMsg("after instance validation")
					self.debugMsg("instance-err: %d" % instance_Err)
					if (instance_Err != 0 and self.val == 1) or (instance_Err == 0 and self.val == 0):
						self.debugMsg("instance result is BAD")
						if (instance_Err != 0):
							self.fail(msgInstanceNotValidButShould)
						else:
							self.fail(msgInstanceValidButShouldNot)

					else:
								self.debugMsg("instance result is OK")
				finally:
					del validationCtxt
			finally:
				del schema
		finally:
			if instance is not None:
				instance.freeDoc()


####################
# Test runner class.
#

class XSTCTestRunner:

	CNT_TOTAL = 0
	CNT_RAN = 1
	CNT_SUCCEEDED = 2
	CNT_FAILED = 3
	CNT_UNIMPLEMENTED = 4
	CNT_INTERNAL = 5
	CNT_BAD = 6
	CNT_EXCEPTED = 7
	CNT_MEMLEAK = 8
	CNT_NOSCHEMA = 9
	CNT_NOTACCEPTED = 10
	CNT_SCHEMA_TEST = 11

	def __init__(self):
		self.logFile = None
		self.counters = self.createCounters()
		self.testList = []
		self.combinesRan = {}
		self.groups = {}
		self.curGroup = None

	def createCounters(self):
		counters = {self.CNT_TOTAL:0, self.CNT_RAN:0, self.CNT_SUCCEEDED:0,
		self.CNT_FAILED:0, self.CNT_UNIMPLEMENTED:0, self.CNT_INTERNAL:0, self.CNT_BAD:0,
		self.CNT_EXCEPTED:0, self.CNT_MEMLEAK:0, self.CNT_NOSCHEMA:0, self.CNT_NOTACCEPTED:0,
		self.CNT_SCHEMA_TEST:0}

		return counters

	def addTest(self, test):
		self.testList.append(test)
		test.initTest(self)

	def getGroup(self, groupName):
		return self.groups[groupName]

	def addGroup(self, group):
		self.groups[group.name] = group

	def updateCounters(self, test, counters):
		if test.memLeak != 0:
			counters[self.CNT_MEMLEAK] += 1
		if not test.failed:
			counters[self.CNT_SUCCEEDED] +=1
		if test.failed:
			counters[self.CNT_FAILED] += 1
		if test.bad:
			counters[self.CNT_BAD] += 1
		if test.unimplemented:
			counters[self.CNT_UNIMPLEMENTED] += 1
		if test.internalErr:
			counters[self.CNT_INTERNAL] += 1
		if test.noSchemaErr:
			counters[self.CNT_NOSCHEMA] += 1
		if test.excepted:
			counters[self.CNT_EXCEPTED] += 1
		if not test.accepted:
			counters[self.CNT_NOTACCEPTED] += 1
		if test.isSchema:
			counters[self.CNT_SCHEMA_TEST] += 1
		return counters

	def displayResults(self, out, all, combName, counters):
		out.write("\n")
		if all:
			if options.combines is not None:
				out.write("combine(s): %s\n" % str(options.combines))
		elif combName is not None:
			out.write("combine : %s\n" % combName)
		out.write("  total           : %d\n" % counters[self.CNT_TOTAL])
		if all or options.combines is not None:
			out.write("  ran             : %d\n" % counters[self.CNT_RAN])
			out.write("    (schemata)    : %d\n" % counters[self.CNT_SCHEMA_TEST])
		# out.write("    succeeded       : %d\n" % counters[self.CNT_SUCCEEDED])
		out.write("  not accepted    : %d\n" % counters[self.CNT_NOTACCEPTED])
		if counters[self.CNT_FAILED] > 0:		    
			out.write("    failed                  : %d\n" % counters[self.CNT_FAILED])
			out.write("     -> internal            : %d\n" % counters[self.CNT_INTERNAL])
			out.write("     -> unimpl.             : %d\n" % counters[self.CNT_UNIMPLEMENTED])
			out.write("     -> skip-invalid-schema : %d\n" % counters[self.CNT_NOSCHEMA])
			out.write("     -> bad                 : %d\n" % counters[self.CNT_BAD])
			out.write("     -> exceptions          : %d\n" % counters[self.CNT_EXCEPTED])
			out.write("    memory leaks            : %d\n" % counters[self.CNT_MEMLEAK])

	def displayShortResults(self, out, all, combName, counters):
		out.write("Ran %d of %d tests (%d schemata):" % (counters[self.CNT_RAN],
				  counters[self.CNT_TOTAL], counters[self.CNT_SCHEMA_TEST]))
		# out.write("    succeeded       : %d\n" % counters[self.CNT_SUCCEEDED])
		if counters[self.CNT_NOTACCEPTED] > 0:
			out.write(" %d not accepted" % (counters[self.CNT_NOTACCEPTED]))
		if counters[self.CNT_FAILED] > 0 or counters[self.CNT_MEMLEAK] > 0:
			if counters[self.CNT_FAILED] > 0:
				out.write(" %d failed" % (counters[self.CNT_FAILED]))
				out.write(" (")
				if counters[self.CNT_INTERNAL] > 0:
					out.write(" %d internal" % (counters[self.CNT_INTERNAL]))
				if counters[self.CNT_UNIMPLEMENTED] > 0:
					out.write(" %d unimplemented" % (counters[self.CNT_UNIMPLEMENTED]))
				if counters[self.CNT_NOSCHEMA] > 0:
					out.write(" %d skip-invalid-schema" % (counters[self.CNT_NOSCHEMA]))
				if counters[self.CNT_BAD] > 0:
					out.write(" %d bad" % (counters[self.CNT_BAD]))
				if counters[self.CNT_EXCEPTED] > 0:
					out.write(" %d exception" % (counters[self.CNT_EXCEPTED]))
				out.write(" )")
			if counters[self.CNT_MEMLEAK] > 0:
				out.write(" %d leaks" % (counters[self.CNT_MEMLEAK]))			
			out.write("\n")
		else:
			out.write(" all passed\n")

	def reportCombine(self, combName):
		global options

		counters = self.createCounters()
		#
		# Compute evaluation counters.
		#
		for test in self.combinesRan[combName]:
			counters[self.CNT_TOTAL] += 1
			counters[self.CNT_RAN] += 1
			counters = self.updateCounters(test, counters)
		if options.reportErrCombines and (counters[self.CNT_FAILED] == 0) and (counters[self.CNT_MEMLEAK] == 0):
			pass
		else:
			if options.enableLog:
				self.displayResults(self.logFile, False, combName, counters)				
			self.displayResults(sys.stdout, False, combName, counters)

	def displayTestLog(self, test):
		sys.stdout.writelines(test.log)
		sys.stdout.write("~~~~~~~~~~\n")

	def reportTest(self, test):
		global options

		error = test.failed or test.memLeak != 0
		#
		# Only erroneous tests will be written to the log,
		# except @verbose is switched on.
		#
		if options.enableLog and (options.verbose or error):
			self.logFile.writelines(test.log)
			self.logFile.write("~~~~~~~~~~\n")
		#
		# if not @silent, only erroneous tests will be
		# written to stdout, except @verbose is switched on.
		#
		if not options.silent:
			if options.reportInternalErrOnly and test.internalErr:
				self.displayTestLog(test)
			if options.reportMemLeakErrOnly and test.memLeak != 0:
				self.displayTestLog(test)
			if options.reportUnimplErrOnly and test.unimplemented:
				self.displayTestLog(test)
			if (options.verbose or error) and (not options.reportInternalErrOnly) and (not options.reportMemLeakErrOnly) and (not options.reportUnimplErrOnly):
				self.displayTestLog(test)


	def addToCombines(self, test):
		found = False
		if self.combinesRan.has_key(test.combineName):
			self.combinesRan[test.combineName].append(test)
		else:
			self.combinesRan[test.combineName] = [test]

	def run(self):

		global options

		if options.info:
			for test in self.testList:
				self.addToCombines(test)
			sys.stdout.write("Combines: %d\n" % len(self.combinesRan))
			sys.stdout.write("%s\n" % self.combinesRan.keys())
			return

		if options.enableLog:
			self.logFile = open(options.logFile, "w")
		try:
			for test in self.testList:
				self.counters[self.CNT_TOTAL] += 1
				#
				# Filter tests.
				#
				if options.singleTest is not None and options.singleTest != "":
					if (test.name != options.singleTest):
						continue
				elif options.combines is not None:
					if not options.combines.__contains__(test.combineName):
						continue
				elif options.testStartsWith is not None:
					if not test.name.startswith(options.testStartsWith):
						continue
				elif options.combineStartsWith is not None:
					if not test.combineName.startswith(options.combineStartsWith):
						continue
				
				if options.maxTestCount != -1 and self.counters[self.CNT_RAN] >= options.maxTestCount:
					break
				self.counters[self.CNT_RAN] += 1
				#
				# Run the thing, dammit.
				#
				try:
					test.setUp()
					try:
						test.run()
					finally:
						test.tearDown()
				finally:
					#
					# Evaluate.
					#
					test.finalize()
					self.reportTest(test)
					if options.reportCombines or options.reportErrCombines:
						self.addToCombines(test)
					self.counters = self.updateCounters(test, self.counters)
		finally:
			if options.reportCombines or options.reportErrCombines:
				#
				# Build a report for every single combine.
				#
				# TODO: How to sort a dict?
				#
				self.combinesRan.keys().sort(None)
				for key in self.combinesRan.keys():
					self.reportCombine(key)

			#
			# Display the final report.
			#
			if options.silent:
				self.displayShortResults(sys.stdout, True, None, self.counters)
			else:
				sys.stdout.write("===========================\n")
				self.displayResults(sys.stdout, True, None, self.counters)