Python 导入排查：同名文件遮住标准库时，为什么 -P 还不够？

"""Only diagnose the top-level decimal module in this controlled fixture."""
import sys

# Check before any diagnostic helper could import decimal.
preloaded = "decimal" in sys.modules
import importlib.util
import json
import os

record = {
    "preloaded": preloaded,
    "executable": sys.executable,
    "version": sys.version,
    "argv": sys.argv,
    "cwd": os.getcwd(),
    "pythonpath": os.environ.get("PYTHONPATH"),
    "sys_path": sys.path,
    "flags": {key: getattr(sys.flags, key) for key in
              ("ignore_environment", "safe_path", "isolated", "no_site",
               "no_user_site", "dont_write_bytecode")},
}


def finish(result, code):
    record["result"] = result
    print(json.dumps(record, ensure_ascii=False, sort_keys=True))
    raise SystemExit(code)


if preloaded or "decimal" in sys.modules:
    finish("preloaded_decimal", 3)
try:
    # Never substitute a dotted name: its parent package may execute.
    spec = importlib.util.find_spec("decimal")
    record["spec_origin"] = None if spec is None else spec.origin
    import decimal
    record["module_origin"] = decimal.__spec__.origin
    record["module_file"] = decimal.__file__
    record["label"] = getattr(decimal, "LABEL", None)
    try:
        from decimal import Decimal
    except ImportError as exc:
        record["exception"] = type(exc).__name__
        record["message"] = str(exc)
        finish("missing_Decimal", 2)
    finish(str(Decimal("0.1") + Decimal("0.2")), 0)
except Exception as exc:
    record["exception"] = type(exc).__name__
    record["message"] = str(exc)
    finish("probe_error", 4)

"""Run fresh CPython subprocesses; never derive expected.json from results."""
import argparse
import base64
import hashlib
import io
import json
import os
from pathlib import Path
import platform
import subprocess
import sys
import tempfile

HERE = Path(__file__).absolute().parent
PYTHON = str(Path(sys.executable).absolute())
EXPECTED_BYTES = (HERE / "expected.json").read_bytes()
EXPECTED = json.loads(EXPECTED_BYTES)
PROJECT_BYTES = b'LABEL = "PROJECT"\n'
INJECTED_BYTES = b'LABEL = "INJECTED"\n'
# id, extra flags, injected PYTHONPATH, entry mode, renamed fixture
CASES = [
    ("script", [], False, "script", False),
    ("script_E", ["-E"], False, "script", False),
    ("script_P", ["-P"], False, "script", False),
    ("script_P_injected", ["-P"], True, "script", False),
    ("script_E_injected", ["-E"], True, "script", False),
    ("script_EP_injected", ["-E", "-P"], True, "script", False),
    ("script_I_injected", ["-I"], True, "script", False),
    ("module", [], False, "module", False),
    ("module_P", ["-P"], False, "module", False),
    ("renamed", [], False, "script", True),
    ("renamed_injected", [], True, "script", True),
]
BASELINE = r'''
import decimal, hashlib, json, pathlib, sys, sysconfig
p = pathlib.Path(decimal.__file__).resolve()
stdlib = pathlib.Path(sysconfig.get_path("stdlib")).resolve()
if p.parent != stdlib or p.name != "decimal.py":
    raise RuntimeError("Unexpected stdlib layout: " + str(p))
if pathlib.Path(decimal.__spec__.origin).resolve() != p:
    raise RuntimeError("Baseline origin mismatch")
v = str(decimal.Decimal("0.1") + decimal.Decimal("0.2"))
if v != "0.3":
    raise RuntimeError("Baseline Decimal contract failed")
print(json.dumps({"origin":str(p), "stdlib":str(stdlib), "result":v,
 "sha256":hashlib.sha256(p.read_bytes()).hexdigest(),
 "executable":sys.executable, "version":sys.version}))
'''


