def __init__(self, env, flag, item):
"""Create PidChecker instance.
Args:
env (testlib.common3.Environment): Environment instance from 'env' fixture.
flag (bool): Flag to know if plugin should skip process validation.
item (pytest.Item): Test case instance.
"""
self.env = env
self.processes = {}
self.flag = flag
self.item_name = item.name
self.skip_details = item.get_marker("skip_pidchecker")
if self.skip_details and self.skip_details.args != ():
self.skip_prcheck = self.skip_details.args
else:
self.skip_prcheck = None
python类Item()的实例源码
def _send_post_request(self, item):
"""Send post request to reporting server or add it to queue.
Args:
item(pytest.Item): test case item
"""
tc_name = get_tcname(item)
try:
env_prop = item.config.env.env_prop
except AttributeError:
buildname = self.UNDEFINED_BUILD
else:
buildname = self.buildname(env_prop)
suite_name = get_suite_name(item.nodeid)
info = {"brief": get_brief(item, tc_name), "description": get_steps(item, tc_name)}
if self.post_queue:
self._send_post_queue(item, buildname)
self.server_cmd("post", [self.self_name, buildname, suite_name, tc_name, "Run", "", info, self._get_build_info(item)])
def _send_post_queue(self, item=None, buildname=None, sanity=False):
"""Send info about test execution to the Reporting Server.
Args:
item(pytest.Item): test case item
buildname(str): buildname
sanity(bool): True if sanity test
"""
if buildname is None:
buildname = 'undetermined'
if sanity:
buildname += "-sanity"
self.server_cmd("post", [self.self_name, buildname, "", "", "Run", "", "", self._get_build_info(item)])
for post_req in self.post_queue:
post_req[1] = buildname
# Add empty description and brief. In synapsert new TC won't create
post_req.append(('', ''))
post_req.append(self._get_build_info(item))
self.server_cmd("post", post_req)
self.post_queue[:] = []
def _get_build_info(self, item=None):
"""Get info about build.
Args:
item(pytest.Item): test case item
Returns:
dict{"platform": str, "build": str}: build info
"""
if item is not None and item.config and hasattr(item.config, 'env')\
and item.config.env and "chipName" in item.config.env.env_prop \
and "switchppVersion" in item.config.env.env_prop and self.platform == 'undetermined':
self.platform = item.config.env.env_prop["chipName"]
self.build = item.config.env.env_prop["switchppVersion"]
return {'platform': self.platform, 'build': self.build}
def _sessionstart(self, item):
"""Tell to XMLRPC Server that we are going to interact with it.
Args:
item(pytest.Item): test case item
"""
self.class_logger.info("Configuring reporting server...")
self.server_cmd("open", [self.self_name])
for _var in MODULES:
if "reports_conf." in _var:
commands = MODULES[_var].ReportingServerConfig._sessionstart( # pylint: disable=protected-access
self.class_logger, item, self.self_name, self.buildname(item.config.env.env_prop))
for comm in commands:
self.server_cmd(*comm)
# Order TM reporting to server.
# Order and configure XML report to server.
def test_xfail_item(testdir):
# Ensure pytest.xfail works with non-Python Item
testdir.makeconftest("""
import pytest
class MyItem(pytest.Item):
nodeid = 'foo'
def runtest(self):
pytest.xfail("Expected Failure")
def pytest_collect_file(path, parent):
return MyItem("foo", parent)
""")
result = testdir.inline_run()
passed, skipped, failed = result.listoutcomes()
assert not failed
xfailed = [r for r in skipped if hasattr(r, 'wasxfail')]
assert xfailed
def test_issue88_initial_file_multinodes(self, testdir):
testdir.makeconftest("""
import pytest
class MyFile(pytest.File):
def collect(self):
return [MyItem("hello", parent=self)]
def pytest_collect_file(path, parent):
return MyFile(path, parent)
class MyItem(pytest.Item):
pass
""")
p = testdir.makepyfile("def test_hello(): pass")
result = testdir.runpytest(p, "--collect-only")
result.stdout.fnmatch_lines([
"*MyFile*test_issue88*",
"*Module*test_issue88*",
])
def test_multiple_items_per_collector_byid(self, testdir):
c = testdir.makeconftest("""
import pytest
class MyItem(pytest.Item):
def runtest(self):
pass
class MyCollector(pytest.File):
def collect(self):
return [MyItem(name="xyz", parent=self)]
def pytest_collect_file(path, parent):
if path.basename.startswith("conftest"):
return MyCollector(path, parent)
""")
result = testdir.runpytest(c.basename+"::"+"xyz")
assert result.ret == 0
result.stdout.fnmatch_lines([
"*1 pass*",
])
def pytest_collectreport(self, report):
if report.failed:
self.stats.setdefault("error", []).append(report)
elif report.skipped:
self.stats.setdefault("skipped", []).append(report)
items = [x for x in report.result if isinstance(x, pytest.Item)]
self._numcollected += len(items)
if self.isatty:
#self.write_fspath_result(report.nodeid, 'E')
self.report_collect()
def pytest_namespace():
collect = dict(Item=Item, Collector=Collector, File=File, Session=Session)
return dict(collect=collect)
def __init__(self, name, parent=None, config=None, session=None):
super(Item, self).__init__(name, parent, config, session)
self._report_sections = []
def _matchnodes(self, matching, names):
if not matching or not names:
return matching
name = names[0]
assert name
nextnames = names[1:]
resultnodes = []
for node in matching:
if isinstance(node, pytest.Item):
if not names:
resultnodes.append(node)
continue
assert isinstance(node, pytest.Collector)
rep = collect_one_node(node)
if rep.passed:
has_matched = False
for x in rep.result:
# TODO: remove parametrized workaround once collection structure contains parametrization
if x.name == name or x.name.split("[")[0] == name:
resultnodes.extend(self.matchnodes([x], nextnames))
has_matched = True
# XXX accept IDs that don't have "()" for class instances
if not has_matched and len(rep.result) == 1 and x.name == "()":
nextnames.insert(0, name)
resultnodes.extend(self.matchnodes([x], nextnames))
node.ihook.pytest_collectreport(report=rep)
return resultnodes
def genitems(self, node):
self.trace("genitems", node)
if isinstance(node, pytest.Item):
node.ihook.pytest_itemcollected(item=node)
yield node
else:
assert isinstance(node, pytest.Collector)
rep = collect_one_node(node)
if rep.passed:
for subnode in rep.result:
for x in self.genitems(subnode):
yield x
node.ihook.pytest_collectreport(report=rep)
def _getscopeitem(self, scope):
if scope == "function":
# this might also be a non-function Item despite its attribute name
return self._pyfuncitem
node = get_scope_node(self._pyfuncitem, scope)
if node is None and scope == "class":
# fallback to function item itself
node = self._pyfuncitem
assert node
return node
def get_last_record_from_log(self, log, item):
"""Return last log records for TC.
Args:
log(str): log file
item(pytest.Item): test case item
Returns:
str: Log related to specified test item
"""
started = False
tc_name = get_tcname(item)
fin = open(log)
lines = []
while True:
line = fin.readline()
# EOF
if line == "":
break
# Select last block "TC started ... TC finished" in log
if not started:
if tc_name in line and "started" in line:
started = True
lines = []
lines.append(line)
else:
lines.append(line)
if tc_name in line and "finished" in line:
started = False
fin.close()
return " ".join(lines)
def __init__(self, env, item):
"""Initialize HeatChecker instance.
Args:
env (testlib.common3.Environment): Environment instance.
item (pytest.Item): Test case instance.
"""
self.env = env
self.logger = item.config.ctlogger
def pytest_runtest_setup(self, item):
"""Add info about test case start time.
Args:
item(pytest.Item): test case item
"""
if not item.config.option.tc_duration:
self.detailed_duration[item.nodeid] = dict()
self.detailed_duration[item.nodeid]['setup'] = time.time()
if self._buildname is not None:
self._send_post_request(item)
def pytest_collectreport(self, report):
if report.failed:
self.stats.setdefault("error", []).append(report)
elif report.skipped:
self.stats.setdefault("skipped", []).append(report)
items = [x for x in report.result if isinstance(x, pytest.Item)]
self._numcollected += len(items)
if self.isatty:
#self.write_fspath_result(report.nodeid, 'E')
self.report_collect()
def pytest_namespace():
collect = dict(Item=Item, Collector=Collector, File=File, Session=Session)
return dict(collect=collect)
def __init__(self, name, parent=None, config=None, session=None):
super(Item, self).__init__(name, parent, config, session)
self._report_sections = []
def _matchnodes(self, matching, names):
if not matching or not names:
return matching
name = names[0]
assert name
nextnames = names[1:]
resultnodes = []
for node in matching:
if isinstance(node, pytest.Item):
if not names:
resultnodes.append(node)
continue
assert isinstance(node, pytest.Collector)
rep = collect_one_node(node)
if rep.passed:
has_matched = False
for x in rep.result:
# TODO: remove parametrized workaround once collection structure contains parametrization
if x.name == name or x.name.split("[")[0] == name:
resultnodes.extend(self.matchnodes([x], nextnames))
has_matched = True
# XXX accept IDs that don't have "()" for class instances
if not has_matched and len(rep.result) == 1 and x.name == "()":
nextnames.insert(0, name)
resultnodes.extend(self.matchnodes([x], nextnames))
node.ihook.pytest_collectreport(report=rep)
return resultnodes
def genitems(self, node):
self.trace("genitems", node)
if isinstance(node, pytest.Item):
node.ihook.pytest_itemcollected(item=node)
yield node
else:
assert isinstance(node, pytest.Collector)
rep = collect_one_node(node)
if rep.passed:
for subnode in rep.result:
for x in self.genitems(subnode):
yield x
node.ihook.pytest_collectreport(report=rep)
def _getscopeitem(self, scope):
if scope == "function":
# this might also be a non-function Item despite its attribute name
return self._pyfuncitem
node = get_scope_node(self._pyfuncitem, scope)
if node is None and scope == "class":
# fallback to function item itself
node = self._pyfuncitem
assert node
return node
def pytest_collectreport(report):
i = 0
for x in report.result:
if isinstance(x, pytest.Item):
try:
# Call our setup (which may do a skip, in which
# case we won't count it).
pytest_runtest_setup(x)
i += 1
except:
continue
State.numcollected += i
def pytest_collectreport(self, report):
if report.failed:
self.stats.setdefault("error", []).append(report)
elif report.skipped:
self.stats.setdefault("skipped", []).append(report)
items = [x for x in report.result if isinstance(x, pytest.Item)]
self._numcollected += len(items)
if self.isatty:
#self.write_fspath_result(report.nodeid, 'E')
self.report_collect()
def pytest_namespace():
collect = dict(Item=Item, Collector=Collector, File=File, Session=Session)
return dict(collect=collect)
def __init__(self, name, parent=None, config=None, session=None):
super(Item, self).__init__(name, parent, config, session)
self._report_sections = []
def _matchnodes(self, matching, names):
if not matching or not names:
return matching
name = names[0]
assert name
nextnames = names[1:]
resultnodes = []
for node in matching:
if isinstance(node, pytest.Item):
if not names:
resultnodes.append(node)
continue
assert isinstance(node, pytest.Collector)
rep = collect_one_node(node)
if rep.passed:
has_matched = False
for x in rep.result:
# TODO: remove parametrized workaround once collection structure contains parametrization
if x.name == name or x.name.split("[")[0] == name:
resultnodes.extend(self.matchnodes([x], nextnames))
has_matched = True
# XXX accept IDs that don't have "()" for class instances
if not has_matched and len(rep.result) == 1 and x.name == "()":
nextnames.insert(0, name)
resultnodes.extend(self.matchnodes([x], nextnames))
node.ihook.pytest_collectreport(report=rep)
return resultnodes
def genitems(self, node):
self.trace("genitems", node)
if isinstance(node, pytest.Item):
node.ihook.pytest_itemcollected(item=node)
yield node
else:
assert isinstance(node, pytest.Collector)
rep = collect_one_node(node)
if rep.passed:
for subnode in rep.result:
for x in self.genitems(subnode):
yield x
node.ihook.pytest_collectreport(report=rep)
def _getscopeitem(self, scope):
if scope == "function":
# this might also be a non-function Item despite its attribute name
return self._pyfuncitem
node = get_scope_node(self._pyfuncitem, scope)
if node is None and scope == "class":
# fallback to function item itself
node = self._pyfuncitem
assert node
return node
