Skip to main content
 首页 » 编程设计

python之Django 单元测试 : model. all() 查询集在单独线程中运行的方法中为空

2024年08月28日29oomusou

我有一个在单独的线程中运行的方法,它会进行一些联系人匹配。

我正在编写测试来检查联系人是否已同步。测试用例是这样的:

class ContactSyncTestCase(TestCase): 
    fixtures = ['fix.json'] 
 
    def setUp(self): 
        # get a few contacts that exist in the database to be sent for matching 
        self.few_contacts = CompanyContact.objects.all().order_by('?')[:5].values_list('contact_number',flat=True) 
 
    def test_submit_contacts(self): 
 
        # Pick up a random user 
        user = User.objects.all().order_by('?')[0] 
 
        # Get API Key for that user 
        key = ApiKey.objects.get(user=user).key 
 
        # The url that submits contacts for matching, returns a matching key immediately and starts a separate thread to sync contacts 
        sync_request_url = '/sync_contacts/?username=%s&api_key=%s'%(user.username,key) 
        sync_request_response = self.client.post(path=sync_request_url, 
                                    data=json.dumps({"contacts":','.join(self.few_contacts)}), 
                                    content_type="application/json") 
 
        # Key that is used to fetch the status of contacts syncing and returns a json if contacts are matched 
        key = sync_request_response.content 
 
        # At this point, the other thread is doing the task of syncing inside the method mentioned next 
 
        # Hence I put the test on pause so that it does not fail and exit 
        try: 
            while True: 
                time.sleep(100) 
        except KeyboardInterrupt: 
            pass 

匹配数字的异步方法是这样开始的:

def match_numbers(key, contacts, user): 
    # Get all contacts stored in the system 
    """ 
 
    :param key: 
    :param contacts: 
    :param user: 
    """ 
    import pdb;pdb.set_trace() 
    system_contacts = CompanyContact.objects.all().values_list('contact_number', flat=True) 

现在这里的奇怪问题是:

CompanyContact.objects.all().values_list('contact_number', flat=True) 

测试时返回一个空的查询集。但是,在运行时它工作正常。

就此而言,任何查询(包括用户模型)都会返回一个空查询集。

有什么想法吗?

编辑:

事实证明,从 TrasactionTestCase 继承解决了这个问题。我仍然有疑问,因此我挖掘了更多。

我的数据库的默认事务级别是可重复读取。

从 this post 读取,

REPEATABLE READ(默认):确保事务发出相同的 SELECT 两次,两次得到相同的结果,而不管其他事务所做的已提交或未提交的更改。换句话说,它从同一查询的不同执行中获得一致的结果。在某些数据库系统中,REPEATABLE READ 隔离级别允许幻象,这样如果另一个事务插入新行,在 SELECT 语句之间的间隔中,第二个 SELECT 将看到它们。 InnoDB 并非如此; REPEATABLE READ 级别不会出现幻影。

总结:我应该还有现有的记录,但我没有。

请您参考如下方法:

您可以在 django LiveServerTestCase 中找到很好的解释

class LiveServerTestCase(TransactionTestCase): 
    """ 
    ... 
    Note that it inherits from TransactionTestCase instead of TestCase because 
    the threads do not share the same transactions (unless if using in-memory 
    sqlite) and each thread needs to commit all their transactions so that the 
    other thread can see the changes. 
    """