小编Ali*_*cka的帖子

我正在尝试测试模块：

包/模块.py

DATA_PATH = os.path.join(os.path.dirname(__file__), "data")
class SomeClass:
    def __init__(self):
        self.filename = os.path.join(DATA_PATH, "ABC.txt")

在tests/module_test.py中我正在尝试做

from package import module
@patch("package.module.DATA_PATH", "data_path_here") # doesn't work
class TestSomeClass(unittest.TestCase):
    def setUp(self):
        module.DATA_PATH = "data_path_here" # doesn't work
        self.obj= SomeClass()

    @patch("package.module.DATA_PATH", "data_path_here") # doesn't work either
    def test_constructor(self):
        self.assertEqual(r"data_path_here\ABC.txt", self.obj.filename)

但是 DATA_PATH 仍然没有被模拟出来。我想我尝试了所有可能的选项来模拟它，但它仍然返回原始路径而不是“data_path_here”

我究竟做错了什么？

编辑：它不是在 Python 单元测试框架中修改全局变量的副本， 因为该解决方案不起作用

我有一个生成任务（io 绑定任务）的函数：

def get_task():
    while True:
        new_task = _get_task()
        if new_task is not None:
            yield new_task
        else:
            sleep(1)

我正在尝试在 asyncio 中编写一个消费者，该消费者将同时处理最多 10 个任务，完成一项任务，然后将执行新任务。我不确定是否应该使用信号量或者是否有任何类型的 asycio 池执行程序？我开始用线程写一个伪代码：

def run(self)
   while True:
       self.semaphore.acquire() # first acquire, then get task
       t = get_task()
       self.process_task(t)

def process_task(self, task):
   try:
       self.execute_task(task)
       self.mark_as_done(task)
   except:
       self.mark_as_failed(task)
   self.semaphore.release()

有人可以帮助我吗？我不知道把 async/await 关键字放在哪里