astitcher commented on a change in pull request #285:
URL: https://github.com/apache/qpid-proton/pull/285#discussion_r557569492
##########
File path: python/tests/proton_tests/main.py
##########
@@ -330,382 +356,401 @@ def filter(self, record):
SKIP = "skip"
FAIL = "fail"
+
class Runner:
- def __init__(self):
- self.exception = None
- self.exception_phase_name = None
- self.skip = False
-
- def passed(self):
- return not self.exception
-
- def skipped(self):
- return self.skip
-
- def failed(self):
- return self.exception and not self.skip
-
- def halt(self):
- """determines if the overall test execution should be allowed to continue
to the next phase"""
- return self.exception or self.skip
-
- def run(self, phase_name, phase):
- """invokes a test-phase method (which can be the test method itself or a
setUp/tearDown
- method). If the method raises an exception the exception is examined
to see if the
- exception should be classified as a 'skipped' test"""
- # we don't try to catch exceptions for jython because currently a
- # jython bug will prevent the java portion of the stack being
- # stored with the exception info in the sys module
- if opts.bare:
- phase()
- else:
- try:
- phase()
- except KeyboardInterrupt:
- raise
- except:
- self.exception_phase_name = phase_name
- self.exception = sys.exc_info()
- exception_type = self.exception[0]
- self.skip = getattr(exception_type, "skipped", False) or
issubclass(exception_type, SkipTest)
-
- def status(self):
- if self.passed():
- return PASS
- elif self.skipped():
- return SKIP
- elif self.failed():
- return FAIL
- else:
- return None
-
- def get_formatted_exception_trace(self):
- if self.exception:
- if self.skip:
- # format skipped tests without a traceback
- output =
indent("".join(traceback.format_exception_only(*self.exception[:2]))).rstrip()
- else:
- output = "Error during %s:" % self.exception_phase_name
- lines = traceback.format_exception(*self.exception)
- val = self.exception[1]
- reflect = False
- if val and hasattr(val, "getStackTrace"):
- jlines = []
- for frame in val.getStackTrace():
- if reflect and frame.getClassName().startswith("org.python."):
- # this is a heuristic to eliminate the jython portion of
- # the java stack trace
- if not opts.javatrace:
- break
- jlines.append(" %s\n" % frame.toString())
- if frame.getClassName().startswith("java.lang.reflect"):
- reflect = True
+ def __init__(self):
+ self.exception = None
+ self.exception_phase_name = None
+ self.skip = False
+
+ def passed(self):
+ return not self.exception
+
+ def skipped(self):
+ return self.skip
+
+ def failed(self):
+ return self.exception and not self.skip
+
+ def halt(self):
+ """determines if the overall test execution should be allowed to
continue to the next phase"""
+ return self.exception or self.skip
+
+ def run(self, phase_name, phase):
+ """invokes a test-phase method (which can be the test method itself or
a setUp/tearDown
+ method). If the method raises an exception the exception is
examined to see if the
+ exception should be classified as a 'skipped' test"""
+ # we don't try to catch exceptions for jython because currently a
+ # jython bug will prevent the java portion of the stack being
+ # stored with the exception info in the sys module
+ if opts.bare:
+ phase()
+ else:
+ try:
+ phase()
+ except KeyboardInterrupt:
+ raise
+ except:
+ self.exception_phase_name = phase_name
+ self.exception = sys.exc_info()
+ exception_type = self.exception[0]
+ self.skip = getattr(exception_type, "skipped", False) or
issubclass(exception_type, SkipTest)
+
+ def status(self):
+ if self.passed():
+ return PASS
+ elif self.skipped():
+ return SKIP
+ elif self.failed():
+ return FAIL
+ else:
+ return None
+
+ def get_formatted_exception_trace(self):
+ if self.exception:
+ if self.skip:
+ # format skipped tests without a traceback
+ output =
indent("".join(traceback.format_exception_only(*self.exception[:2]))).rstrip()
else:
- reflect = False
- lines.extend(jlines)
- output += indent("".join(lines)).rstrip()
- return output
-
- def get_exception_type(self):
- if self.exception:
- return self.exception[0]
- else:
- return None
+ output = "Error during %s:" % self.exception_phase_name
+ lines = traceback.format_exception(*self.exception)
+ val = self.exception[1]
+ reflect = False
+ if val and hasattr(val, "getStackTrace"):
+ jlines = []
+ for frame in val.getStackTrace():
+ if reflect and
frame.getClassName().startswith("org.python."):
+ # this is a heuristic to eliminate the jython
portion of
+ # the java stack trace
+ if not opts.javatrace:
+ break
+ jlines.append(" %s\n" % frame.toString())
+ if
frame.getClassName().startswith("java.lang.reflect"):
+ reflect = True
+ else:
+ reflect = False
+ lines.extend(jlines)
+ output += indent("".join(lines)).rstrip()
+ return output
+
+ def get_exception_type(self):
+ if self.exception:
+ return self.exception[0]
+ else:
+ return None
+
+ def get_exception_message(self):
+ if self.exception:
+ return self.exception[1]
+ else:
+ return None
- def get_exception_message(self):
- if self.exception:
- return self.exception[1]
- else:
- return None
ST_WIDTH = 8
+
def run_test(name, test, config):
- patterns = filter.patterns
- level = root.level
- filter.patterns = config.log_categories
- root.setLevel(config.log_level)
-
- parts = name.split(".")
- line = None
- output = ""
- for part in parts:
+ patterns = filter.patterns
+ level = root.level
+ filter.patterns = config.log_categories
+ root.setLevel(config.log_level)
+
+ parts = name.split(".")
+ line = None
+ output = ""
+ for part in parts:
+ if line:
+ if len(line) + len(part) >= (WIDTH - ST_WIDTH - 1):
+ output += "%s. \\\n" % line
+ line = " %s" % part
+ else:
+ line = "%s.%s" % (line, part)
+ else:
+ line = part
+
if line:
- if len(line) + len(part) >= (WIDTH - ST_WIDTH - 1):
- output += "%s. \\\n" % line
- line = " %s" % part
- else:
- line = "%s.%s" % (line, part)
- else:
- line = part
-
- if line:
- output += "%s %s" % (line, (((WIDTH - ST_WIDTH) - len(line))*"."))
- sys.stdout.write(output)
- sys.stdout.flush()
- interceptor.begin()
- start = time.time()
- try:
- runner = test()
- finally:
- interceptor.reset()
- end = time.time()
- if interceptor.dirty:
- if interceptor.last != "\n":
- sys.stdout.write("\n")
+ output += "%s %s" % (line, (((WIDTH - ST_WIDTH) - len(line))*"."))
sys.stdout.write(output)
- print(" %s" % colorize_word(runner.status()))
- if runner.failed() or runner.skipped():
- print(runner.get_formatted_exception_trace())
- root.setLevel(level)
- filter.patterns = patterns
- return TestResult(end - start,
- runner.passed(),
- runner.skipped(),
- runner.failed(),
- runner.get_exception_type(),
- runner.get_exception_message(),
- runner.get_formatted_exception_trace())
+ sys.stdout.flush()
+ interceptor.begin()
+ start = time.time()
+ try:
+ runner = test()
+ finally:
+ interceptor.reset()
+ end = time.time()
+ if interceptor.dirty:
+ if interceptor.last != "\n":
+ sys.stdout.write("\n")
+ sys.stdout.write(output)
+ print(" %s" % colorize_word(runner.status()))
+ if runner.failed() or runner.skipped():
+ print(runner.get_formatted_exception_trace())
+ root.setLevel(level)
+ filter.patterns = patterns
+ return TestResult(end - start,
+ runner.passed(),
+ runner.skipped(),
+ runner.failed(),
+ runner.get_exception_type(),
+ runner.get_exception_message(),
+ runner.get_formatted_exception_trace())
+
class TestResult:
- def __init__(self, time, passed, skipped, failed, exception_type,
exception_message, formatted_exception_trace):
- self.time = time
- self.passed = passed
- self.skipped = skipped
- self.failed = failed
- self.exception_type = exception_type
- self.exception_message = exception_message
- self.formatted_exception_trace = formatted_exception_trace
+ def __init__(self, time, passed, skipped, failed, exception_type,
exception_message, formatted_exception_trace):
+ self.time = time
+ self.passed = passed
+ self.skipped = skipped
+ self.failed = failed
+ self.exception_type = exception_type
+ self.exception_message = exception_message
+ self.formatted_exception_trace = formatted_exception_trace
+
class FunctionTest:
- def __init__(self, test):
- self.test = test
+ def __init__(self, test):
+ self.test = test
+
+ def name(self):
+ return "%s.%s" % (self.test.__module__, self.test.__name__)
- def name(self):
- return "%s.%s" % (self.test.__module__, self.test.__name__)
+ def run(self):
+ return run_test(self.name(), self._run, config)
- def run(self):
- return run_test(self.name(), self._run, config)
+ def _run(self):
+ runner = Runner()
+ runner.run("test", lambda: self.test(config))
+ return runner
- def _run(self):
- runner = Runner()
- runner.run("test", lambda: self.test(config))
- return runner
+ def __repr__(self):
+ return "FunctionTest(%r)" % self.test
- def __repr__(self):
- return "FunctionTest(%r)" % self.test
class MethodTest:
- def __init__(self, cls, method):
- self.cls = cls
- self.method = method
+ def __init__(self, cls, method):
+ self.cls = cls
+ self.method = method
- def name(self):
- return "%s.%s.%s" % (self.cls.__module__, self.cls.__name__, self.method)
+ def name(self):
+ return "%s.%s.%s" % (self.cls.__module__, self.cls.__name__,
self.method)
- def run(self):
- return run_test(self.name(), self._run, config)
+ def run(self):
+ return run_test(self.name(), self._run, config)
- def _run(self):
- runner = Runner()
- inst = self.cls(self.method)
- test = getattr(inst, self.method)
+ def _run(self):
+ runner = Runner()
+ inst = self.cls(self.method)
+ test = getattr(inst, self.method)
- if hasattr(inst, "configure"):
- runner.run("configure", lambda: inst.configure(config))
- if runner.halt(): return runner
- if hasattr(inst, "setUp"):
- runner.run("setup", inst.setUp)
- if runner.halt(): return runner
+ if hasattr(inst, "configure"):
+ runner.run("configure", lambda: inst.configure(config))
+ if runner.halt():
+ return runner
+ if hasattr(inst, "setUp"):
+ runner.run("setup", inst.setUp)
+ if runner.halt():
+ return runner
- runner.run("test", test)
+ runner.run("test", test)
- if hasattr(inst, "tearDown"):
- runner.run("teardown", inst.tearDown)
+ if hasattr(inst, "tearDown"):
+ runner.run("teardown", inst.tearDown)
- return runner
+ return runner
+
+ def __repr__(self):
+ return "MethodTest(%r, %r)" % (self.cls, self.method)
- def __repr__(self):
- return "MethodTest(%r, %r)" % (self.cls, self.method)
class PatternMatcher:
- def __init__(self, *patterns):
- self.patterns = patterns
+ def __init__(self, *patterns):
+ self.patterns = patterns
+
+ def matches(self, name):
+ for p in self.patterns:
+ if match(name, p):
+ return True
+ return False
- def matches(self, name):
- for p in self.patterns:
- if match(name, p):
- return True
- return False
class FunctionScanner(PatternMatcher):
- def inspect(self, obj):
- return type(obj) == types.FunctionType and self.matches(obj.__name__)
+ def inspect(self, obj):
+ return type(obj) == types.FunctionType and self.matches(obj.__name__)
+
+ def descend(self, func):
+ # the None is required for older versions of python
+ return
+ yield None
- def descend(self, func):
- # the None is required for older versions of python
- return; yield None
+ def extract(self, func):
+ yield FunctionTest(func)
- def extract(self, func):
- yield FunctionTest(func)
class ClassScanner(PatternMatcher):
- def inspect(self, obj):
- return type(obj) in CLASS_TYPES and self.matches(obj.__name__)
+ def inspect(self, obj):
+ return type(obj) in CLASS_TYPES and self.matches(obj.__name__)
- def descend(self, cls):
- # the None is required for older versions of python
- return; yield None
+ def descend(self, cls):
+ # the None is required for older versions of python
+ return
+ yield None
Review comment:
Huh, isn't this the same code again?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscr...@qpid.apache.org
For additional commands, e-mail: dev-h...@qpid.apache.org
