"""
扫描所有已导入的 BoxHandler / Limit 子类：
- 如果子类重写了 sync 方法且源码中含阻塞 IO 调用
- 但未重写对应的 async_* 方法
- → 视为违反契约，PR 阶段拦截
豁免机制：在类上加 `allow_default_async_delegation(reason="...")` 装饰器即可跳过检查。
"""
import ast
import importlib
import inspect
import pkgutil
import unittest
from typing import List, Tuple
import olc
from olc.control.handler_chain.box_handler import BoxHandler
from olc.limit.limiter import Limit
BLOCKING_IO_SYMBOLS = {
"requests",
"urlopen",
"urllib",
"socket",
}
LIMITER_SYNC_METHODS = {
"try_acquire",
"pre_check",
"roll_back",
"dec_token",
}
EXEMPT_CLASSES = {
}
def allow_default_async_delegation(reason: str):
"""装饰器：明确豁免某个子类不需要覆盖 async_*。
必须给出 reason —— 便于审查时判断豁免是否仍然成立。
"""
if not reason or not isinstance(reason, str):
raise ValueError("reason must be a non-empty string")
def decorator(cls):
cls.__olc_async_delegation_exempt__ = reason
return cls
return decorator
def _iter_subclasses(base):
seen = set()
stack = [base]
while stack:
cls = stack.pop()
for sub in cls.__subclasses__():
if sub in seen:
continue
seen.add(sub)
stack.append(sub)
yield sub
def _import_all_modules(package):
for _, name, _ in pkgutil.walk_packages(
package.__path__, prefix=package.__name__ + "."
):
try:
importlib.import_module(name)
except Exception:
pass
def _source_calls_blocking_io(source: str) -> List[str]:
"""简单 AST 扫描：识别源码中疑似阻塞 IO 的调用"""
hits = []
try:
tree = ast.parse(source)
except SyntaxError:
return hits
for node in ast.walk(tree):
if isinstance(node, ast.Attribute):
full = _dotted(node)
if full:
head = full.split(".")[0]
if head in BLOCKING_IO_SYMBOLS and head != "socket":
hits.append(full)
elif head == "socket" and full in (
"socket.recv",
"socket.send",
"socket.sendall",
"socket.recvfrom",
):
hits.append(full)
elif isinstance(node, ast.Call):
if isinstance(node.func, ast.Name) and node.func.id in BLOCKING_IO_SYMBOLS:
hits.append(node.func.id)
return hits
def _source_calls_limiter_methods(source: str) -> List[str]:
"""专用扫描：识别 BoxHandler 间接调用 limiter 同步方法的场景。
例如 FlowHandler.in_coming 里 `limiter.try_acquire(n)` 这种调用。
"""
hits = []
try:
tree = ast.parse(source)
except SyntaxError:
return hits
for node in ast.walk(tree):
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
if node.func.attr in LIMITER_SYNC_METHODS:
hits.append(node.func.attr)
return hits
def _dotted(node):
parts = []
cur = node
while isinstance(cur, ast.Attribute):
parts.append(cur.attr)
cur = cur.value
if isinstance(cur, ast.Name):
parts.append(cur.id)
return ".".join(reversed(parts))
return None
def _async_override_missing(cls, sync_name: str, async_name: str) -> bool:
"""sync 在 cls 自己有重写，但 async 仍然停留在 ABC 默认实现"""
sync_owner = next((c for c in cls.__mro__ if sync_name in c.__dict__), None)
async_owner = next((c for c in cls.__mro__ if async_name in c.__dict__), None)
if sync_owner is None:
return False
return sync_owner is cls and async_owner is not cls
CONTRACT_PAIRS_HANDLER = [
("do_in_coming", "async_do_in_coming"),
("do_out_coming", "async_do_out_coming"),
]
CONTRACT_PAIRS_LIMIT = [
("try_acquire", "async_try_acquire"),
("pre_check", "async_pre_check"),
("roll_back", "async_roll_back"),
("dec_token", "async_dec_token"),
]
def _collect_violations(
base, pairs, extra_detector=None
) -> List[Tuple[type, str, List[str]]]:
"""收集违规：sync 方法在子类重写、但 async 仍走 ABC 默认委托。
含 IO 的判定：默认走 _source_calls_blocking_io；如果传入 extra_detector，
则两个判定**或**关系（任何一个识别到 IO 信号即纳入候选）。
"""
violations = []
for cls in _iter_subclasses(base):
if inspect.isabstract(cls):
continue
full_name = f"{cls.__module__}.{cls.__qualname__}"
if full_name in EXEMPT_CLASSES:
continue
if getattr(cls, "__olc_async_delegation_exempt__", None):
continue
try:
src = inspect.getsource(cls)
except (OSError, TypeError):
continue
blocking_hits = _source_calls_blocking_io(src)
if extra_detector is not None:
blocking_hits = blocking_hits + extra_detector(src)
if not blocking_hits:
continue
for sync_name, async_name in pairs:
if _async_override_missing(cls, sync_name, async_name):
violations.append((cls, async_name, blocking_hits))
return violations
def _format_violations(violations) -> str:
lines = ["以下类含阻塞 IO 但未覆盖对应 async_* 方法："]
for cls, async_name, hits in violations:
lines.append(
f" - {cls.__module__}.{cls.__qualname__}: "
f"缺失 {async_name}，疑似 IO 调用 = {sorted(set(hits))}"
)
lines.append(
"\n请显式覆盖对应的 async_* 方法走真异步路径，"
"或加 @allow_default_async_delegation(reason='...') 装饰器豁免。"
)
return "\n".join(lines)
class TestAsyncContract(unittest.TestCase):
"""spec §14.2：异步契约的 CI 强制检查"""
@classmethod
def setUpClass(cls):
_import_all_modules(olc)
def test_handler_async_contract(self):
violations = _collect_violations(
BoxHandler,
CONTRACT_PAIRS_HANDLER,
extra_detector=_source_calls_limiter_methods,
)
self.assertFalse(violations, _format_violations(violations))
def test_limit_async_contract(self):
violations = _collect_violations(Limit, CONTRACT_PAIRS_LIMIT)
self.assertFalse(violations, _format_violations(violations))
class TestContractInfrastructure(unittest.TestCase):
"""验证 _collect_violations / allow_default_async_delegation 装饰器本身可用"""
def test_decorator_requires_reason(self):
with self.assertRaises(ValueError):
allow_default_async_delegation("")
def test_decorator_sets_marker(self):
@allow_default_async_delegation(reason="test")
class Foo:
pass
self.assertEqual(Foo.__olc_async_delegation_exempt__, "test")
def test_blocking_io_detection_positive(self):
src = "def x():\n requests.get('http://x')\n"
hits = _source_calls_blocking_io(src)
self.assertTrue(any("requests" in h for h in hits))
def test_blocking_io_detection_negative(self):
src = "def x():\n return 1 + 2\n"
hits = _source_calls_blocking_io(src)
self.assertEqual(hits, [])
def test_limiter_method_detection_positive(self):
src = "def x():\n limiter.try_acquire(1)\n limiter.dec_token(3)\n"
hits = _source_calls_limiter_methods(src)
self.assertIn("try_acquire", hits)
self.assertIn("dec_token", hits)
def test_limiter_method_detection_negative(self):
src = "def x():\n return 1 + 2\n"
hits = _source_calls_limiter_methods(src)
self.assertEqual(hits, [])
if __name__ == "__main__":
unittest.main(verbosity=2)