def test_lead_conversion(self):
        """
        Create a Lead object within Salesforce and try to
        convert it, then clean all the generated objects.
        """
        lead = Lead(FirstName="Foo", LastName="Bar", Company="django-salesforce")
        lead.save()

        # Get a status name setting for converted Leads
        cur = connections['salesforce'].cursor()
        cur.execute("SELECT MasterLabel FROM LeadStatus WHERE IsConverted=true")
        converted_status = cur.fetchone()['MasterLabel']

        ret = convert_lead(lead, converted_status=converted_status)
        print("Response from convertLead: " +
                ', '.join('%s: %s' % (k, v) for k, v in sorted(ret.items())))
        expected_names = set(('accountId', 'contactId', 'leadId', 'opportunityId', 'success'))
        self.assertEqual(set(ret), expected_names)
        self.assertEqual(ret['success'], 'true')

        # Cleaning up...
        # Deleting the Account object will also delete the related Contact
        # and Opportunity objects.
        account = Account.objects.get(pk=ret['accountId'])
        account.delete()

        lead.delete()   # FYI, ret['leadId'] == lead.pk

	def test_insert(self):
		"""
		Create a lead record, and make sure it ends up with a valid Salesforce ID.
		"""
		test_lead = Lead(FirstName="User", LastName="Unittest Inserts", Email='[email protected]')
		test_lead.save()
		self.assertEqual(len(test_lead.pk), 18)
		test_lead.delete()

	def test_unicode(self):
		"""
		Make sure weird unicode breaks properly.
		"""
		test_lead = Lead(FirstName=u'\u2603', LastName="Unittest Unicode", Email='[email protected]')
		test_lead.save()
		self.assertEqual(test_lead.FirstName, u'\u2603')
		test_lead.delete()

	def test_delete(self):
		"""
		Create a lead record, then delete it, and make sure it's gone.
		"""
		test_lead = Lead(FirstName="User", LastName="Unittest Deletes", Email='[email protected]')
		test_lead.save()
		test_lead.delete()
		
		self.assertRaises(Lead.DoesNotExist, Lead.objects.get, Email='[email protected]')

 def test_unicode(self):
     """Make sure weird unicode breaks properly.
     """
     test_lead = Lead(FirstName=u'\u2603', LastName="Unittest Unicode",
             Email='[email protected]',
             Company="Some company")
     test_lead.save()
     try:
         self.assertEqual(refresh(test_lead).FirstName, u'\u2603')
     finally:
         test_lead.delete()

 def test_lead_conversion(self):
     """
     Create a Lead object within Salesforce and try to
     convert it, convert/merge it with the information from a duplicit Lead,
     then clean all the generated objects.
     """
     lead = Lead(FirstName="Foo", LastName="Bar", Company="django-salesforce",
                 Street='Test Avenue 45')
     lead.save()
     lead2 = Lead(FirstName="Foo", LastName="Bar", Company="django-salesforce",
                  Phone='123456789')
     lead2.save()
     ret = None
     try:
         # convert the first Lead
         ret = convert_lead(lead, doNotCreateOpportunity=True)
         # print("Response from convertLead: " +
         #       ', '.join('%s: %s' % (k, v) for k, v in sorted(ret.items())))
         expected_names = set(('accountId', 'contactId', 'leadId', 'opportunityId', 'success'))
         self.assertEqual(set(ret), expected_names)
         self.assertEqual(ret['success'], 'true')
         # merge the new Account with the second Lead
         ret2 = convert_lead(lead2, doNotCreateOpportunity=True, accountId=ret['accountId'])
         account = Account.objects.get(pk=ret['accountId'])
         # verify that account is merged
         self.assertEqual(ret2['accountId'], account.pk)
         self.assertEqual(account.BillingStreet, 'Test Avenue 45')
         self.assertEqual(account.Phone, '123456789')
     finally:
         # Cleaning up...
         if ret:
             # Deleting the Account object will also delete the related Contact
             # and Opportunity objects.
             try:
                 account = Account.objects.get(pk=ret['accountId'])
             except Exception:  # pylint:disable=broad-except
                 # this allows to recycle the account even if the queryset code is broken
                 account = Account(pk=ret['accountId'])
                 account._state.db = lead._state.db
             account.delete()
         lead.delete()   # FYI, ret['leadId'] == lead.pk
         lead2.delete()

    def test_lead_conversion(self):
        """
        Create a Lead object within Salesforce and try to
        convert it, then clean all the generated objects.
        """
        lead = Lead(FirstName="Foo", LastName="Bar", Company="django-salesforce")
        lead.save()
        r = convert_lead(lead)
        print("Response from convertLead: " + str(r))
        print("Account ID: " + str(r[0]))
        print("Contact ID: " + str(r[1]))
        print("Opportunity ID: " + str(r[3]))

        # Cleaning up...
        # Deleting the Account object will also delete the related Contact
        # and Opportunity objects.
        account = Account.objects.get(pk=str(r[0]))
        account.delete()
        
        lead.delete()   # FYI, r[2] == lead.pk

	def setUp(self):
		"""
		Create our test lead record.
		"""
		self.test_lead = Lead(
			FirstName	= "User",
			LastName	= "Unittest General",
			Email		= test_email,
			Status		= 'Open',
		)
		self.test_lead.save()

    def test_filter_by_more_fk_to_the_same_model_subquery(self):
        """Test a filter with more relations to the same model.

        Verify that aliases are correctly decoded in the query compiler.
        """
        test_lead = Lead(Company='sf_test lead', LastName='name')
        test_lead.save()
        test_task = Task(who=test_lead)
        test_task.save()
        try:
            qs = Task.objects.filter(who__in=Lead.objects.filter(pk=test_lead.pk,
                    owner__Username=current_user,
                    last_modified_by__Username=current_user))
            sql, params = qs.query.get_compiler('salesforce').as_sql()
            self.assertRegexpMatches(sql, 'SELECT Task.Id, .* FROM Task WHERE Task.WhoId IN \(SELECT ')
            self.assertIn('Lead.Owner.Username = %s', sql)
            self.assertIn('Lead.LastModifiedBy.Username = %s', sql)
            refreshed_lead = qs[:1]
            self.assertEqual(len(refreshed_lead), 1)
        finally:
            test_task.delete()
            test_lead.delete()

    def test_filter_by_more_fk_to_the_same_model(self):
        """Test a filter with more relations to the same model.

        Verify that aliases are correctly decoded in the query compiler.
        """
        test_lead = Lead(Company='sf_test lead', LastName='name')
        test_lead.save()
        try:
            qs = Lead.objects.filter(pk=test_lead.pk,
                    owner__Username=current_user,
                    last_modified_by__Username=current_user)
            # Verify that a coplicated analysis is not performed on old Django
            # so that the query can be at least somehow simply compiled to SOQL
            # without exceptions, in order to prevent regressions.
            sql, params = qs.query.get_compiler('salesforce').as_sql()
            # Verify expected filters in SOQL compiled by new Django
            self.assertIn('Lead.Owner.Username = %s', sql)
            self.assertIn('Lead.LastModifiedBy.Username = %s', sql)
            # verify validity for SFDC, verify results
            refreshed_lead = qs.get()
            self.assertEqual(refreshed_lead.pk, test_lead.pk)
        finally:
            test_lead.delete()

	def setUp(self):
		"""
		Create our test lead record.
		"""
		def add_obj(obj):
			obj.save()
			self.objs.append(obj)
		#
		self.test_lead = Lead(
			FirstName	= "User",
			LastName	= "Unittest General",
			Email		= test_email,
			Status		= 'Open',
			Company = "Some company, Ltd.",
		)
		self.objs = []
		self.test_lead.save()
		if not default_is_sf:
			add_obj(Contact(LastName='Test contact 1'))
			add_obj(Contact(LastName='Test contact 2'))
			add_obj(User(Username=current_user))

 def setUp(self):
     """Create our test lead record.
     """
     def add_obj(obj):
         obj.save()
         self.objs.append(obj)
     #
     self.test_lead = Lead(
         FirstName="User" + uid,
         LastName="Unittest General",
         Email=test_email,
         Status='Open',
         Company="Some company, Ltd.",
     )
     self.objs = []
     self.test_lead.save()
     # This is only for demonstration that some test can be run even with
     # non SFDC database SALESFORCE_DB_ALIAS, even if the test expects some
     # contacts and the current user, but most of tests can pass only on SFDC.
     if not default_is_sf:
         add_obj(Contact(last_name='Test contact 1'))
         add_obj(Contact(last_name='Test contact 2'))
         add_obj(User(Username=current_user))