class ValidationError(RuntimeError):
    pass


class ProbeTimeout(ValidationError):
    def __init__(self, receipt):
        self.receipt = receipt
        super().__init__(f"Subprocess timed out after {receipt['timeout_seconds']} seconds")


def clean_env(inherited=None):
    source = os.environ if inherited is None else inherited
    return {key: value for key, value in source.items()
            if not key.upper().startswith("PYTHON")}


def execute(args, cwd, env):
    command = [PYTHON, "-B", "-S", *args]
    try:
        proc = subprocess.run(command, cwd=cwd, env=env,
                              capture_output=True, text=False, timeout=15)
    except subprocess.TimeoutExpired as exc:
        # Capture in binary mode on every platform. Keep exact partial bytes,
        # including incomplete UTF-8, rather than decoding or discarding.
        def captured(value):
            return None if value is None else base64.b64encode(value).decode("ascii")
        receipt = {"command": command, "cwd": str(cwd),
                   "target_pythonpath": env.get("PYTHONPATH"),
                   "timeout_seconds": exc.timeout,
                   "stdout_b64": captured(exc.stdout),
                   "stderr_b64": captured(exc.stderr)}
        # run() has killed and waited for its direct child; no returncode is
        # exposed by TimeoutExpired, and this is not a successful probe row.
        raise ProbeTimeout(receipt) from exc
    def decoded(value):
        # Match the former text-mode receipt: default encoding and universal
        # newline translation, but only after a process completes normally.
        with io.TextIOWrapper(io.BytesIO(value)) as stream:
            return stream.read()
    return {"command": proc.args, "cwd": str(cwd),
            "target_pythonpath": env.get("PYTHONPATH"),
            "returncode": proc.returncode, "stdout": decoded(proc.stdout),
            "stderr": decoded(proc.stderr)}


def check_record(raw, expected, known):
    if expected["source"] == "NO_ENTRY":
        # Do not accept any arbitrary failing command as an entry failure.
        suffix = ": No module named labprobe\n"
        if not (raw["returncode"] == 1 and raw["stdout"] == "" and
                raw["stderr"] == PYTHON + suffix):
            raise ValidationError("Unexpected -m entry failure: " + repr(raw))
        return dict(expected)
    if raw["stderr"]:
        raise ValidationError("Unexpected stderr: " + raw["stderr"])
    try:
        data = json.loads(raw["stdout"])
    except (ValueError, TypeError) as exc:
        raise ValidationError("Probe output is not one JSON record") from exc
    if data.get("preloaded") is not False:
        raise ValidationError("decimal was already loaded")
    origins = [data.get(k) for k in ("spec_origin", "module_origin", "module_file")]
    if any(not isinstance(p, str) for p in origins) or len(set(origins)) != 1:
        raise ValidationError("Missing or different module origins")
    origin = str(Path(origins[0]).resolve())
    if origin not in known:
        raise ValidationError("Unknown source, refusing to normalize: " + origin)
    source = known[origin]
    if data.get("label") != {"PROJECT":"PROJECT", "INJECTED":"INJECTED",
                             "STDLIB":None}[source]:
        raise ValidationError("Source label mismatch")
    if data.get("executable") != PYTHON:
        raise ValidationError("Interpreter changed")
    if not data["flags"]["no_site"] or not data["flags"]["dont_write_bytecode"]:
        raise ValidationError("Missing common -B -S controls")
    if data.get("pythonpath") != raw["target_pythonpath"]:
        raise ValidationError("Environment receipt mismatch")
    actual = {"case": expected["case"], "source": source,
              "result": data.get("result"), "returncode": raw["returncode"]}
    if actual != expected:
        raise ValidationError(f"Expected {expected!r}; got {actual!r}")
    if source != "STDLIB" and data.get("exception") != "ImportError":
        raise ValidationError("Expected missing Decimal ImportError")
    raw["probe"] = data
    return actual


