def test_requeue_weekly_saved_search(self):
        """
        Tests that weekly saved searches are requeued correctly individually in
        addition to as part of a digest.
        """
        today = datetime.date.today()
        two_days_ago = today.isoweekday() - 2
        two_month_days_ago = today.day - 2
        if two_days_ago <= 0:
            # Dates can't be negative or zero
            two_days_ago += 7
        if two_month_days_ago == 0:
            # According to mysearches/models.py, we can't have saved searches on
            # the 31st. If two_month_days_ago==0, this would be the 31st.
            two_days_ago -= 1

        digest = SavedSearchDigestFactory(user=self.user, is_active=True)
        search = SavedSearchFactory(user=self.user, is_active=True,
                                    frequency='W', day_of_week=two_days_ago)

        self.requeue(search, digest)

        digest.is_active = False
        digest.save()
        search.last_sent = None
        search.save()
        mail.outbox = []

        self.requeue(search, digest)

    def test_requeue_monthly_saved_search(self):
        """
        Tests that monthly saved searches are requeued correctly individually
        in addition to as part of a digest.
        """
        today = datetime.date.today()
        two_days_ago = today.day - 2
        if two_days_ago < 0:
            two_days_ago += 31
        elif two_days_ago == 0:
            two_days_ago = 30

        digest = SavedSearchDigestFactory(user=self.user, is_active=True)
        search = SavedSearchFactory(user=self.user, is_active=True,
                                    frequency='M', day_of_month=two_days_ago)

        self.requeue(search, digest)

        digest.is_active = False
        digest.save()
        search.last_sent = None
        search.save()
        mail.outbox = []

        self.requeue(search, digest)

 def test_inactive_user_receives_saved_search(self):
     self.assertEqual(len(mail.outbox), 0)
     self.user.is_active = False
     self.user.save()
     saved_search = SavedSearchFactory(user=self.user)
     saved_search.send_email()
     self.assertEqual(len(mail.outbox), 1)

    def test_email_contains_activate_link(self):
        search = SavedSearchFactory(user=self.user)
        self.assertTrue(self.user.is_active)
        search.send_email()
        email = mail.outbox.pop()
        self.assertFalse('activate your account' in email.body)

        self.user.is_active = False
        self.user.save()
        search.send_email()
        email = mail.outbox.pop()
        self.assertTrue('activate your account' in email.body)

    def test_unicode_in_saved_search(self):
        """Tests that saved search urls with unicode don't cause errors."""
        search = SavedSearchFactory(
            user=self.user,
            url=u"warehouse.jobs/search?location=Roswell%2C+GA&q=Delivery+I"
                "+%E2%80%93+Material+Handler%2FDriver+Helper+%E2%80%93+3rd"
                "+Shift%2C+Part-time")

        try:
            search.send_email()
        except UnicodeEncodeError as e:
            self.fail(e)

    def test_send_initial_email(self):
        search = SavedSearchFactory(user=self.user, is_active=False,
                                    url='www.my.jobs/search?q=new+search')
        search.send_initial_email()
        self.assertEqual(len(mail.outbox), 1)

        email = mail.outbox.pop()
        self.assertEqual(email.from_email, settings.SAVED_SEARCH_EMAIL)
        self.assertEqual(email.to, [self.user.email])
        self.assertEqual("My.jobs New Saved Search" in email.subject, True)
        self.assertTrue("table" in email.body)
        self.assertTrue(email.to[0] in email.body)

    def test_email_profile_completion(self):

        search = SavedSearchFactory(user=self.user)
        search.send_email()
        self.assertEqual(len(settings.PROFILE_COMPLETION_MODULES), 6)
        self.assert_modules_in_hrefs(settings.PROFILE_COMPLETION_MODULES)

        PrimaryNameFactory(user=self.user)

        search.send_email()

        new_modules = [module for module in settings.PROFILE_COMPLETION_MODULES
                       if module != 'name']
        self.assertEqual(len(new_modules), 5)
        self.assert_modules_in_hrefs(new_modules)

    def test_send_update_email(self):
        search = SavedSearchFactory(user=self.user, is_active=False,
                                    url='www.my.jobs/search?q=new+search')
        search.send_update_email('Your search is updated')
        self.assertEqual(len(mail.outbox), 1)
        log = SavedSearchLog.objects.get()
        self.assertTrue('Jobs are not sent' in log.reason)
        self.assertTrue(log.was_sent)

        email = mail.outbox.pop()
        self.assertEqual(email.from_email, 'My.jobs Saved Search <[email protected]>')
        self.assertEqual(email.to, [self.user.email])
        self.assertEqual("My.jobs Saved Search Updated" in email.subject, True)
        self.assertTrue("table" in email.body)
        self.assertTrue("Your search is updated" in email.body)
        self.assertTrue(email.to[0] in email.body)

    def test_saved_search_invitation_context(self):
        """
        Test that creating an invitation context from a saved search instance
        produces a context with a message, the saved search itself, an unsent
        initial email, and whether that saved search is text only.

        """
        saved_search = SavedSearchFactory(user=self.user, text_only=True)
        context = invitation_context(saved_search)

        self.assertEqual(context, {
            "message": " in order to begin receiving their available job "
                       "opportunities on a regular basis.",
            "saved_search": saved_search,
            "initial_search_email": saved_search.initial_email(send=False),
            "text_only": saved_search.text_only})

 def test_get_unsent_jobs(self):
     """
     When sending a saved search email, we should retrieve all new jobs since
     last send, not all new jobs based on frequency.
     """
     self.patcher.stop()
     self.patcher = patch('urllib2.urlopen',
                          return_file(time_=datetime.datetime.now() -
                                      datetime.timedelta(days=3)))
     self.mock_urlopen = self.patcher.start()
     last_sent = datetime.datetime.now() - datetime.timedelta(days=3)
     search = SavedSearchFactory(frequency='D',
                                 last_sent=last_sent,
                                 user=self.user,
                                 email=self.user.email)
     search.send_email()
     self.assertEqual(len(mail.outbox), 1)

    def test_saved_search_new_job_indicator(self):
        """
        Partner saved searches should include indicators for unseen jobs, while
        job seeker saved searches should not.
        """
        new_job_indicator = '>New! <'
        search = SavedSearchFactory(user=self.user)
        search.send_email()
        search_email = mail.outbox.pop()
        new_jobs = self.num_occurrences(search_email.body,
                                        new_job_indicator)
        self.assertEqual(len(new_jobs), 0)

        self.partner_search.send_email()
        partner_search_email = mail.outbox.pop()
        new_jobs = self.num_occurrences(partner_search_email.body,
                                        new_job_indicator)
        self.assertEqual(len(new_jobs), 1)

    def test_initial_email(self):
        search = SavedSearchFactory(user=self.user, is_active=False,
                                    url='www.my.jobs/search?q=new+search')
        search.initial_email()
        self.assertEqual(len(mail.outbox), 1)
        self.assertEqual(SavedSearchLog.objects.count(), 1)
        log = SavedSearchLog.objects.get()
        self.assertTrue('Jobs are not sent' in log.reason)
        self.assertTrue(log.was_sent)

        email = mail.outbox.pop()
        self.assertEqual(email.from_email, 'My.jobs Saved Search <[email protected]>')
        self.assertEqual(email.to, [self.user.email])
        self.assertEqual("My.jobs New Saved Search" in email.subject, True)
        self.assertTrue("table" in email.body)
        self.assertTrue(email.to[0] in email.body)
        # Search url appears twice - once displaying the saved search source
        # and once in the view link.
        self.assertEqual(email.body.count(search.url), 2)

    def test_requeue_monthly_saved_search(self):
        """
        Tests that monthly saved searches are requeued correctly individually
        in addition to as part of a digest.
        """
        today = datetime.date.today().day
        last_week = today - 7
        if last_week <= 0:
            last_week += 31
        digest = SavedSearchDigestFactory(user=self.user,
                                          is_active=True)
        search = SavedSearchFactory(user=self.user, is_active=True,
                                    frequency='M', day_of_month=last_week)

        self.requeue(search, digest)

        digest.is_active = False
        digest.save()
        search.last_sent = None
        search.save()
        mail.outbox = []

        self.requeue(search, digest)

    def setUp(self):
        super(CronEventTests, self).setUp()
        # I'm using SavedSearch for these tests, but you can use any
        # model with a DateTimeField for the "*_with_field" option, and
        # any object for the "*_no_field" option.
        # It might actually be worth using additional object types
        # in future tests.
        yesterday = datetime.now() - timedelta(1)
        self.saved_search = SavedSearchFactory(last_sent=yesterday)

        model = self.saved_search._meta.model
        self.saved_search_contenttype = ContentType.objects.get_for_model(model)
        cron_kwargs = {'model': self.saved_search_contenttype}
        self.cron_event_no_field = factories.CronEventFactory(**cron_kwargs)
        cron_kwargs['field'] = 'last_sent'
        self.cron_event_with_field = factories.CronEventFactory(**cron_kwargs)

 def setUp(self):
     super(SavedSearchSendingTests, self).setUp()
     self.feed = 'http://rushenterprises-veterans.jobs/alabama/usa/jobs/feed/rss'
     self.saved_search = SavedSearchFactory(user=self.user, feed=self.feed,
                                            frequency='D')
     self.company = CompanyFactory()
     self.partner = PartnerFactory(owner=self.company)
     self.contact = ContactFactory(user=self.user,
                                   partner=self.partner)
     self.partner_search = PartnerSavedSearchFactory(user=self.user,
                                                     feed=self.feed,
                                                     frequency='D',
                                                     created_by=self.user,
                                                     provider=self.company,
                                                     partner=self.partner)
     mail.outbox = []

    def test_saved_search_all_jobs_link(self):
        search = SavedSearchFactory(user=self.user)
        search.send_email()

        email = mail.outbox.pop()
        # When search.url does not start with my.jobs, use it as the all jobs
        # link
        self.assertFalse(search.url.startswith('http://my.jobs'))
        self.assertNotEqual(email.body.find(search.url), -1)

        # When search.url starts with my.jobs, strip /feed/rss from search.feed
        # if it exists and use that as the all jobs link
        search.url = 'http://my.jobs/' + '1'*32
        search.save()
        search.send_email()
        email = mail.outbox.pop()
        self.assertEqual(email.body.find(search.url),
                         -1)
        self.assertNotEqual(email.body.find(search.feed.replace('/feed/rss', '')),
                            -1)