class BasicSOQLTest(TestCase):
	def setUp(self):
		"""
		Create our test lead record.
		"""
		self.test_lead = Lead(
			FirstName	= "User",
			LastName	= "Unittest General",
			Email		= test_email,
			Status		= 'Open',
			Company = "Some company, Ltd.",
		)
		self.test_lead.save()
	
	def tearDown(self):
		"""
		Clean up our test lead record.
		"""
		self.test_lead.delete()


	def test_raw(self):
		"""
		Get the first two contact records.
		"""
		contacts = Contact.objects.raw(
				"SELECT Id, LastName, FirstName FROM Contact "
				"LIMIT 2")
		self.assertEqual(len(contacts), 2)
		'%s' % contacts[0].__dict__  # Check that all fields are accessible
	
	def test_raw_foreignkey_id(self):
		"""
		Get the first two contacts by raw query with a ForeignKey id field.
		"""
		contacts = Contact.objects.raw(
				"SELECT Id, LastName, FirstName, OwnerId FROM Contact "
				"LIMIT 2")
		self.assertEqual(len(contacts), 2)
		'%s' % contacts[0].__dict__  # Check that all fields are accessible
		self.assertIn('@', contacts[0].Owner.Email)
	
	def test_select_all(self):
		"""
		Get the first two contact records.
		"""
		contacts = Contact.objects.all()[0:2]
		self.assertEqual(len(contacts), 2)

	def test_exclude_query_construction(self):
		"""
		Test that exclude query construction returns valid SOQL.
		"""
		contacts = Contact.objects.filter(FirstName__isnull=False).exclude(Email="[email protected]", LastName="Wozniak").exclude(LastName="smith")
		number_of_contacts = contacts.count()
		self.assertIsInstance(number_of_contacts, int)

	def test_foreign_key(self):
		"""
		Verify that the owner of an Contact is the currently logged admin.
		"""
		current_sf_user = User.objects.get(Username=current_user)
		contact = Contact.objects.filter(Owner=current_sf_user)[0]
		user = contact.Owner
		# This user can be e.g. '[email protected]'.
		self.assertEqual(user.Username, current_user)
	
	def test_update_date(self):
		"""
		Test updating a date.
		"""
		now = round_datetime_utc(datetime.datetime.utcnow())
		contact = Contact.objects.all()[0]
		old_date = contact.EmailBouncedDate
		contact.EmailBouncedDate = now.replace(tzinfo=pytz.utc)
		contact.save()
		try:
			self.assertEqual(refresh(contact).EmailBouncedDate, now)
		finally:
			contact.EmailBouncedDate = old_date
			contact.save()
		self.assertEqual(refresh(contact).EmailBouncedDate, old_date)
	
	def test_insert_date(self):
		"""
		Test inserting a date.
		"""
		now = round_datetime_utc(datetime.datetime.utcnow())
		contact = Contact(
				FirstName = 'Joe',
				LastName = 'Freelancer',
				EmailBouncedDate=now.replace(tzinfo=pytz.utc))
		contact.save()
		try:
			self.assertEqual(refresh(contact).EmailBouncedDate, now)
		finally:
			contact.delete()

	def test_default_specified_by_sf(self):
		"""
#.........这里部分代码省略.........

class BasicSOQLTest(TestCase):
	def setUp(self):
		"""
		Create our test lead record.
		"""
		def add_obj(obj):
			obj.save()
			self.objs.append(obj)
		#
		self.test_lead = Lead(
			FirstName	= "User",
			LastName	= "Unittest General",
			Email		= test_email,
			Status		= 'Open',
			Company = "Some company, Ltd.",
		)
		self.objs = []
		self.test_lead.save()
		if not default_is_sf:
			add_obj(Contact(LastName='Test contact 1'))
			add_obj(Contact(LastName='Test contact 2'))
			add_obj(User(Username=current_user))
	
	def tearDown(self):
		"""
		Clean up our test records.
		"""
		if self.test_lead.pk is not None:
			self.test_lead.delete()
		for obj in self.objs:
			if obj.pk is not None:
				obj.delete()
		self.objs = []


	@skipUnless(default_is_sf, "Default database should be any Salesforce.")
	def test_raw(self):
		"""
		Get the first two contact records.
		(At least 3 manually created Contacts must exist before these read-only tests.)
		"""
		contacts = Contact.objects.raw(
				"SELECT Id, LastName, FirstName FROM Contact "
				"LIMIT 2")
		self.assertEqual(len(contacts), 2)
		# It had a side effect that the same assert failed second times.
		self.assertEqual(len(contacts), 2)
		'%s' % contacts[0].__dict__  # Check that all fields are accessible

	@skipUnless(default_is_sf, "Default database should be any Salesforce.")
	def test_raw_foreignkey_id(self):
		"""
		Get the first two contacts by raw query with a ForeignKey id field.
		"""
		contacts = Contact.objects.raw(
				"SELECT Id, LastName, FirstName, OwnerId FROM Contact "
				"LIMIT 2")
		self.assertEqual(len(contacts), 2)
		'%s' % contacts[0].__dict__  # Check that all fields are accessible
		self.assertIn('@', contacts[0].owner.Email)

	def test_select_all(self):
		"""
		Get the first two contact records.
		"""
		contacts = Contact.objects.all()[0:2]
		self.assertEqual(len(contacts), 2)

	def test_exclude_query_construction(self):
		"""
		Test that exclude query construction returns valid SOQL.
		"""
		contacts = Contact.objects.filter(first_name__isnull=False).exclude(email="[email protected]", last_name="Wozniak").exclude(last_name="smith")
		number_of_contacts = contacts.count()
		self.assertIsInstance(number_of_contacts, int)
		# the default self.test_lead shouldn't be excluded by only one nondition
		leads = Lead.objects.exclude(Email="[email protected]", LastName="Unittest General").filter(FirstName="User", LastName="Unittest General")
		self.assertEqual(leads.count(), 1)

	@skipUnless(default_is_sf, "Default database should be any Salesforce.")
	def test_foreign_key(self):
		"""
		Verify that the owner of an Contact is the currently logged admin.
		"""
		current_sf_user = User.objects.get(Username=current_user)
		test_contact = Contact(first_name = 'sf_test', last_name='my')
		test_contact.save()
		try:
			contact = Contact.objects.filter(owner=current_sf_user)[0]
			user = contact.owner
			# This user can be e.g. '[email protected]'.
			self.assertEqual(user.Username, current_user)
		finally:
			test_contact.delete()

	def test_foreign_key_column(self):
		"""
		Verify filtering by a column of related parent object.
		"""
		test_account = Account(Name = 'sf_test account')
#.........这里部分代码省略.........

class BasicLeadSOQLTest(TestCase):
    """Tests that use a test Lead"""

    def setUp(self):
        """Create our test lead record.
        """
        def add_obj(obj):
            obj.save()
            self.objs.append(obj)
        #
        self.test_lead = Lead(
            FirstName="User" + uid,
            LastName="Unittest General",
            Email=test_email,
            Status='Open',
            Company="Some company, Ltd.",
        )
        self.objs = []
        self.test_lead.save()
        # This is only for demonstration that some test can be run even with
        # non SFDC database SALESFORCE_DB_ALIAS, even if the test expects some
        # contacts and the current user, but most of tests can pass only on SFDC.
        if not default_is_sf:
            add_obj(Contact(last_name='Test contact 1'))
            add_obj(Contact(last_name='Test contact 2'))
            add_obj(User(Username=current_user))

    def tearDown(self):
        """Clean up our test records.
        """
        if self.test_lead.pk is not None:
            self.test_lead.delete()
        for obj in self.objs:
            if obj.pk is not None:
                obj.delete()
        self.objs = []


    def test_exclude_query_construction(self):
        """Test that exclude query construction returns valid SOQL.
        """
        contacts = Contact.objects.filter(first_name__isnull=False
                ).exclude(email="[email protected]", last_name="Wozniak").exclude(last_name="smith")
        number_of_contacts = contacts.count()
        self.assertIsInstance(number_of_contacts, int)
        # the default self.test_lead shouldn't be excluded by only one nondition
        leads = Lead.objects.exclude(Email="[email protected]", LastName="Unittest General"
                ).filter(FirstName="User" + uid, LastName="Unittest General")
        self.assertEqual(leads.count(), 1)

    def test_get(self):
        """Get the test lead record.
        """
        lead = Lead.objects.get(Email=test_email)
        self.assertEqual(lead.FirstName, 'User' + uid)
        self.assertEqual(lead.LastName, 'Unittest General')
        if not default_is_sf:
            self.skipTest("Default database should be any Salesforce.")
        # test a read only field (formula of full name)
        self.assertEqual(lead.Name, 'User%s Unittest General' % uid)

    def test_not_null(self):
        """Get the test lead record by isnull condition.
        """
        lead = Lead.objects.get(Email__isnull=False, FirstName='User' + uid)
        self.assertEqual(lead.FirstName, 'User' + uid)
        self.assertEqual(lead.LastName, 'Unittest General')

    def test_update(self):
        """Update the test lead record.
        """
        test_lead = Lead.objects.get(Email=test_email)
        self.assertEqual(test_lead.FirstName, 'User' + uid)
        test_lead.FirstName = 'Tested'
        test_lead.save()
        self.assertEqual(refresh(test_lead).FirstName, 'Tested')

    def test_save_update_fields(self):
        """Test the save method with parameter `update_fields`

        that updates only required fields.
        """
        company_orig = self.test_lead.Company
        self.test_lead.Company = 'nonsense'
        self.test_lead.FirstName = 'John'
        self.test_lead.save(update_fields=['FirstName'])
        test_lead = refresh(self.test_lead)
        self.assertEqual(test_lead.FirstName, 'John')
        self.assertEqual(test_lead.Company, company_orig)

    def test_query_all_deleted(self):
        """Test query for deleted objects (queryAll resource).
        """
        self.test_lead.delete()
        # TODO optimize counting because this can load thousands of records
        count_deleted = Lead.objects.db_manager(sf_alias).query_all(
                ).filter(IsDeleted=True, LastName="Unittest General").count()
        if not default_is_sf:
            self.skipTest("Default database should be any Salesforce.")
        self.assertGreaterEqual(count_deleted, 1)
#.........这里部分代码省略.........

class BasicSOQLTest(TestCase):
	def setUp(self):
		"""
		Create our test lead record.
		"""
		self.test_lead = Lead(
			FirstName	= "User",
			LastName	= "Unittest General",
			Email		= test_email,
			Status		= 'Open',
		)
		self.test_lead.save()
	
	def tearDown(self):
		"""
		Clean up our test lead record.
		"""
		self.test_lead.delete()
	
	def test_select_all(self):
		"""
		Get the first five account records.
		"""
		accounts = Account.objects.all()[0:5]
		self.assertEqual(len(accounts), 5)
	
	def test_select_all(self):
		"""
		Get the first five account records.
		"""
		accounts = Account.objects.all()[0:5]
		self.assertEqual(len(accounts), 5)
	
	def test_foreign_key(self):
		account = Account.objects.all()[0]
		user = account.Owner
		self.assertEqual(user.Email, '[email protected]')
	
	def test_update_date(self):
		"""
		Test updating a date.
		"""
		self.skipTest("Need to find a suitable *standard* model field to test datetime updates.")
		
		account = Account.objects.all()[0]
		account.LastLogin = now = datetime.datetime.now()
		account.save()
		
		saved = Account.objects.get(pk=account.pk)
		self.assertEqual(account.LastLogin, now)
	
	def test_insert_date(self):
		"""
		Test inserting a date.
		"""
		self.skipTest("Need to find a suitable *standard* model field to test datetime inserts.")
		
		now = datetime.datetime.now()
		account = Account(
			FirstName = 'Joe',
			LastName = 'Freelancer',
			LastLogin = now,
			IsPersonAccount = False,
		)
		account.save()
		
		saved = Account.objects.get(pk=account.pk)
		self.assertEqual(saved.LastLogin, now)
		self.assertEqual(saved.IsPersonAccount, False)
		
		saved.delete()
	
	def test_get(self):
		"""
		Get the test lead record.
		"""
		lead = Lead.objects.get(Email=test_email)
		self.assertEqual(lead.FirstName, 'User')
		self.assertEqual(lead.LastName, 'Unittest General')
	
	def test_not_null(self):
		"""
		Get the test lead record.
		"""
		lead = Lead.objects.get(Email__isnull=False)
		self.assertEqual(lead.FirstName, 'User')
		self.assertEqual(lead.LastName, 'Unittest General')
	
	def test_unicode(self):
		"""
		Make sure weird unicode breaks properly.
		"""
		test_lead = Lead(FirstName=u'\u2603', LastName="Unittest Unicode", Email='[email protected]')
		test_lead.save()
		self.assertEqual(test_lead.FirstName, u'\u2603')
		test_lead.delete()
	
	def test_date_comparison(self):
		"""
		Test that date comparisons work properly.
#.........这里部分代码省略.........