python类timedelta()的实例源码

def test_post_redirects_when_user_has_already_applied(self):
        self.app_info.start_date = timezone.now().date()
        self.app_info.end_date = timezone.now().date() + timezone.timedelta(days=2)
        self.app_info.save()

        data = {
            'phone': faker.phone_number(),
            'works_at': faker.job(),
            'skype': faker.word()
        }

        create_application(user=self.user, application_info=self.app_info, skype=data.get('skype'))
        with self.login(email=self.user.email, password=self.test_password):
            response = self.post(self.url, data=data)
            expected_url = reverse('dashboard:applications:user-applications')
            self.assertRedirects(response,
                                 expected_url=expected_url)

def test_sends_mail_to_address_upon_successful_application(self, mock_send_mail):
        self.app_info.start_date = timezone.now().date()
        self.app_info.end_date = timezone.now().date() + timezone.timedelta(days=2)
        self.app_info.save()

        data = {
            'phone': faker.phone_number(),
            'works_at': faker.job(),
            'skype': faker.word
        }
        with self.login(email=self.user.email, password=self.test_password):
            response = self.post(self.url, data=data)
            self.assertRedirects(response, expected_url=reverse('dashboard:applications:user-applications'))
            self.assertEqual(mock_send_mail.called, True)
            (template_name, recipients, context), kwargs = mock_send_mail.call_args
            self.assertEqual([self.user.email], recipients)

def setUp(self):
        self.test_password = faker.password()
        self.user = BaseUserFactory(password=self.test_password)
        self.user.is_active = True
        self.user.save()
        self.course = CourseFactory(
            start_date=timezone.now() + timezone.timedelta(days=10),
            end_date=timezone.now() + timezone.timedelta(days=20)
        )
        self.app_info = ApplicationInfoFactory(
            start_date=timezone.now(),
            end_date=timezone.now() + timezone.timedelta(days=4),
            start_interview_date=timezone.now() + timezone.timedelta(days=5),
            end_interview_date=timezone.now() + timezone.timedelta(days=6),
            course=self.course
        )
        self.application = ApplicationFactory(application_info=self.app_info, user=self.user)
        self.url = reverse('dashboard:applications:user-applications')

def setUp(self):
        self.test_password = faker.password()
        self.user = BaseUserFactory(password=self.test_password)
        self.user.is_active = True
        self.user.save()
        self.course = CourseFactory(
            start_date=timezone.now() + timezone.timedelta(days=10),
            end_date=timezone.now() + timezone.timedelta(days=20)
        )
        self.app_info = ApplicationInfoFactory(
            start_date=timezone.now(),
            end_date=timezone.now() + timezone.timedelta(days=4),
            start_interview_date=timezone.now() + timezone.timedelta(days=5),
            end_interview_date=timezone.now() + timezone.timedelta(days=6),
            course=self.course
        )
        self.application = ApplicationFactory(application_info=self.app_info, user=self.user)
        self.url = reverse('dashboard:applications:edit-application',
                           kwargs={
                               'course_id': self.course.id
                           })

def setUp(self):
        self.start_date = timezone.now().date()
        self.end_date = timezone.now().date() + timezone.timedelta(days=2)
        self.open_applications = ApplicationInfoFactory.create_batch(size=5,
                                                                     start_date=self.start_date,
                                                                     end_date=self.end_date,
                                                                     start_interview_date=self.start_date,
                                                                     end_interview_date=self.end_date)

        false_start_date = timezone.now().date() - timezone.timedelta(days=2)
        false_end_date = timezone.now().date() - timezone.timedelta(days=1)
        self.closed_applications = ApplicationInfoFactory.create_batch(
                                      size=5,
                                      start_date=false_start_date,
                                      end_date=false_end_date,
                                      start_interview_date=false_start_date,
                                      end_interview_date=false_end_date)

def test_set_next_notification_date(self, now_mock):
        """ Notifications: Test if next notification date is set """
        now_mock.return_value = timezone.make_aware(
            timezone.datetime(2016, 11, 16))

        now = timezone.localtime(timezone.now())
        tomorrow = (timezone.localtime(timezone.now()) +
                    timezone.timedelta(hours=24)).date()

        settings = NotificationSetting.get_solo()
        settings.next_notification = now
        settings.save()

        dsmr_notification.services.set_next_notification(settings, now)

        self.assertEqual(settings.next_notification, tomorrow)