class SavedSearchSendingTests(MyJobsBase):
    def setUp(self):
        super(SavedSearchSendingTests, self).setUp()
        self.feed = 'http://rushenterprises-veterans.jobs/alabama/usa/jobs/feed/rss'
        self.user = UserFactory()
        self.saved_search = SavedSearchFactory(user=self.user, feed=self.feed,
                                               frequency='D')
        self.company = CompanyFactory()
        self.partner = PartnerFactory(owner=self.company)
        self.contact = ContactFactory(user=self.user,
                                      partner=self.partner)
        self.partner_search = PartnerSavedSearchFactory(user=self.user,
                                                        feed=self.feed,
                                                        frequency='D',
                                                        created_by=self.user,
                                                        provider=self.company,
                                                        partner=self.partner)
        mail.outbox = []

    @patch('urllib2.urlopen')
    def test_all_jobs_new(self, urlopen_mock):
        mock_obj = Mock()
        mock_obj.read.side_effect = [jobs, jobs, jobs]
        urlopen_mock.return_value = mock_obj

        three_days_ago = datetime.datetime.now() - datetime.timedelta(days=365)
        self.partner_search.last_sent = three_days_ago
        self.saved_search.last_sent = three_days_ago
        self.partner_search.save()
        self.saved_search.save()

        # All of the jobs were sent within the past year, so if we set
        # last_sent to one year ago all of the jobs should be old.
        self.saved_search.send_email()
        email = mail.outbox.pop()
        self.assertIn('Showing the top 3 jobs', email.body)

        self.partner_search.send_email()
        email = mail.outbox.pop()
        self.assertIn('Showing the top 3 jobs', email.body)

    @patch('urllib2.urlopen')
    def test_some_jobs_new(self, urlopen_mock):
        mock_obj = Mock()
        mock_obj.read.side_effect = [jobs, jobs, jobs]
        urlopen_mock.return_value = mock_obj

        three_days_ago = datetime.datetime.now() - datetime.timedelta(days=3)
        self.partner_search.last_sent = three_days_ago
        self.saved_search.last_sent = three_days_ago
        self.partner_search.save()
        self.saved_search.save()

        # One job was sent within the past 3 days, so if we set last_sent to
        # three days ago one of the jobs should be old.
        self.saved_search.send_email()
        email = mail.outbox.pop()
        self.assertIn('Showing the top job', email.body)

        self.partner_search.send_email()
        email = mail.outbox.pop()
        self.assertIn('Showing the top 3 jobs', email.body)

    @patch('urllib2.urlopen')
    def test_no_jobs_new(self, urlopen_mock):
        mock_obj = Mock()
        mock_obj.read.side_effect = [jobs, jobs, jobs]
        urlopen_mock.return_value = mock_obj

        self.partner_search.last_sent = datetime.datetime.now()
        self.saved_search.last_sent = datetime.datetime.now()
        self.partner_search.save()
        self.saved_search.save()

        # All jobs were sent over 2 days ago, so if we set last_sent to
        # today none of the jobs should be old.
        self.saved_search.send_email()
        self.assertEqual(len(mail.outbox), 0)

        self.partner_search.send_email()
        email = mail.outbox.pop()
        self.assertIn('Showing the top 3 jobs', email.body)

    @patch('urllib2.urlopen')
    def test_partner_saved_search_backfill(self, urlopen_mock):
        mock_obj = Mock()
        mock_obj.read.side_effect = [jobs, jobs, jobs, jobs, jobs, jobs]
        urlopen_mock.return_value = mock_obj

        # Make it so there should be no new jobs.
        self.partner_search.last_sent = datetime.datetime.now()
        self.partner_search.save()

        # jobs_per_email is default, so all 3 should get sent.
        self.partner_search.send_email()
        email = mail.outbox.pop()
        self.assertIn('Showing the top 3 jobs', email.body)

        self.partner_search.jobs_per_email = 2
        self.partner_search.save()
