"""``PostgresTaskStore`` 的惰性导入契约——**不需要数据库**。 钉住两条承诺: 1. 取到 ``chongming_task.PostgresTaskStore`` 之前,iss 不导入 psycopg——核心包保持 零第三方运行时依赖(``import chongming_task`` 不会因为没装驱动而变慢或失败); 2. 没装可选依赖时,报错必须**可操作**:给出安装命令,而不是一个光秃秃的 ``ModuleNotFoundError``。 真正的行为测试在 ``tests/integration/test_postgres_store.py``(需要 DSN)。 """ from __future__ import annotations import os import subprocess import sys import pytest import chongming_task def _run(script: str) -> subprocess.CompletedProcess[str]: environment = {**os.environ} return subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, env=environment, timeout=60, ) def test_star_import_does_not_pull_the_driver() -> None: """``from chongming_task import *`` 不该要求安装数据库驱动。""" assert "PostgresTaskStore" not in chongming_task.__all__ script = ( "import sys\n" "from chongming_task import *\n" "assert 'psycopg' not in sys.modules, '星号导入拉起了 psycopg'\n" "assert 'PostgresTaskStore' not in dir(), '它不该出现在星号导入里'\n" "print('ok')\n" ) completed = _run(script) assert completed.returncode == 0, completed.stderr assert completed.stdout.strip() == "ok" def test_lazy_attribute_does_not_import_the_driver_until_used() -> None: script = ( "import sys\n" "import chongming_task\n" "assert 'psycopg' not in sys.modules, '导入核心包就拉起了 psycopg'\n" "print('ok')\n" ) completed = _run(script) assert completed.returncode == 0, completed.stderr assert completed.stdout.strip() == "ok" def test_missing_driver_gives_an_actionable_error() -> None: """把 psycopg 屏蔽成"未安装",取 ``PostgresTaskStore`` 必须报出安装指引。""" script = ( "import sys\n" "sys.modules['psycopg'] = None\n" "sys.modules['psycopg_pool'] = None\n" "import chongming_task\n" "try:\n" " chongming_task.PostgresTaskStore\n" "except ImportError as exc:\n" " text = str(exc)\n" " assert 'chongming-task[postgres]' in text, text\n" " print('ok')\n" "else:\n" " raise SystemExit('没有抛出 ImportError')\n" ) completed = _run(script) assert completed.returncode == 0, completed.stderr assert completed.stdout.strip() == "ok" def test_unknown_attribute_still_raises_attribute_error() -> None: """``__getattr__`` 只放行已知的可选导出,其余照常抛 AttributeError。""" with pytest.raises(AttributeError, match="no attribute"): _ = chongming_task.NoSuchThing # type: ignore[attr-defined] def test_postgres_store_resolves_when_the_extra_is_installed() -> None: """装了可选依赖时必须能取到——并在 stores 包与顶层给出同一个类。""" pytest.importorskip("psycopg") from chongming_task.stores import PostgresTaskStore as FromStores from chongming_task.stores.postgres import PostgresTaskStore as FromModule assert chongming_task.PostgresTaskStore is FromModule assert FromStores is FromModule