Will Keeling

你发布的代码有几个问题。

首先,你不能通过 is_admin 我是说, is_superuser 和 is_staff 当您在中创建用户模型时 create_user() 是的。这些论点在 kwargs 应该这样通过:

def create_user(self, email, shop_name='', mall_name='', password=None, **kwargs):
    if not email:
        raise ValueError('You must enter an email address')

    user = self.model(
        email=self.normalize_email(email), 
        shop_name=shop_name, 
        mall_name=mall_name,
        **kwargs  # Pass the additional keyword arguments 
    )
    ...

第二,在 save() 方法 Shop ,你只是打电话 保存() 在没有弹头的超类上对…的呼唤 super(Shop, self).save() 可能总是会发生-例如。

def save(self, *args, **kwargs):
    try:
        if not self.slug:
            self.slug = slugify(self.shop_name + " shop " + get_random_string(20) )
        super(Shop, self).save(*args, **kwargs)  # Always call the save in the superclass
        #add shop_name with 10 randomly generated characters to create slug

    except IntegrityError:
        pass  # Bad idea to swallow exceptions.

悄悄地忽略 IntegrityError 是的。最好完全放弃try/except,或者至少记录错误。

paginate_queryset() 接受 request 作为它的第二个位置参数。这个 请求 是为了 page_size 可以从中获得。

你不能通过 请求 所以我很好奇为什么这个视图能工作-因为python应该 TypeError 因为参数数目不正确。

无论如何,请尝试更新代码以通过 请求 作为第二个论点:

page = self.paginate_queryset(pc, request)

默认情况下,Docker Compose使用与容器名称相同的主机名使容器可发现。因此,可以通过主机名发现Redis容器 redis . 但是,Django容器使用的是主机名 redis_db .

更新您的 docker-compose.yml 并且改变 REDIS_URI 要引用正确的主机名:

REDIS_URI: redis://redis:6379

实现这一点有几种方法。最简单的方法可能就是放弃 getitem() 嵌套函数并使用 OperationalError s.然后您可以用补丁验证尝试的次数。 gi 对象的 call_count . 例如:

def test_wait_for_db(self):
    """Test waiting for db"""

    with patch('django.db.utils.ConnectionHandler.__getitem__') as gi:
        gi.side_effect = [OperationalError] * 5 + [True]
        call_command('wait_for_db')
        self.assertGreaterEqual(gi.call_count, 5)  # Verify using the call_count

如果你想保留 Ge() 功能,那么我想你只需要 attempt 变量 nonlocal 因此可以在嵌套函数中看到:

def test_wait_for_db(self):
    """Test waiting for db"""
    attempt = 0

    def getitem(alias):
        nonlocal attempt  # Make the outer attempt variable visible
        if attempt < 5:
            attempt += 1
            raise OperationalError()
        else:
            return True

    with patch('django.db.utils.ConnectionHandler.__getitem__') as gi:
        gi.side_effect = getitem
        call_command('wait_for_db')
        self.assertGreaterEqual(attempt, 5)

第三,正如评论中所建议的,您可以创建一个具有 尝试 属性,并使用类的实例作为副作用:

def test_wait_for_db(self):
    """Test waiting for db"""

    class Getitem:
        def __init__(self):
            self.attempt = 0

        def __call__(self, item):
            if self.attempt < 5:
                self.attempt += 1
                raise OperationalError()
            else:
                return True

    with patch('django.db.utils.ConnectionHandler.__getitem__') as gi:
        getitem = Getitem()
        gi.side_effect = getitem
        call_command('wait_for_db')
        self.assertGreaterEqual(getitem.attempt, 5)  # Access the attempts from the instance