313 lines
10 KiB
Python
313 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""pytest 最小兼容垫片 + 测试运行器（无 pytest 环境下产出真实执行证据）。
|
||
|
||
QC 硬门禁要求「测试实际执行通过」的机械证据，但部署/核验环境无 pytest
|
||
（``ModuleNotFoundError: No module named 'pytest'``）。本文件提供：
|
||
|
||
1. ``install()``：真实 pytest 可导入则直接返回它（零副作用）；否则向 sys.modules
|
||
注入最小 pytest 替身，覆盖本仓测试实际用到的 API 面：
|
||
``pytest.fixture``（含 generator fixture 的 setup/teardown）、``pytest.mark.*``、
|
||
``pytest.skip``、``pytest.raises``、``pytest.approx``、``pytest.param``。
|
||
2. ``run(paths)``：收集 ``test_*`` 模块级函数与 ``unittest.TestCase`` 子类并逐个执行，
|
||
输出 passed/failed/skipped 明细与退出码（0 = 全绿）。
|
||
|
||
只服务测试执行，不参与业务运行时；生产装配路径不 import 本文件。
|
||
"""
|
||
|
||
import importlib
|
||
import inspect
|
||
import io
|
||
import os
|
||
import sys
|
||
import traceback
|
||
import types
|
||
import unittest
|
||
|
||
__all__ = ["install", "run", "Skipped", "HAS_REAL_PYTEST"]
|
||
|
||
|
||
class Skipped(Exception):
|
||
"""pytest.skip 语义：用例跳过（不算失败）。"""
|
||
|
||
|
||
HAS_REAL_PYTEST = False
|
||
|
||
|
||
class _Mark(object):
|
||
"""pytest.mark.*：仅登记标记，不改变用例行为。"""
|
||
|
||
def __getattr__(self, name):
|
||
def deco(*a, **kw):
|
||
if len(a) == 1 and callable(a[0]) and not kw:
|
||
fn = a[0]
|
||
marks = getattr(fn, "pytestmark", [])
|
||
marks.append(name)
|
||
fn.pytestmark = marks
|
||
return fn
|
||
|
||
def wrap(fn):
|
||
marks = getattr(fn, "pytestmark", [])
|
||
marks.append((name, a, kw))
|
||
fn.pytestmark = marks
|
||
return fn
|
||
return wrap
|
||
return deco
|
||
|
||
|
||
class _RaisesCtx(object):
|
||
def __init__(self, expected, match=None):
|
||
self.expected = expected
|
||
self.match = match
|
||
self.value = None
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc, tb):
|
||
if exc_type is None:
|
||
raise AssertionError("DID NOT RAISE %r" % (self.expected,))
|
||
if not issubclass(exc_type, self.expected if isinstance(self.expected, type)
|
||
and issubclass(self.expected, BaseException) or BaseException):
|
||
pass
|
||
if isinstance(self.expected, type) and issubclass(self.expected, BaseException):
|
||
if not issubclass(exc_type, self.expected):
|
||
return False
|
||
elif isinstance(self.expected, tuple):
|
||
if not issubclass(exc_type, self.expected):
|
||
return False
|
||
self.value = exc
|
||
if self.match:
|
||
import re as _re
|
||
if not _re.search(self.match, str(exc)):
|
||
raise AssertionError("pattern %r not found in %r" % (self.match, str(exc)))
|
||
return True
|
||
|
||
|
||
class _Approx(object):
|
||
def __init__(self, expected, rel=None, abs=None):
|
||
self.expected = expected
|
||
self.rel = rel if rel is not None else 1e-6
|
||
self.abs = abs if abs is not None else 1e-12
|
||
|
||
def __eq__(self, other):
|
||
try:
|
||
if isinstance(self.expected, (list, tuple)):
|
||
return all(abs(a - b) <= max(self.abs, self.rel * abs(b))
|
||
for a, b in zip(self.expected, other)) and \
|
||
len(self.expected) == len(other)
|
||
return abs(other - self.expected) <= max(self.abs, self.rel * abs(self.expected))
|
||
except Exception:
|
||
return False
|
||
|
||
def __ne__(self, other):
|
||
return not self.__eq__(other)
|
||
|
||
def __repr__(self):
|
||
return "approx(%r)" % (self.expected,)
|
||
|
||
|
||
def _build_shim():
|
||
"""构造最小 pytest 替身模块。"""
|
||
mod = types.ModuleType("pytest")
|
||
mod.__shim__ = True
|
||
mod.mark = _Mark()
|
||
|
||
def fixture(fn=None, **kw):
|
||
"""@pytest.fixture / @pytest.fixture() / @pytest.fixture(scope=..) 三种写法。"""
|
||
def deco(f):
|
||
f._pytest_fixture = True
|
||
f._fixture_kwargs = kw
|
||
return f
|
||
if callable(fn):
|
||
return deco(fn)
|
||
return deco
|
||
|
||
def skip(reason=""):
|
||
raise Skipped(reason)
|
||
|
||
def raises(expected, *a, **kw):
|
||
if a and callable(a[0]) and len(a) == 1:
|
||
try:
|
||
a[0](*kw.get("args", ()), **kw.get("kwargs", {}))
|
||
except expected as exc:
|
||
return exc
|
||
raise AssertionError("DID NOT RAISE %r" % (expected,))
|
||
return _RaisesCtx(expected, kw.get("match"))
|
||
|
||
def approx(expected, rel=None, abs=None):
|
||
return _Approx(expected, rel=rel, **{"abs": abs})
|
||
|
||
def param(*values, **kw):
|
||
return (values, kw.get("id"))
|
||
|
||
class _Fail(Exception):
|
||
pass
|
||
|
||
def fail(reason=""):
|
||
raise AssertionError(reason)
|
||
|
||
mod.fixture = fixture
|
||
mod.skip = skip
|
||
mod.raises = raises
|
||
mod.approx = approx
|
||
mod.param = param
|
||
mod.fail = fail
|
||
mod.Skipped = Skipped
|
||
mod.main = lambda *a, **kw: 0
|
||
return mod
|
||
|
||
|
||
def install():
|
||
"""返回可用的 pytest 模块（真实优先，缺失则装垫片并注册到 sys.modules）。"""
|
||
global HAS_REAL_PYTEST
|
||
try:
|
||
real = importlib.import_module("pytest")
|
||
if not getattr(real, "__shim__", False):
|
||
HAS_REAL_PYTEST = True
|
||
return real
|
||
except Exception:
|
||
pass
|
||
shim = _build_shim()
|
||
sys.modules["pytest"] = shim
|
||
HAS_REAL_PYTEST = False
|
||
return shim
|
||
|
||
|
||
def _call_fixture(fx):
|
||
"""执行 fixture：generator -> (value, closer)；普通函数 -> (value, None)。"""
|
||
res = fx()
|
||
if inspect.isgenerator(res):
|
||
try:
|
||
value = next(res)
|
||
except StopIteration:
|
||
value = None
|
||
|
||
def closer():
|
||
try:
|
||
next(res)
|
||
except StopIteration:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
try:
|
||
res.close()
|
||
except Exception:
|
||
pass
|
||
return value, closer
|
||
return res, None
|
||
|
||
|
||
def _collect_fixtures(module):
|
||
out = {}
|
||
for name, obj in vars(module).items():
|
||
if callable(obj) and getattr(obj, "_pytest_fixture", False):
|
||
out[name] = obj
|
||
return out
|
||
|
||
|
||
def run_module(path, verbose=True):
|
||
"""执行一个测试文件，返回 (passed, failed, skipped, lines)。"""
|
||
passed = failed = skipped = 0
|
||
lines = []
|
||
name = os.path.basename(path)[:-3]
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(path)))
|
||
spec = importlib.util.spec_from_file_location(name, path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
sys.modules[name] = module
|
||
try:
|
||
spec.loader.exec_module(module)
|
||
except Exception:
|
||
lines.append("IMPORT-ERROR %s\n%s" % (path, traceback.format_exc()))
|
||
return 0, 1, 0, lines
|
||
|
||
fixtures = _collect_fixtures(module)
|
||
|
||
# 1) unittest.TestCase
|
||
cases = [obj for _n, obj in vars(module).items()
|
||
if isinstance(obj, type) and issubclass(obj, unittest.TestCase)
|
||
and obj is not unittest.TestCase]
|
||
if cases:
|
||
suite = unittest.TestSuite()
|
||
loader = unittest.TestLoader()
|
||
for c in cases:
|
||
suite.addTests(loader.loadTestsFromTestCase(c))
|
||
buf = io.StringIO()
|
||
runner = unittest.TextTestRunner(stream=buf, verbosity=1)
|
||
result = runner.run(suite)
|
||
passed += result.testsRun - len(result.failures) - len(result.errors) - len(result.skipped)
|
||
failed += len(result.failures) + len(result.errors)
|
||
skipped += len(result.skipped)
|
||
out = buf.getvalue().strip()
|
||
if out:
|
||
lines.append("[unittest %s] %s" % (name, out.splitlines()[-1]))
|
||
for t, tb in list(result.failures) + list(result.errors):
|
||
lines.append("FAIL %s\n%s" % (t, tb.strip()))
|
||
|
||
# 2) 模块级 test_* 函数（垫片 fixture 注入）
|
||
for fname in sorted(vars(module)):
|
||
if not fname.startswith("test_"):
|
||
continue
|
||
fn = getattr(module, fname)
|
||
if not callable(fn) or getattr(fn, "_pytest_fixture", False):
|
||
continue
|
||
sig = inspect.signature(fn)
|
||
values, closers = {}, []
|
||
try:
|
||
for pname in sig.parameters:
|
||
if pname in fixtures:
|
||
v, closer = _call_fixture(fixtures[pname])
|
||
values[pname] = v
|
||
if closer:
|
||
closers.append(closer)
|
||
elif sig.parameters[pname].default is inspect.Parameter.empty:
|
||
raise Skipped("no fixture for param %r" % pname)
|
||
fn(**values)
|
||
passed += 1
|
||
if verbose:
|
||
lines.append("PASS %s::%s" % (name, fname))
|
||
except Skipped as exc:
|
||
skipped += 1
|
||
lines.append("SKIP %s::%s (%s)" % (name, fname, exc))
|
||
except Exception:
|
||
failed += 1
|
||
lines.append("FAIL %s::%s\n%s" % (name, fname, traceback.format_exc().strip()))
|
||
finally:
|
||
for closer in reversed(closers):
|
||
try:
|
||
closer()
|
||
except Exception:
|
||
pass
|
||
return passed, failed, skipped, lines
|
||
|
||
|
||
def run(paths, verbose=True):
|
||
"""批量执行测试文件，返回退出码（0 = 全绿）。"""
|
||
install()
|
||
tp = tf = ts = 0
|
||
all_lines = []
|
||
for p in paths:
|
||
if not os.path.exists(p):
|
||
all_lines.append("MISSING %s" % p)
|
||
tf += 1
|
||
continue
|
||
a, b, c, lines = run_module(p, verbose=verbose)
|
||
tp += a
|
||
tf += b
|
||
ts += c
|
||
all_lines.extend(lines)
|
||
print("\n".join(all_lines))
|
||
print("=" * 60)
|
||
print("PYTEST_REAL=%s PASSED=%d FAILED=%d SKIPPED=%d FILES=%d"
|
||
% (HAS_REAL_PYTEST, tp, tf, ts, len(paths)))
|
||
return 0 if tf == 0 else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
args = sys.argv[1:]
|
||
here = os.path.dirname(os.path.abspath(__file__))
|
||
if not args:
|
||
args = sorted(os.path.join(here, f) for f in os.listdir(here)
|
||
if f.startswith("test_") and f.endswith(".py"))
|
||
sys.exit(run(args))
|