datatracker/ietf/person/management/commands/tests.py at 7.40.0 · https-github-com-bit/datatracker · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
# Copyright The IETF Trust 2021, All Rights Reserved
# -*- coding: utf-8 -*-

import datetime
from io import StringIO

from django.core.management import call_command, CommandError

from ietf.person.factories import PersonApiKeyEventFactory
from ietf.person.models import PersonApiKeyEvent, PersonEvent
from ietf.utils.test_utils import TestCase


class CommandTests(TestCase):
    @staticmethod
    def _call_command(command_name, *args, **options):
        out = StringIO()
        options['stdout'] = out
        call_command(command_name, *args, **options)
        return out.getvalue()

    def _assert_purge_results(self, cmd_output, expected_delete_count, expected_kept_events):
        self.assertNotIn('Dry run requested', cmd_output)
        if expected_delete_count == 0:
            delete_text = 'No events older than'
        else:
            delete_text = 'Deleting {} event'.format(expected_delete_count)
        self.assertIn(delete_text, cmd_output)
        self.assertCountEqual(
            PersonApiKeyEvent.objects.all(),
            expected_kept_events,
            'Wrong events were deleted'
        )

    def _assert_purge_dry_run_results(self, cmd_output, expected_delete_count, expected_kept_events):
        self.assertIn('Dry run requested', cmd_output)
        if expected_delete_count == 0:
            delete_text = 'No events older than'
        else:
            delete_text = 'Would delete {} event'.format(expected_delete_count)
        self.assertIn(delete_text, cmd_output)
        self.assertCountEqual(
            PersonApiKeyEvent.objects.all(),
            expected_kept_events,
            'Events were deleted when dry-run option was used'
        )

    def test_purge_old_personal_api_key_events(self):
        keep_days = 10

        # Remember how many PersonEvents were present so we can verify they're cleaned up properly.
        personevents_before = PersonEvent.objects.count()

        now = datetime.datetime.now()
        # The first of these events will be timestamped a fraction of a second more than keep_days
        # days ago by the time we call the management command, so will just barely chosen for purge.
        old_events = [
            PersonApiKeyEventFactory(time=now - datetime.timedelta(days=n))
            for n in range(keep_days, 2 * keep_days + 1)
        ]
        num_old_events = len(old_events)

        recent_events = [
            PersonApiKeyEventFactory(time=now - datetime.timedelta(days=n))
            for n in range(0, keep_days)
        ]
        # We did not create recent_event timestamped exactly keep_days ago because it would
        # be treated as an old_event by the management command. Create an event a few seconds
        # on the "recent" side of keep_days old to test the threshold.
        recent_events.append(
            PersonApiKeyEventFactory(
                time=now + datetime.timedelta(seconds=3) - datetime.timedelta(days=keep_days)
            )
        )
        num_recent_events = len(recent_events)

        # call with dry run
        output = self._call_command('purge_old_personal_api_key_events', str(keep_days), '--dry-run')
        self._assert_purge_dry_run_results(output, num_old_events, old_events + recent_events)

        # call for real
        output = self._call_command('purge_old_personal_api_key_events', str(keep_days))
        self._assert_purge_results(output, num_old_events, recent_events)
        self.assertEqual(PersonEvent.objects.count(), personevents_before + num_recent_events,
                         'PersonEvents were not cleaned up properly')

        # repeat - there should be nothing left to delete
        output = self._call_command('purge_old_personal_api_key_events', '--dry-run', str(keep_days))
        self._assert_purge_dry_run_results(output, 0, recent_events)

        output = self._call_command('purge_old_personal_api_key_events', str(keep_days))
        self._assert_purge_results(output, 0, recent_events)
        self.assertEqual(PersonEvent.objects.count(), personevents_before + num_recent_events,
                         'PersonEvents were not cleaned up properly')

        # and now delete the remaining events
        output = self._call_command('purge_old_personal_api_key_events', '0')
        self._assert_purge_results(output, num_recent_events, [])
        self.assertEqual(PersonEvent.objects.count(), personevents_before,
                         'PersonEvents were not cleaned up properly')

    def test_purge_old_personal_api_key_events_rejects_invalid_arguments(self):
        """The purge_old_personal_api_key_events command should reject invalid arguments"""
        event = PersonApiKeyEventFactory(time=datetime.datetime.now() - datetime.timedelta(days=30))

        with self.assertRaises(CommandError):
            self._call_command('purge_old_personal_api_key_events')

        with self.assertRaises(CommandError):
            self._call_command('purge_old_personal_api_key_events', '-15')

        with self.assertRaises(CommandError):
            self._call_command('purge_old_personal_api_key_events', '15.3')

        with self.assertRaises(CommandError):
            self._call_command('purge_old_personal_api_key_events', '15', '15')

        with self.assertRaises(CommandError):
            self._call_command('purge_old_personal_api_key_events', 'abc', '15')

        self.assertCountEqual(PersonApiKeyEvent.objects.all(), [event])