def run_matrix(root, inherited=None):
    if platform.python_implementation() != "CPython" or sys.version_info < (3, 11):
        raise RuntimeError("This lab requires CPython 3.11+")
    root = Path(root).absolute()
    root.mkdir(parents=True, exist_ok=False)
    project, injected, neutral = [root / name for name in
                                  ("project", "injected", "neutral")]
    for directory in (project, injected, neutral):
        directory.mkdir()
    (project / "decimal.py").write_bytes(PROJECT_BYTES)
    (injected / "decimal.py").write_bytes(INJECTED_BYTES)
    (project / "labprobe.py").write_bytes((HERE / "labprobe.py").read_bytes())
    env = clean_env(inherited)
    baseline_raw = execute(["-I", "-c", BASELINE], neutral, env)
    if baseline_raw["returncode"] != 0 or baseline_raw["stderr"]:
        raise ValidationError("Baseline failed: " + repr(baseline_raw))
    baseline = json.loads(baseline_raw["stdout"])
    known = {str((project / "decimal.py").resolve()): "PROJECT",
             str((injected / "decimal.py").resolve()): "INJECTED",
             baseline["origin"]: "STDLIB"}
    if [row[0] for row in CASES] != [row["case"] for row in EXPECTED]:
        raise ValidationError("Case order differs from fixed expected.json")
    rows, normalized = [], []
    for (name, flags, inject, mode, rename), expected in zip(CASES, EXPECTED):
        if rename and (project / "decimal.py").exists():
            (project / "decimal.py").rename(project / "decimal_fixture.py")
        case_env = dict(env)
        if inject:
            case_env["PYTHONPATH"] = str(injected)
        cwd = project if mode == "module" else neutral
        entry = ["-m", "labprobe"] if mode == "module" else [str(project / "labprobe.py")]
        raw = execute([*flags, *entry], cwd, case_env)
        raw["case"] = name
        rows.append(raw)
        try:
            normalized.append(check_record(raw, expected, known))
        except ValidationError as exc:
            raise ValidationError(f"{name}: {exc}; raw={raw!r}") from exc
    (project / "decimal_fixture.py").rename(project / "decimal.py")
    if (project / "decimal.py").read_bytes() != PROJECT_BYTES:
        raise ValidationError("Project fixture bytes changed")
    if (injected / "decimal.py").read_bytes() != INJECTED_BYTES:
        raise ValidationError("Injected fixture bytes changed")
    if list(root.rglob("*.pyc")) or list(root.rglob("__pycache__")):
        raise ValidationError("Bytecode appeared in the new fixture")
    return {"root":str(root), "platform":platform.platform(),
            "expected_sha256":hashlib.sha256(EXPECTED_BYTES).hexdigest(),
            "baseline":baseline, "baseline_raw":baseline_raw,
            "normalized":normalized, "rows":rows,
            "restored_fixture_sha256":{
                "project/decimal.py":hashlib.sha256(PROJECT_BYTES).hexdigest(),
                "injected/decimal.py":hashlib.sha256(INJECTED_BYTES).hexdigest()}}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    output = Path(args.output).absolute()
    # Exclusive creation avoids silently overwriting an earlier receipt.
    with output.open("x", encoding="utf-8") as stream:
        try:
            with tempfile.TemporaryDirectory(prefix="import-shadow-") as temp:
                report = run_matrix(Path(temp) / "fixture")
            json.dump(report, stream, ensure_ascii=False, indent=2)
            stream.write("\n")
        except Exception as exc:
            failure = {"status":"failed", "error":repr(exc)}
            if isinstance(exc, ProbeTimeout):
                failure["timeout"] = exc.receipt
            json.dump(failure, stream, ensure_ascii=False, indent=2)
            stream.write("\n")
            raise
    print(f"PASS {len(report['normalized'])} fixed cases; receipt: {output}")


