为什么我的 App Engine 查询计数在我的测试中不正确？

Question

我正在尝试创建测试以验证我的实体是否已保存在数据库中。当我在 post 函数中放置断点时，我可以看到客户计数在记录保存后发生了变化。

> /Users/Bryan/work/GoogleAppEngine/dermalfillersecrets/main.py(137)post()  
-> customer.put()  
(Pdb) l  
134             query = Customer.query()  
135             orig_customer_count = query.count()  
136             import pdb; pdb.set_trace()  
137  ->         customer.put()  
138             import pdb; pdb.set_trace()  
139             query_params = {'leadbook_name': leadbook_name}  
140             self.redirect('/?' + urllib.urlencode(query_params))  
141       
142     config = {}  
(Pdb) orig_customer_count  
5  
(Pdb) c  
> /Users/Bryan/work/GoogleAppEngine/dermalfillersecrets/main.py(139)post()  
-> query_params = {'leadbook_name': leadbook_name}  
(Pdb) l  
134             query = Customer.query()  
135             orig_customer_count = query.count()  
136             import pdb; pdb.set_trace()  
137             customer.put()  
138             import pdb; pdb.set_trace()  
139  ->         query_params = {'leadbook_name': leadbook_name}  
140             self.redirect('/?' + urllib.urlencode(query_params))  
141       
142     config = {}  
143     config['webapp2_extras.sessions'] = {  
144         'secret_key': 'my-super-secret-key',  
(Pdb) query.count()  
6

实体也显示在数据存储区查看器中。

然而，我的测试一直失败。

F  
======================================================================  
FAIL: test_guest_can_submit_contact_info (dermalfillersecrets.functional_tests.NewVisitorTest)  
----------------------------------------------------------------------  
Traceback (most recent call last):  
  File "/Users/Bryan/work/GoogleAppEngine/dermalfillersecrets/functional_tests.py", line 75, in test_guest_can_submit_contact_info  
    self.assertNotEqual(orig_custs, query.count())  
AssertionError: 0 == 0

这是 functional_test.py 文件内容：

import os, sys  
sys.path.append("/usr/local/google_appengine")  
sys.path.append("/usr/local/google_appengine/lib/yaml/lib")  
sys.path.append("/usr/local/google_appengine/lib/webapp2-2.5.2")  
sys.path.append("/usr/local/google_appengine/lib/django-1.5")  
sys.path.append("/usr/local/google_appengine/lib/cherrypy")  
sys.path.append("/usr/local/google_appengine/lib/concurrent")  
sys.path.append("/usr/local/google_appengine/lib/docker")  
sys.path.append("/usr/local/google_appengine/lib/requests")  
sys.path.append("/usr/local/google_appengine/lib/websocket")  
sys.path.append("/usr/local/google_appengine/lib/fancy_urllib")  
sys.path.append("/usr/local/google_appengine/lib/antlr3")  

import unittest  
from selenium import webdriver  
from google.appengine.api import memcache  
from google.appengine.ext import db  
from google.appengine.ext import testbed  
import dev_appserver    
from google.appengine.tools.devappserver2 import devappserver2  


class NewVisitorTest(unittest.TestCase):  

    def setUp(self):  
        self.testbed = testbed.Testbed()  
        self.testbed.activate()  
        #self.testbed.setup_env(app_id='dermalfillersecrets')  
        self.testbed.init_user_stub()  
        self.testbed.init_datastore_v3_stub()  
        self.testbed.init_memcache_stub()  

        # setup the dev_appserver  
        APP_CONFIGS = ['app.yaml']  

        self.browser = webdriver.Firefox()  
        self.browser.implicitly_wait(3)  

    def tearDown(self):  
        self.browser.quit()  
        self.testbed.deactivate()  

    def test_guest_can_submit_contact_info(self):  
        from main import Customer  
        query = Customer.query()  
        orig_custs = query.count()  
        self.browser.get('http://localhost:8080')  
        self.browser.find_element_by_name('id_name').send_keys("Kallie Wheelock")  
        self.browser.find_element_by_name('id_street').send_keys("123 main st")  
        self.browser.find_element_by_name('id_phone').send_keys('(404)555-1212')  
        self.browser.find_element_by_name('id_zip').send_keys("30306")  
        self.browser.find_element_by_name('submit').submit()  
        # this should return 1 more record  
        #import pdb; pdb.set_trace()  
        query = Customer.query()   
        self.assertNotEqual(orig_custs, query.count())  
        assert(Customer.query(Customer.name == "Kallie Wheelock").get())  
        # Delete the Customer record  
        Customer.query(Customer.name =="Kallie Wheelock").delete()

Answer 1

Appengine 高复制数据存储在 'eventually consistent' 模型上工作。也就是说，如果您插入一个实体，则不能保证后续查询将获取该实体。会有一个非零的时间延迟量。为了使您的查询保持一致，您必须确保您的实体属于同一实体组并使用祖先查询。

https://cloud.google.com/developers/articles/balancing-strong-and-eventual-consistency-with-google-cloud-datastore/

Answer 2

您可以设置测试台来模拟强一致性（如果它们对您的应用程序来说并不重要，那么它们实际上是一致的）：

from google.appengine.datastore import datastore_stub_util

policy = datastore_stub_util.PseudoRandomHRConsistencyPolicy(probability=1)
testbed.init_datastore_v3_stub(consistency_policy=policy)

了解更多信息 here。

顺便说一句，考虑使用 nosetests 插件，让你的测试更干净。

为什么我的 App Engine 查询计数在我的测试中不正确？

Why are my App Engine query counts incorrect in my tests?

testing

google-app-engine

selenium-webdriver