等待最终一致性在 Selenium 测试中传播需要多长时间？

Question

我正在尝试创建测试以验证在使用 Selenium 时我的实体是否保存在数据库中。当我手动输入表单数据时，代码工作正常，但自动测试失败。

当我在代码的post函数中放置断点时，我可以看到保存记录后客户数发生了变化。我读 https://cloud.google.com/appengine/docs/python/tools/localunittesting#Python_Writing_High_Replication_Datastore_tests

我正在使用我认为具有强一致性的 Ancestor 查询，所以我完全不确定为什么我会遇到这些问题。

据我所知，测试失败是因为最终一致性，解决这个问题的方法是更改 PseudoRandomHRConsistencyPolicy 设置。

policy = datastore_stub_util.PseudoRandomHRConsistencyPolicy(probability=1)

当我再次运行测试时，我得到了同样的错误。

也有人提出，如果我等一段时间，Eventual Consistency就有时间传播。这对测试也不起作用。

创建这些测试我做错了什么？

> /Users/Bryan/work/GoogleAppEngine/dermalfillersecrets/main.py(130)post()  
-> customer.put()  
(Pdb) l  
125             customer.phone = self.request.get('id_phone')  
126             customer.zipcode = int(self.request.get('id_zip'))  
127             # show original number of customer to show the code works  
128             starting_customer_count = Customer.query(ancestor=client.key).count()  
129             import pdb; pdb.set_trace()  
130  ->         customer.put()  
131             final_customer_count = Customer.query(ancestor=client.key).count()  
132             import pdb; pdb.set_trace()  
133             query_params = {'leadbook_name': leadbook_name}  
134             self.redirect('/?' + urllib.urlencode(query_params))  
135       
(Pdb) starting_customer_count  
19  
(Pdb) c  
> /Users/Bryan/work/GoogleAppEngine/dermalfillersecrets/main.py(133)post()  
-> query_params = {'leadbook_name': leadbook_name}  
(Pdb) l     
128             starting_customer_count = Customer.query(ancestor=client.key).count()  
129             import pdb; pdb.set_trace()  
130             customer.put()  
131             final_customer_count = Customer.query(ancestor=client.key).count()  
132             import pdb; pdb.set_trace()  
133  ->         query_params = {'leadbook_name': leadbook_name}  
134             self.redirect('/?' + urllib.urlencode(query_params))  
135       
136     config = {}  
137     config['webapp2_extras.sessions'] = {  
138         'secret_key': 'my-super-secret-key',  
(Pdb) final_customer_count  
20

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

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

$ nosetests --with-gae    
test_guest_can_submit_contact_info (dermalfillersecrets.functional_tests.NewVisitorTest) ... FAIL  

======================================================================  
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_customer_count, final_customer_count)  
AssertionError: 0 == 0

这是 functional_tests.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 Client, Customer  
        # below query didn't work because of eventual consistency  
        #query = Customer.query()  
        client = Client.query( Client.name == "Bryan Wheelock").get()  
        orig_customer_count = Customer.query(ancestor=client.key).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 time; time.sleep(10)  
        final_customer_count = Customer.query(ancestor=client.key).count()  
        self.assertNotEqual(orig_customer_count, final_customer_count)  
        assert(Customer.query(Customer.name == "Kallie Wheelock").get())  
        # Delete the Customer record  
        Customer.query(Customer.name =="Kallie Wheelock").delete()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 Client, Customer  

                client = Client.query( Client.name == "Bryan Wheelock").get()  
                orig_customer_count = Customer.query(ancestor=client.key).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 time; time.sleep(10)  
                final_customer_count = Customer.query(ancestor=client.key).count()  
                self.assertNotEqual(orig_customer_count, final_customer_count)  
                assert(Customer.query(Customer.name == "Kallie Wheelock").get())  
                # Delete the Customer record  
                Customer.query(Customer.name =="Kallie Wheelock").delete()

Answer 1

你想测试什么？

如果您尝试测试保存实体的代码，那么您可以编写一个简单的单元测试：保存实体，检查它是否已保存。

如果您尝试测试客户端代码，那么引入延迟会破坏此类测试的想法。不保证最终一致性的上限。您的 UI 代码应该能够处理任何时间延迟 - 从 none 到至少几秒。

例如，如果实体的数量对于用户工作流程的延续很重要，大多数应用程序会阻止 UI（例如显示某种进度条或微调器），直到调用成功完全的。如果等待这个实体被保存不是用户工作流程的一部分，那么就没有必要在 Selenium 中对其进行测试——同样，您可以编写一个简单的单元测试。

Answer 2

我认为问题在于您的测试设置，而不是最终一致性。

有不同的数据存储实例，您需要确保使用的是正确的实例：

dev_appserver.py 有自己的数据存储
Testbed 也有自己的数据存储

默认情况下，Testbed 不访问 dev_appserver.py 使用的数据存储，反之亦然。

由于 Selenium 使用 dev_appserver.py，您需要明确告诉 dev_appserver.py 和 Testbed 使用相同的数据存储。

下面是我的测试设置中的一个片段。我不确定它是否会运行原样，但它应该能让你朝着正确的方向前进。

请注意，我还明确地启动和停止了 dev_appserver.py。

import subprocess, time, os, unittest, shlex
from google.appengine.ext import db, testbed
from google.appengine.api import apiproxy_stub, apiproxy_stub_map
from selenium import webdriver

class SeleniumTest(unittest.TestCase):

    def setUp(self):
        # Start the dev server
        cmd = "/.../bin/dev_appserver.py /.../app.yaml --storage_path /tmp/datastore --clear_datastore --skip_sdk_update_check"
        self.dev_appserver = subprocess.Popen(shlex.split(cmd), 
                                              stdout=subprocess.PIPE)
        time.sleep(2) # Important, let dev_appserver start up
        self.testbed = testbed.Testbed()
        self.testbed.setup_env(app_id="dev~myapp")
        self.testbed.activate()
        self.testbed.init_app_identity_stub()
        self.testbed.init_datastore_v3_stub(
            datastore_file="/tmp/datastore/datastore.db", use_sqlite=True)
        self.testbed.init_taskqueue_stub()
        self.testbed.init_memcache_stub()
        self.testbed.init_blobstore_stub()
        self.testbed.init_user_stub()
        self.testbed.init_mail_stub()
        self.testbed.init_urlfetch_stub()        
        self.taskqueue_stub = apiproxy_stub_map.apiproxy.GetStub('taskqueue')
        self.mail_stub = apiproxy_stub_map.apiproxy.GetStub('mail')
        self.datastore_stub = apiproxy_stub_map.apiproxy.GetStub('datastore_v3')
        # Start the headless browser for Selenium tests
        self.driver = webdriver.Firefox()

    def tearDown(self):
        self.testbed.deactivate()
        self.driver.quit()
        self.dev_appserver.terminate()

等待最终一致性在 Selenium 测试中传播需要多长时间？

How long to wait for Eventual Consistency to propagate in a Selenium test?

python

google-app-engine

selenium-webdriver