if __name__ == "__main__":
    main()

[
  {"case":"script","source":"PROJECT","result":"missing_Decimal","returncode":2},
  {"case":"script_E","source":"PROJECT","result":"missing_Decimal","returncode":2},
  {"case":"script_P","source":"STDLIB","result":"0.3","returncode":0},
  {"case":"script_P_injected","source":"INJECTED","result":"missing_Decimal","returncode":2},
  {"case":"script_E_injected","source":"PROJECT","result":"missing_Decimal","returncode":2},
  {"case":"script_EP_injected","source":"STDLIB","result":"0.3","returncode":0},
  {"case":"script_I_injected","source":"STDLIB","result":"0.3","returncode":0},
  {"case":"module","source":"PROJECT","result":"missing_Decimal","returncode":2},
  {"case":"module_P","source":"NO_ENTRY","result":"probe_not_started","returncode":1},
  {"case":"renamed","source":"STDLIB","result":"0.3","returncode":0},
  {"case":"renamed_injected","source":"INJECTED","result":"missing_Decimal","returncode":2}
]

"""Boundary tests plus two independently created experiment roots."""
import base64
import copy
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest import mock
import run_matrix as lab


class ImportMatrixTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.temp = tempfile.TemporaryDirectory(prefix="import-boundaries-")
        cls.addClassCleanup(cls.temp.cleanup)
        cls.base = Path(cls.temp.name)
        cls.report = lab.run_matrix(cls.base / "first")
        cls.known = {
            str(cls.base / "first/project/decimal.py"): "PROJECT",
            str(cls.base / "first/injected/decimal.py"): "INJECTED",
            cls.report["baseline"]["origin"]: "STDLIB",
        }

    def valid_record(self):
        return copy.deepcopy(self.report["rows"][0])

    def reject(self, raw, message):
        with self.assertRaisesRegex(lab.ValidationError, message):
            lab.check_record(raw, lab.EXPECTED[0], self.known)

    def test_two_roots_and_poisoned_parent_environment(self):
        poison = dict(os.environ, PYTHONPATH="/not-used/from-parent",
                      PYTHONHOME="/nonexistent-python-home", PYTHONSAFEPATH="1",
                      PYTHONWARNINGS="error", PYTHONINSPECT="1")
        second = lab.run_matrix(self.base / "second", poison)
        self.assertNotEqual(self.report["root"], second["root"])
        self.assertEqual(self.report["normalized"], lab.EXPECTED)
        self.assertEqual(second["normalized"], lab.EXPECTED)
        self.assertEqual(self.report["baseline"], second["baseline"])
        self.assertEqual(self.report["restored_fixture_sha256"],
                         second["restored_fixture_sha256"])
        for report in (self.report, second):
            for row in report["rows"]:
                if "probe" in row:
                    self.assertNotIn("/not-used/from-parent", row["probe"]["sys_path"])

    def test_environment_removes_all_python_variables(self):
        self.assertEqual(lab.clean_env({"PYTHONHOME":"bad", "PYTHONPATH":"bad",
                                       "PYTHONSAFEPATH":"1", "PATH":"keep"}),
                         {"PATH":"keep"})

    def test_preloaded_module_is_a_real_process_failure(self):
        code = ("import decimal, runpy; runpy.run_path(" +
                repr(str(lab.HERE / "labprobe.py")) + ", run_name='__main__')")
        raw = lab.execute(["-I", "-c", code], self.base, lab.clean_env())
        self.assertEqual(raw["returncode"], 3)
        self.assertEqual(json.loads(raw["stdout"])["result"], "preloaded_decimal")
        self.reject(raw, "already loaded")

    def test_unknown_source_is_not_normalized(self):
        unknown = self.base / "unknown"
        unknown.mkdir()
        (unknown / "decimal.py").write_text('LABEL = "PROJECT"\n', encoding="utf-8")
        env = lab.clean_env()
        env["PYTHONPATH"] = str(unknown)
        raw = lab.execute(["-P", str(lab.HERE / "labprobe.py")], self.base, env)
        self.assertEqual(raw["returncode"], 2)
        self.reject(raw, "Unknown source")

    def test_malformed_stdout_is_rejected(self):
        raw = self.valid_record()
        raw["stdout"] += "unexpected second line\n"
        self.reject(raw, "not one JSON")

    def test_mismatched_origins_are_rejected(self):
        raw = self.valid_record()
        data = json.loads(raw["stdout"])
        data["module_file"] = "/different/decimal.py"
        raw["stdout"] = json.dumps(data)
        self.reject(raw, "different module origins")

    def test_wrong_exit_status_is_rejected(self):
        raw = self.valid_record()
        raw["returncode"] = 0
        self.reject(raw, "Expected")

    def test_unexpected_stderr_is_rejected(self):
        raw = self.valid_record()
        raw["stderr"] = "surprise\n"
        self.reject(raw, "Unexpected stderr")

    def test_any_failure_is_not_a_missing_entry(self):
        raw = copy.deepcopy(self.report["rows"][8])
        raw["stderr"] = "SyntaxError: unrelated failure\n"
        with self.assertRaisesRegex(lab.ValidationError, "entry failure"):
            lab.check_record(raw, lab.EXPECTED[8], self.known)

    def test_binary_capture_keeps_normal_text_receipt_contract(self):
        command = [lab.PYTHON, "-B", "-S", "-c", "pass"]
        completed = subprocess.CompletedProcess(command, 0, b"first\r\nsecond\n", b"note\r\n")
        with mock.patch.object(lab.subprocess, "run", return_value=completed) as run:
            raw = lab.execute(["-c", "pass"], self.base, lab.clean_env())
        self.assertIs(run.call_args.kwargs["text"], False)
        self.assertEqual(raw["stdout"], "first\nsecond\n")
        self.assertEqual(raw["stderr"], "note\n")
        self.assertEqual(raw["returncode"], 0)

    def test_timeout_report_retains_exact_partial_bytes(self):
        output = self.base / "timeout-receipt.json"
        code = ("import sys, time; "
                "sys.stdout.buffer.write(b'partial-out-\\xff'); sys.stdout.flush(); "
                "sys.stderr.buffer.write(b'partial-err-\\xfe'); sys.stderr.flush(); "
                "time.sleep(30)")
        observed = {}

        def stalled_matrix(root):
            observed["cwd"] = str(root.parent)
            return lab.execute(["-c", code], root.parent, lab.clean_env())

        with mock.patch.object(lab, "run_matrix", side_effect=stalled_matrix), \
                mock.patch.object(lab.sys, "argv", ["run_matrix.py", "--output", str(output)]):
            with self.assertRaises(lab.ProbeTimeout):
                lab.main()
        failure = json.loads(output.read_text(encoding="utf-8"))
        self.assertEqual(failure["status"], "failed")
        receipt = failure["timeout"]
        self.assertEqual(receipt["command"], [lab.PYTHON, "-B", "-S", "-c", code])
        self.assertEqual(receipt["cwd"], observed["cwd"])
        self.assertIsNone(receipt["target_pythonpath"])
        self.assertEqual(receipt["timeout_seconds"], 15)
        self.assertEqual(base64.b64decode(receipt["stdout_b64"], validate=True),
                         b"partial-out-\xff")
        self.assertEqual(base64.b64decode(receipt["stderr_b64"], validate=True),
                         b"partial-err-\xfe")
        self.assertNotIn("returncode", receipt)
        self.assertNotIn("normalized", failure)

    def test_timeout_preserves_unobserved_versus_empty_output(self):
        for value, expected in ((None, None), (b"", "")):
            with self.subTest(value=value):
                timeout = subprocess.TimeoutExpired([lab.PYTHON], 15,
                                                    output=value, stderr=value)
                with mock.patch.object(lab.subprocess, "run", side_effect=timeout):
                    with self.assertRaises(lab.ProbeTimeout) as result:
                        lab.execute(["-c", "pass"], self.base, lab.clean_env())
                self.assertEqual(result.exception.receipt["stdout_b64"], expected)
                self.assertEqual(result.exception.receipt["stderr_b64"], expected)

    def test_isolated_flag_does_not_imply_no_site(self):
        # Deliberately omit -S for this separate flag test, not the matrix.
        proc = subprocess.run(
            [lab.PYTHON, "-B", "-I", "-c",
             "import sys; print(sys.flags.no_site, sys.flags.no_user_site, "
             "sys.flags.ignore_environment, int(sys.flags.safe_path))"],
            cwd=self.base, env=lab.clean_env(), capture_output=True,
            text=True, timeout=15)
        self.assertEqual((proc.returncode, proc.stdout, proc.stderr),
                         (0, "0 1 1 1\n", ""))