def create_notification_message(day, stats):
    """
    Create the action notification message
    :param day:
    :param stats:
    :return:
    """
    capabilities = dsmr_backend.services.get_capabilities()
    day_date = (day - timezone.timedelta(hours=1)).strftime("%d-%m-%Y")
    message = _('Your daily usage statistics for {}\n').format(day_date)

    if capabilities['electricity']:
        electricity_merged = dsmr_consumption.services.round_decimal(stats.electricity_merged)
        message += _('Electricity consumed: {} kWh\n').format(electricity_merged)

    if capabilities['electricity_returned']:
        electricity_returned_merged = dsmr_consumption.services.round_decimal(stats.electricity_returned_merged)
        message += _('Electricity returned: {} kWh\n').format(electricity_returned_merged)

    if capabilities['gas']:
        gas = dsmr_consumption.services.round_decimal(stats.gas)
        message += _('Gas consumed: {} m3\n').format(gas)

    message += _('Total cost: € {}').format(dsmr_consumption.services.round_decimal(stats.total_cost))
    return message

def test_check_interval_restriction(self, now_mock, create_backup_mock):
        """ Test whether backups are restricted by one backup per day. """
        now_mock.return_value = timezone.make_aware(timezone.datetime(2016, 1, 1, hour=1, minute=5))

        # Fake latest backup.
        now = timezone.localtime(timezone.now())
        backup_settings = BackupSettings.get_solo()
        backup_settings.latest_backup = now
        backup_settings.backup_time = (now - timezone.timedelta(minutes=1)).time()
        backup_settings.save()

        self.assertIsNotNone(BackupSettings.get_solo().latest_backup)
        self.assertFalse(create_backup_mock.called)

        # Should not do anything.
        dsmr_backup.services.backup.check()
        self.assertFalse(create_backup_mock.called)

        backup_settings.latest_backup = now - timezone.timedelta(days=1)
        backup_settings.save()

        # Should be fine to backup now.
        dsmr_backup.services.backup.check()
        self.assertTrue(create_backup_mock.called)

def test_check_backup_time_restriction(self, now_mock, create_backup_mock):
        """ Test whether backups are restricted by user's backup time preference. """
        now_mock.return_value = timezone.make_aware(timezone.datetime(2016, 1, 1, hour=1, minute=5))

        now = timezone.localtime(timezone.now())
        backup_settings = BackupSettings.get_solo()
        backup_settings.latest_backup = now - timezone.timedelta(days=1)
        backup_settings.backup_time = (now + timezone.timedelta(seconds=15)).time()
        backup_settings.save()

        # Should not do anything, we should backup a minute from now.
        self.assertFalse(create_backup_mock.called)
        dsmr_backup.services.backup.check()
        self.assertFalse(create_backup_mock.called)

        # Should be fine to backup now. Passed prefered time of user.
        backup_settings.backup_time = now.time()
        backup_settings.save()

        dsmr_backup.services.backup.check()
        self.assertTrue(create_backup_mock.called)

def test_sync_last_modified(self, now_mock, upload_chunked_mock, get_backup_directory_mock):
        """ Test whether syncs are skipped when file was not modified. """
        now_mock.return_value = timezone.make_aware(timezone.datetime(2016, 1, 1))

        dropbox_settings = DropboxSettings.get_solo()
        dropbox_settings.latest_sync = timezone.now() - timezone.timedelta(weeks=1)
        dropbox_settings.save()

        with tempfile.TemporaryDirectory() as temp_dir:
            get_backup_directory_mock.return_value = temp_dir
            temp_file = tempfile.NamedTemporaryFile(dir=temp_dir, delete=False)
            temp_file.write(b'Meh.')
            temp_file.flush()

            # 1420070400: 01 Jan 2015 00:00:00 GMT
            os.utime(temp_file.name, times=(1420070400, 1420070400))
            self.assertFalse(upload_chunked_mock.called)

            # File should be ignored, as it's modification timestamp is before latest sync.
            dsmr_backup.services.dropbox.sync()
            self.assertFalse(upload_chunked_mock.called)

def insert_notifications(apps, schema_editor):
    import dsmr_frontend.services
    Notification = apps.get_model('dsmr_frontend', 'Notification')

    # Search for any applied migrations in the past. This should indicate a long(er) living instance of the project.
    existing_project = MigrationRecorder.Migration.objects.filter(
        applied__lt=timezone.now() - timezone.timedelta(hours=24)
    ).exists()

    if existing_project:
        return

    Notification.objects.create(
        message=dsmr_frontend.services.get_translated_string(
            text=_('Welcome to DSMR-reader! Please make sure to check your settings in the Configuration page!')
        ),
        redirect_to='admin:index'
    )

    Notification.objects.create(
        message=dsmr_frontend.services.get_translated_string(
            text=_('You may check the status of your readings and data in the Status page.')
        ),
        redirect_to='frontend:status'
    )

