Added a sample that demonstrates email migration using multithreading

samples/apps/email_migration_multithreading_test.py

Index: samples/apps/email_migration_multithreading_test.py 
=================================================================== 
new file mode 100644 
--- /dev/null 
+++ b/samples/apps/email_migration_multithreading_test.py 
@@ -0,0 +1,94 @@ 
+#!/usr/bin/python2.4 
+# 
+# Copyright 2012 Google Inc. All Rights Reserved. 
+# 
+# Licensed under the Apache License, Version 2.0 (the "License"); 
+# you may not use this file except in compliance with the License. 
+# You may obtain a copy of the License at 
+# 
+# http://www.apache.org/licenses/LICENSE-2.0 
+# 
+# Unless required by applicable law or agreed to in writing, software 
+# distributed under the License is distributed on an "AS IS" BASIS, 
+# 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. 
+ 
+"""Live tests for EmailMigrationSample.""" 
+ 
+__author__ = 'Gunjan Sharma <gunjansharma@google.com>' 
+ 
+import random 
+import threading 
+import time 
+import unittest 
+import gdata.test_config as conf 
+from email_migration_multithreading import EmailMigrationSample 
+ 
+# Operation name to set shared retry count to 0 
+RESET = 'operation_reset' 
+# Operation name to increment shared retry count 
+INCREMENT = 'operation_increment' 
+ 
+conf.options.register_option(conf.APPS_DOMAIN_OPTION) 
+ 
+ 
+class EmailMigrationSampleTest(unittest.TestCase): 
+ """Sample demonstrating how to migrate emails for a Google Apps user.""" 
+ 
+ def setUp(self): 
+ consumer_key = raw_input('Please enter Consumer Key: ').strip() 
+ consumer_secret = raw_input('Please enter Consumer Secret: ').strip() 
+ domain = raw_input('Please enter domain: ').strip() 
+ self.sample_object = EmailMigrationSample( 
+ consumer_key, consumer_secret, domain) 
+ 
+ def testSampleFunctions(self): 
+ self.Test_CheckUsername() 
+ self.Test_SetLock() 
+ 
+ def Test_CheckUsername(self): 
+ self.assertEqual( 
+ True, self.sample_object._CheckUsername('Allowedusername1234+.\'-_')) 
+ self.assertEqual( 
+ False, self.sample_object._CheckUsername('NotAllowed$#@%^username1')) 
+ self.assertEqual( 
+ False, self.sample_object._CheckUsername('NotAllowed username2')) 
+ self.assertEqual( 
+ False, self.sample_object._CheckUsername( 
+ 'Not_Allowed_username3_very_long_username_' 
+ 'which_has_more_than_64_chracters')) 
+ 
+ def Test_SetLock(self): 
+ self.sample_object.retries = 2 
+ self.sample_object._SetLock(INCREMENT) 
+ self.assertEqual(3, self.sample_object.retries) 
+ 
+ self.sample_object._SetLock(RESET) 
+ self.assertEqual(0, self.sample_object.retries) 
+ 
+ # Test for race condition when using _SetLock 
+ threads = [] 
+ num_threads = 1000 
+ for i in range(1, num_threads + 1): 
+ thread_name = 'Thread-' + str(i) 
+ thread = threading.Thread( 
+ target=self._TestRaceCondition, name=thread_name) 
+ threads.append(thread) 
+ for thread in threads: 
+ thread.start() 
+ for thread in threads: 
+ thread.join() 
+ self.assertEqual(num_threads, self.sample_object.retries) 
+ 
+ def _TestRaceCondition(self): 
+ time.sleep(1 + (random.randint(0, 1000) / 1000)) 
+ self.sample_object._SetLock(INCREMENT) 
+ 
+ 
+def suite(): 
+ return conf.build_suite([EmailMigrationSampleTest]) 
+ 
+ 
+if __name__ == '__main__': 
+ unittest.TextTestRunner().run(suite())