add testcase and changelog for pull:#1020 'improve _created status when switch collection and db'

tests/test_signals.py

@@ -279,5 +279,78 @@ class SignalTests(unittest.TestCase):
         # second time, it must be an update
         self.assertEqual(self.get_signal_output(ei.save), ['Is updated'])
 
+    def test_signals_with_switch_collection(self):
+        ei = self.ExplicitId(id=123)
+        ei.switch_collection("explicit__1")
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+        ei.switch_collection("explicit__1")
+        self.assertEqual(self.get_signal_output(ei.save), ['Is updated'])
+
+        ei.switch_collection("explicit__1", keep_created=False)
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+        ei.switch_collection("explicit__1", keep_created=False)
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+
+    def test_signals_with_switch_db(self):
+        connect('mongoenginetest')
+        register_connection('testdb-1', 'mongoenginetest2')
+
+        ei = self.ExplicitId(id=123)
+        ei.switch_db("testdb-1")
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+        ei.switch_db("testdb-1")
+        self.assertEqual(self.get_signal_output(ei.save), ['Is updated'])
+
+        ei.switch_db("testdb-1", keep_created=False)
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+        ei.switch_db("testdb-1", keep_created=False)
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+
+    def test_signals_with_switch_sharding_db(self):
+
+        import pymongo
+        from mongoengine.connection import get_connection
+
+        connect('mongoenginetest', alias='testdb1')
+        expected_connection = get_connection('testdb1')
+
+        connect('mongoenginetest', alias='testdb2')
+        actual_connection = get_connection('testdb2')
+
+        if pymongo.version_tuple[0] < 3:
+            IS_PYMONGO_3 = False
+        else:
+            IS_PYMONGO_3 = True
+
+        ei = self.ExplicitId(id=123)
+        ei.switch_db("testdb1")
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+        ei.switch_db("testdb1")
+        self.assertEqual(self.get_signal_output(ei.save), ['Is updated'])
+
+        ei.switch_db("testdb2", keep_created=False)
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+        ei.switch_db("testdb2", keep_created=False)
+        self.assertEqual(self.get_signal_output(ei.save), ['Is created'])
+
+        # Handle PyMongo 3+ Async Connection
+        if IS_PYMONGO_3:
+            # Ensure we are connected, throws ServerSelectionTimeoutError otherwise.
+            # Purposely not catching exception to fail test if thrown.
+            expected_connection.server_info()
+
+        self.assertEqual(expected_connection, actual_connection)
+
+
+
+
+
+
+
+
+
+
+
+
 if __name__ == '__main__':
     unittest.main()