def test_status_back_to_the_future(self, now_mock):
        """ Test some weird situation having the smart meter reporting a future timestamp in the telegrams. """
        if not self.support_data:
            self.skipTest('No data')

        now_mock.return_value = timezone.make_aware(timezone.datetime(2017, 2, 1))

        latest_reading = DsmrReading.objects.all().order_by('-timestamp')[0]
        DsmrReading.objects.exclude(pk=latest_reading.pk).delete()
        latest_reading.timestamp = timezone.now() + timezone.timedelta(seconds=15)  # Future reading.
        latest_reading.save()

        response = self.client.get(
            reverse('{}:status'.format(self.namespace))
        )
        self.assertEqual(response.status_code, 200)

        # These should be reset for convenience.
        self.assertEqual(response.context['latest_reading'].timestamp, timezone.now())
        self.assertEqual(response.context['delta_since_latest_reading'], 0)

def test_next_sync_setting_retroactive(self):
        """ Test whether the migration can also handle existing data. """
        now = timezone.now().replace(microsecond=0)

        TemperatureReading.objects.create(
            read_at=now + timezone.timedelta(hours=1),
            degrees_celcius=20,
        )
        TemperatureReading.objects.create(
            read_at=now,
            degrees_celcius=20,
        )

        self.assertIsNone(WeatherSettings.get_solo().next_sync)

        # Now we fake applying the migration (again for this test).
        MigrationRecorder.Migration.objects.filter(
            app='dsmr_weather', name='0004_next_sync_setting_retroactive'
        ).delete()
        MigrationExecutor(connection=connection).migrate([(self.app, '0004_next_sync_setting_retroactive')])

        # When having existing data, next_sync should be based on latest reading.
        self.assertEqual(WeatherSettings.get_solo().next_sync, now + timezone.timedelta(hours=2))

def _convert_legacy_dsmr_gas_line(parsed_reading, current_line, next_line):
    """ Legacy support for DSMR 2.x gas. """
    legacy_gas_line = current_line

    if next_line.startswith('('):  # pragma: no cover
        legacy_gas_line = current_line + next_line

    legacy_gas_result = re.search(
        r'[^(]+\((\d+)\)\(\d+\)\(\d+\)\(\d+\)\([0-9-.:]+\)\(m3\)\(([0-9.]+)\)',
        legacy_gas_line
    )
    gas_timestamp = legacy_gas_result.group(1)

    if timezone.now().dst() != timezone.timedelta(0):
        gas_timestamp += 'S'
    else:
        gas_timestamp += 'W'

    parsed_reading['extra_device_timestamp'] = reading_timestamp_to_datetime(
        string=gas_timestamp
    )
    parsed_reading['extra_device_delivered'] = legacy_gas_result.group(2)
    return parsed_reading

def test_export_fail(self, now_mock, should_export_mock, requests_post_mock):
        """ Test export() failing by denied API call. """
        now_mock.return_value = timezone.make_aware(timezone.datetime(2015, 12, 12, hour=4, minute=45))
        should_export_mock.return_value = True

        settings = MinderGasSettings.get_solo()
        self.assertFalse(settings.export)
        self.assertIsNone(settings.next_export)
        self.assertFalse(requests_post_mock.called)

        # Mindergas error codes according to docs.
        for current_error_code in (401, 422):
            requests_post_mock.return_value = mock.MagicMock(status_code=current_error_code, text='Error message')

            dsmr_mindergas.services.export()

        settings = MinderGasSettings.get_solo()

        # This should be set one hour forward now.
        self.assertEqual(settings.next_export, timezone.now() + timezone.timedelta(hours=1))
        self.assertTrue(requests_post_mock.called)

def setUp(self):
        calendar = Calendar.objects.create_calendar('default')
        now = timezone.now()
        day = timezone.timedelta(days=1)
        hour = timezone.timedelta(hours=1)
        self.event = Event.objects.create(
            name='Test',
            nb_path='/pseudo/test',
            start_time=now + 3 * day,
            end_time=now + 3 * day + 4 * hour,
            calendar=calendar
        )

        self.group = SupportGroup.objects.create(
            name='Test',
            nb_path='/12/grouptest'
        )

def setUp(self):
        self.calendar = Calendar.objects.create(
            name="My calendar",
            slug="my_calendar",
        )

        now = timezone.now()
        day = timezone.timedelta(days=1)
        hour = timezone.timedelta(hours=1)

        for i in range(20):
            Event.objects.create(
                name="Event {}".format(i),
                calendar=self.calendar,
                start_time=now + i * day,
                end_time=now + i * day + hour
            )