codabench/src/apps/api/tests/test_cleanup.py at 801a7bd983eafba0658ed0e26f666fb2308f21e1 · codalab/codabench · 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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
from rest_framework.test import APITestCase
from django.urls import reverse
import json
from factories import (
    UserFactory,
    CompetitionFactory,
    PhaseFactory,
    TaskFactory,
    SubmissionFactory,
    DataFactory
)
from competitions.models import Submission
from datasets.models import Data


class CleanUpTests(APITestCase):
    def setUp(self):

        # Create a user
        user = UserFactory(username='test_user', password='test_user')

        # Create a competition
        comp = CompetitionFactory(created_by=user)

        # Create used tasks
        self.used_tasks = [
            TaskFactory(created_by=user),
            TaskFactory(created_by=user)
        ]

        # Create unused task
        self.unused_tasks = [
            TaskFactory(created_by=user),
            TaskFactory(created_by=user)
        ]

        # Create phase with used tasks
        phase = PhaseFactory(competition=comp, tasks=self.used_tasks)

        # Create used-failed submission
        self.failed_submissions = [SubmissionFactory(
            phase=phase,
            owner=user,
            status=Submission.FAILED,
            data=DataFactory(created_by=user, type=Data.SUBMISSION, competition=comp)
        )]

        # Create unused submission
        self.unused_submissions = [
            DataFactory(created_by=user, type=Data.SUBMISSION),
            DataFactory(created_by=user, type=Data.SUBMISSION)
        ]

        # Create unused datasets and programs
        self.unused_datasets_programs = [
            DataFactory(created_by=user, type=Data.INGESTION_PROGRAM),
            DataFactory(created_by=user, type=Data.SCORING_PROGRAM),
            DataFactory(created_by=user, type=Data.INPUT_DATA),
            DataFactory(created_by=user, type=Data.REFERENCE_DATA)
        ]

        # Create unused starting kits
        self.unused_starting_kits = [
            DataFactory(created_by=user, type=Data.STARTING_KIT),
            DataFactory(created_by=user, type=Data.STARTING_KIT)
        ]

        # Create unused competition bundles
        self.unused_competition_bundles = [
            DataFactory(created_by=user, type=Data.COMPETITION_BUNDLE),
            DataFactory(created_by=user, type=Data.COMPETITION_BUNDLE)
        ]

        self.client.login(username='test_user', password='test_user')

    def test_cleanup_stats(self):

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_tasks"] == len(self.unused_tasks)
        assert content["unused_datasets_programs"] == len(self.unused_datasets_programs)
        assert content["unused_submissions"] == len(self.unused_submissions)
        assert content["failed_submissions"] == len(self.failed_submissions)
        assert content["unused_starting_kits"] == len(self.unused_starting_kits)
        assert content["unused_competition_bundles"] == len(self.unused_competition_bundles)

    def test_delete_unused_tasks(self):

        url = reverse('delete_unused_tasks')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Unused tasks deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_tasks"] == 0

    def test_delete_unused_datasets(self):

        url = reverse('delete_unused_datasets')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Unused datasets and programs deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_datasets_programs"] == 0

    def test_delete_unused_submissions(self):

        url = reverse('delete_unused_submissions')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Unused submissions deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_submissions"] == 0

    def test_delete_failed_submissions(self):

        url = reverse('delete_failed_submissions')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Failed submissions deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["failed_submissions"] == 0

    def test_delete_unused_starting_kits(self):

        url = reverse('delete_unused_starting_kits')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Unused starting kits deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_starting_kits"] == 0

    def test_delete_unused_competition_bundles(self):

        url = reverse('delete_unused_competition_bundles')
        resp = self.client.delete(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["success"]
        assert content["message"] == "Unused competition bundles deleted successfully"

        url = reverse('user_quota_cleanup')
        resp = self.client.get(url)
        assert resp.status_code == 200
        content = json.loads(resp.content)
        assert content["unused_competition_bundles"] == 0