Process子类属性测试返回错误值：MongoDB变更流监听器重启判断异常

问题背景 #

基于Process子类实现了一个简易MongoDB变更流监听器，用于规避GIL限制。监听器设计为数据库连接中断时自动重启，但如果5分钟内重启次数达到5次则停止运行。

监听器类代码 #

import sys
from multiprocessing import Queue, Process
import time

class Listener(Process):

    # 记录监听器重启次数
    __no_restarts: int = 0
    __last_start_time: float = 0

    def __init__(self, 
            max_restart: int = 5,
            max_restart_minute: int = 5
        ) -> None:

        # 设置最大重启次数及时间窗口（转换为秒）
        self.max_restart = max_restart
        self.max_restart_sec = max_restart_minute * 60

        # 调用父类初始化方法
        super(Listener, self).__init__()

    @property
    def max_restarts_reached(self) -> bool:
        if (self.__no_restarts >= self.max_restart) and ((time.time() - self.__last_start_time) <= self.max_restart_sec):
            return True
        else:
            return False

    def run(self):
        
        # 更新启动时间与重启计数
        self.__last_start_time = time.time()
        self.__no_restarts += 1

        try:
            ### 变更流监听逻辑（省略）

        # 捕获MongoDB异常后尝试重启
        except PyMongoError as e:
            
            # 判断是否达到最大重启限制
            if self.max_restarts_reached:
                print('Failed')
                return sys.exit(1)

            # 指数退避后重启
            else:
                time.sleep(pow(1, self.__no_restarts))
                return self.run()

测试代码 #

import pytest
import time

import Listener

def test_max_restarts_reached():

    # 初始化监听器实例
    mdb_listener = Listener(max_restart = 5, max_restart_minute = 5)
    
    # 模拟启动时间设置
    mdb_listener.__last_start_time = time.time()

    # 初始状态下未达到重启限制
    assert mdb_listener.max_restarts_reached == False

    # 设置重启次数为10，预期触发限制
    mdb_listener.__no_restarts = 10
    assert mdb_listener.max_restarts_reached == True

    # 让时间窗口过期，预期取消限制
    mdb_listener.__last_start_time = time.time() - 50
    assert mdb_listener.max_restarts_reached == False

问题现象 #

测试时未启动进程，手动模拟run()的状态变更：

• 第一个断言正常通过
• 第二个断言中，即使将__no_restarts设为10，max_restarts_reached仍返回False，与预期逻辑不符
• 将max_restarts_reached从属性改为函数后，结果依旧错误

问题原因 #

Python中以双下划线开头的变量（如__no_restarts、__last_start_time）会触发**名称修饰（name mangling）**机制，Python会自动将这些变量重命名为_类名__变量名（此处为_Listener__no_restarts、_Listener__last_start_time），目的是防止子类意外覆盖父类的私有属性。

测试代码中直接赋值mdb_listener.__no_restarts = 10，实际上是给实例新增了一个普通属性__no_restarts，而非修改类内部真正使用的_Listener__no_restarts，因此max_restarts_reached访问的还是原始的0值，导致判断错误。

解决办法 #

方案1：测试代码中使用名称修饰后的变量名 #

直接访问重命名后的属性，绕过名称修饰：

def test_max_restarts_reached():
    mdb_listener = Listener(max_restart = 5, max_restart_minute = 5)
    
    current_time = time.time()
    # 使用名称修饰后的变量名
    mdb_listener._Listener__last_start_time = current_time
    assert mdb_listener.max_restarts_reached == False

    mdb_listener._Listener__no_restarts = 10
    assert mdb_listener.max_restarts_reached == True

    mdb_listener._Listener__last_start_time = current_time - 50
    assert mdb_listener.max_restarts_reached == False

方案2：给Listener类添加私有状态设置方法（推荐） #

在Listener类中新增专门用于测试的方法，封装私有属性的修改逻辑：

class Listener(Process):
    # ... 原有代码 ...

    def _set_restart_state(self, no_restarts: int, last_start_time: float):
        """仅用于测试：设置重启次数与启动时间"""
        self.__no_restarts = no_restarts
        self.__last_start_time = last_start_time

测试代码修改为：

def test_max_restarts_reached():
    mdb_listener = Listener(max_restart = 5, max_restart_minute = 5)
    
    current_time = time.time()
    mdb_listener._set_restart_state(0, current_time)
    assert mdb_listener.max_restarts_reached == False

    mdb_listener._set_restart_state(10, current_time)
    assert mdb_listener.max_restarts_reached == True

    mdb_listener._set_restart_state(10, current_time - 50)
    assert mdb_listener.max_restarts_reached == False

方案3：将私有属性改为单下划线开头（不推荐） #

如果不需要严格的私有属性保护，可以将双下划线改为单下划线，避免名称修饰：

class Listener(Process):
    _no_restarts: int = 0
    _last_start_time: float = 0

    # ... 其他代码中对应替换变量名 ...

    @property
    def max_restarts_reached(self) -> bool:
        if (self._no_restarts >= self.max_restart) and ((time.time() - self._last_start_time) <= self.max_restart_sec):
            return True
        else:
            return False

    def run(self):
        self._last_start_time = time.time()
        self._no_restarts += 1
        # ... 其他代码 ...

内容的提问来源于stack exchange，提问作者Hbio