Trying to improve datastore flaky tests.

datastore/api/snippets_test.py

@@ -11,21 +11,39 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-from flaky import flaky
-import gcloud
+from functools import wraps
+import time
+
 from gcloud import datastore
-from nose.plugins.attrib import attr
-from tests import CloudBaseTest
+from tests import CloudBaseTest, mark_flaky
 
 from . import snippets
 
 
-def flaky_filter(e, *args):
-    return isinstance(e, gcloud.exceptions.GCloudError)
+def eventually_consistent(f):
+    @wraps(f)
+    def inner(self, *args, **kwargs):
+        # This is pretty hacky, but make datastore wait 1s after any
+        # put operation to in order to account for eventual consistency.
+        original_put_multi = self.client.put_multi
+
+        def put_multi(*args, **kwargs):
+            result = original_put_multi(*args, **kwargs)
+            time.sleep(1)
+            return result
+
+        self.client.put_multi = put_multi
+
+        try:
+            result = f(self, *args, **kwargs)
+        finally:
+            self.client.put_multi = original_put_multi
+
+        return result
+    return inner
 
 
-@attr('slow')
-@flaky(rerun_filter=flaky_filter)
+@mark_flaky
 class DatastoreSnippetsTest(CloudBaseTest):
 
     def setUp(self):
@@ -110,16 +128,19 @@ def test_batch_lookup(self):
     def test_batch_delete(self):
         snippets.batch_delete(self.client)
 
+    @eventually_consistent
     def test_unindexed_property_query(self):
         tasks = snippets.unindexed_property_query(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_basic_query(self):
         tasks = snippets.basic_query(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_projection_query(self):
         priorities, percents = snippets.projection_query(self.client)
         self.to_delete_entities.extend(self.client.query(kind='Task').fetch())
@@ -131,9 +152,11 @@ def test_ancestor_query(self):
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_run_query(self):
         snippets.run_query(self.client)
 
+    @eventually_consistent
     def test_cursor_paging(self):
         for n in range(6):
             self.to_delete_entities.append(
@@ -147,46 +170,55 @@ def test_cursor_paging(self):
         self.assertTrue(cursor_one)
         self.assertTrue(cursor_two)
 
+    @eventually_consistent
     def test_property_filter(self):
         tasks = snippets.property_filter(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_composite_filter(self):
         tasks = snippets.composite_filter(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_key_filter(self):
         tasks = snippets.key_filter(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_ascending_sort(self):
         tasks = snippets.ascending_sort(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_descending_sort(self):
         tasks = snippets.descending_sort(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_multi_sort(self):
         tasks = snippets.multi_sort(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_keys_only_query(self):
         keys = snippets.keys_only_query(self.client)
         self.to_delete_keys.extend(keys)
         self.assertTrue(keys)
 
+    @eventually_consistent
     def test_distinct_query(self):
         tasks = snippets.distinct_query(self.client)
         self.to_delete_entities.extend(tasks)
         self.assertTrue(tasks)
 
+    @eventually_consistent
     def test_distinct_on_query(self):
         tasks = snippets.distinct_on_query(self.client)
         self.to_delete_entities.extend(tasks)
@@ -241,25 +273,29 @@ def transactional_single_entity_group_read_only(self):
         self.assertTrue(task_list)
         self.assertTrue(tasks_in_list)
 
+    @eventually_consistent
     def test_namespace_run_query(self):
         all_namespaces, filtered_namespaces = snippets.namespace_run_query(
             self.client)
         self.assertTrue(all_namespaces)
         self.assertTrue(filtered_namespaces)
         self.assertTrue('google' in filtered_namespaces)
 
+    @eventually_consistent
     def test_kind_run_query(self):
         kinds = snippets.kind_run_query(self.client)
         self.to_delete_entities.extend(self.client.query(kind='Task').fetch())
         self.assertTrue(kinds)
         self.assertTrue('Task' in kinds)
 
+    @eventually_consistent
     def test_property_run_query(self):
         kinds = snippets.property_run_query(self.client)
         self.to_delete_entities.extend(self.client.query(kind='Task').fetch())
         self.assertTrue(kinds)
         self.assertTrue('Task' in kinds)
 
+    @eventually_consistent
     def test_property_by_kind_run_query(self):
         reprs = snippets.property_by_kind_run_query(self.client)
         self.to_delete_entities.extend(self.client.query(kind='Task').fetch())

datastore/api/tasks_test.py

@@ -10,22 +10,14 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-#
-from flaky import flaky
-import gcloud
+
 from gcloud import datastore
-from nose.plugins.attrib import attr
-from tests import CloudBaseTest
+from tests import CloudBaseTest, mark_flaky
 
 from . import tasks
 
 
-def flaky_filter(e, *args):
-    return isinstance(e, gcloud.exceptions.GCloudError)
-
-
-@attr('slow')
-@flaky(rerun_filter=flaky_filter)
+@mark_flaky
 class DatastoreTasksTest(CloudBaseTest):
 
     def setUp(self):

tests/__init__.py

@@ -16,14 +16,18 @@
     AppEngineTestbedCase,
     capture_stdout,
     CloudBaseTest,
+    flaky_filter,
     Http2Mock,
+    mark_flaky,
     RESOURCE_PATH)
 
 
 __all__ = [
     'AppEngineTestbedCase',
     'capture_stdout',
     'CloudBaseTest',
+    'flaky_filter',
     'Http2Mock',
+    'mark_flaky',
     'RESOURCE_PATH'
 ]

tests/utils.py

@@ -21,7 +21,10 @@
 import tempfile
 import unittest
 
+from flaky import flaky
+import gcloud
 import httplib2
+from nose.plugins.attrib import attr
 from nose.plugins.skip import SkipTest
 from six.moves import cStringIO
 
@@ -40,6 +43,18 @@
 BUCKET_NAME_ENV_VAR = 'TEST_BUCKET_NAME'
 
 
+def flaky_filter(e, *args):
+    exception_class, exception_instance, traceback = e
+    return isinstance(
+        exception_instance,
+        (gcloud.exceptions.GCloudError,))
+
+
+def mark_flaky(f):
+    return flaky(max_runs=3, rerun_filter=flaky_filter)(
+        attr('flaky')(f))
+
+
 class CloudBaseTest(unittest.TestCase):
 
     def setUp(self):

tox.ini

@@ -52,9 +52,8 @@ deps =
 commands = 
     nosetests \
         --exclude-dir=appengine \
-        -a '!slow' \
         {[testenv]commonargs} \
-        {posargs}
+        {posargs:-a '!slow,!flaky'}
 
 [testenv:py34]
 basepython = python3.4
@@ -63,28 +62,25 @@ deps =
 commands = 
     nosetests \
         --exclude-dir=appengine \
-        -a '!slow' \
         {[testenv]commonargs} \
-        {posargs}
+        {posargs:-a '!slow,!flaky'}
 
-[testenv:py27-slow]
+[testenv:py27-all]
 deps =
     {[testenv]deps}
 commands = 
     nosetests \
         --exclude-dir=appengine \
-        -a 'slow' \
         {[testenv]commonargs} \
         {posargs}
 
-[testenv:py34-slow]
+[testenv:py34-all]
 basepython = python3.4
 deps =
     {[testenv]deps}
 commands = 
     nosetests \
         --exclude-dir=appengine \
-        -a 'slow' \
         {[testenv]commonargs} \
         {posargs}