if __name__ == "__main__":
    unittest.main(verbosity=2)

# Python 导入来源实验

要求：CPython 3.11+，标准文件布局，标准库即可。本文实际验证 Linux / CPython 3.12.14。

将 labprobe.py、run_matrix.py、test_matrix.py、expected.json 与本文件保存到同一目录。
在该目录运行（选择你已安装的 CPython；无需安装任何包）：

    python -E -B -S run_matrix.py --output report.json
    python -E -B -S -m unittest -v test_matrix.py

report.json 必须尚不存在；再次运行请更换输出文件名。驱动使用当前解释器的同一绝对路径启动所有子进程，先清掉继承的 PYTHON*，再按案例注入 PYTHONPATH。
每个案例都新建进程并显式使用 -B -S。临时夹具位于系统临时目录，结束后清理。只有指定的报告留在磁盘上。
不读取用户项目，不安装依赖，不改全局环境或 site-packages。

固定预期有 11 项，由源码作者先写定。成功应输出 PASS 11 fixed cases；unittest 应输出 Ran 13 tests 和 OK。
PROJECT / INJECTED 均缺少 Decimal，探针预期退出 2；NO_ENTRY 预期退出 1。它们是被验证的失败样例，不是忽略错误。
报告保留完整 stdout/stderr、命令、cwd、sys.path 和来源。陌生来源、来源不一致、提前加载或额外 stderr 都使验收失败；不要为了通过而改 expected.json。
不支持的解释器、布局、超时或失败案例使驱动非零退出，并在报告写入失败原因。报告名已存在则直接报错，不覆盖原报告。
超时报告额外包含 timeout 对象：command、cwd、target_pythonpath、timeout_seconds，以及 stdout_b64 / stderr_b64。后两项保存 TimeoutExpired 已捕获的原始字节，用 base64.b64decode(value, validate=True) 还原；不是普通 stdout/stderr 字符串。null 表示异常没有提供该流的输出，空字符串表示零字节，不能混为一谈。
超时仍是 status=failed，不生成成功矩阵或伪造 returncode；该失败报告只保留本次超时的上下文，不包含此前完成的矩阵行。未刷新、尚未写出或未捕获的内容无法恢复。
新增的真实超时测试会额外等待约 15 秒，并核对含非 UTF-8 字节的两路输出；另外两项回归测试区分 null 和空字节，并确认二进制捕获仍保留正常报告的文本和换行约定。subprocess.run 终止并等待直接子进程，不保证管理任意后代进程或构成严格的总时限。

-I 含 -E、-P、-s，不含 -S。测试中的额外 -S 仅用于控制 site 初始化。它们不是运行不可信代码的安全沙箱。
find_spec 仅用于顶层 decimal；对点分名称调用它可能执行父包。
不覆盖 IDE、.pth、自定义 finder、editable install、包布局与已运行会话缓存。