#.........这里部分代码省略.........

    def test_saved_search_no_jobs(self):
        search = SavedSearchFactory(feed='http://google.com', user=self.user)
        search.send_email()

        self.assertEqual(len(mail.outbox), 0)

class CronEventTests(MyJobsBase):
    def setUp(self):
        super(CronEventTests, self).setUp()
        # I'm using SavedSearch for these tests, but you can use any
        # model with a DateTimeField for the "*_with_field" option, and
        # any object for the "*_no_field" option.
        # It might actually be worth using additional object types
        # in future tests.
        yesterday = datetime.now() - timedelta(1)
        self.saved_search = SavedSearchFactory(last_sent=yesterday)

        model = self.saved_search._meta.model
        self.saved_search_contenttype = ContentType.objects.get_for_model(model)
        cron_kwargs = {'model': self.saved_search_contenttype}
        self.cron_event_no_field = factories.CronEventFactory(**cron_kwargs)
        cron_kwargs['field'] = 'last_sent'
        self.cron_event_with_field = factories.CronEventFactory(**cron_kwargs)

    def test_cron_event_schedule_task_no_field(self):
        today = datetime.now().date()
        task = self.cron_event_no_field.schedule_task(self.saved_search)

        self.assertEqual(task.object_id, self.saved_search.id)
        self.assertEqual(task.object_model, self.saved_search_contenttype)
        self.assertEqual(task.related_event, self.cron_event_no_field)
        self.assertEqual(task.scheduled_for.date(), today)
        self.assertEqual(task.scheduled_at.date(), today)
        self.assertIsNone(task.completed_on)

    def test_cron_event_schedule_task_with_field(self):
        yesterday = (datetime.now() - timedelta(1)).date()
        today = datetime.now().date()
        task = self.cron_event_with_field.schedule_task(self.saved_search)

        self.assertEqual(task.object_id, self.saved_search.id)
        self.assertEqual(task.object_model, self.saved_search_contenttype)
        self.assertEqual(task.related_event, self.cron_event_with_field)
        self.assertEqual(task.scheduled_for.date(), yesterday)
        self.assertEqual(task.scheduled_at.date(), today)
        self.assertIsNone(task.completed_on)

    def test_scheduled_for_with_field(self):
        """
        CronEvents that have an associated field should schedule from the
        time contained within the associated field.
        
        """
        tomorrow = datetime.now() + timedelta(1)
        self.saved_search.last_sent = tomorrow
        self.saved_search.save()

        self.cron_event_with_field.minutes = 60*25
        self.cron_event_with_field.save()
        should_be_scheduled_for = (tomorrow + timedelta(1)).date()

        task = self.cron_event_with_field.schedule_task(self.saved_search)
        self.assertEqual(task.scheduled_for.date(), should_be_scheduled_for)

    def test_scheduled_for_no_field(self):
        """
        CronEvents that don't have an associated field should instead be
        scheduled based on the current time.

        """
        self.cron_event_no_field.minutes = 60
        self.cron_event_no_field.save()
        should_be_scheduled_for = datetime.now() + timedelta(minutes=60)
        should_be_scheduled_for = should_be_scheduled_for.date()

        task = self.cron_event_no_field.schedule_task(self.saved_search)
        self.assertEqual(task.scheduled_for.date(), should_be_scheduled_for)