pytest_workload.py 文件源码

def __init__(self, env, workers):
        """Initialize WorkloadInterrupted object instance.

        Args:
            env(testlib.common3.Environment): TAF environment instance

        """
        self.env = env

        # Filter environment device for workload
        # get device with hw.stress_tool_attributes
        self.devices = [dev for dev in self.env.id_map.values()
                        if hasattr(dev, 'hw') and hasattr(dev.hw, 'stress_tool_attributes')]
        manager = Manager()
        self.workload_results = {}
        for dev in self.devices:
            self.workload_results[dev.id] = manager.list([])  # pylint: disable=no-member
        self.pool = ThreadPool(len(self.devices))
        self.workers = get_workers(workers)
        if not self.workers:
            self.workers = {'time': WORKLOAD_TIME}
        else:
            if not int(self.workers.get('time', 0)):
                self.workers['time'] = WORKLOAD_TIME