Skip to content

tests

myapp.tests.test_models

CourseTestCase

Bases: TestCase

Source code in myapp/tests/test_models.py
 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
class CourseTestCase(TestCase):

    def setUp(self):
        c1 = Course.objects.create(
            school_bulletin_prefix_code='123',
            suffix_two='11',
            subject_area_code='A101',
            course_number='123',
            course_identifier='ABC123',
            course_name='Corp. Finance',
            course_description='bla bla bla 123',
            last_mod_user_name='Fred Gary')
        c1.save()

        c2 = Course.objects.create(
            school_bulletin_prefix_code='888',
            suffix_two='22',
            subject_area_code='A102',
            course_number='456',
            course_identifier='XYZ321',
            course_name='Data Science III',
            course_description='bla bla bla 456',
            last_mod_user_name='Tom Smith')
        c2.save()

        CourseTerm.objects.create(
            term_identifier='term_id_1',
            audit_permitted_code=123,
            exam_credit_flag=True,
            last_mod_user_name='Fred Gary',
            effective_start_date=datetime.now(),
            effective_end_date=date(2019, 1, 31),
            course=c1
        ).save()

        CourseTerm.objects.create(
            term_identifier='term_id_2',
            audit_permitted_code=222,
            exam_credit_flag=False,
            last_mod_user_name='Tom Frank',
            effective_start_date=datetime.now(),
            effective_end_date=date(2019, 2, 28),
            course=c1
        ).save()

        CourseTerm.objects.create(
            term_identifier='term_id_21',
            audit_permitted_code=321,
            exam_credit_flag=True,
            last_mod_user_name='Fred Gary',
            effective_start_date=datetime.now(),
            effective_end_date=date(2020, 1, 31),
            course=c2
        ).save()

        CourseTerm.objects.create(
            term_identifier='term_id_22',
            audit_permitted_code=422,
            exam_credit_flag=False,
            last_mod_user_name='Tom Frank',
            last_mod_date=datetime.now(),
            effective_start_date=datetime.now(),
            effective_end_date=date(2020, 12, 28),
            course=c2
        ).save()

    def test_str(self):
        courses = Course.objects.all()
        c = courses.get(course_identifier='ABC123')
        self.assertEqual(c.course_identifier, 'ABC123')

    def test_date(self):
        courses = Course.objects.all()
        c = courses.get(course_identifier='ABC123')
        self.assertEqual(c.last_mod_date, date.today())

    def test_num(self):
        cs = Course.objects.all()
        self.assertEqual(2, cs.count())

        terms = CourseTerm.objects.all()
        self.assertEqual(4, terms.count())

        # field look up
        ts = CourseTerm.objects.filter(course__course_identifier='123ABC')
        self.assertEqual(0, ts.count())

        ts = CourseTerm.objects.filter(course__course_identifier='ABC123')
        self.assertEqual(2, ts.count())

        ts = CourseTerm.objects.filter(last_mod_user_name__startswith='Tom')
        self.assertEqual(2, ts.count())

        cs = Course.objects.filter(course_terms__last_mod_user_name__startswith='Tom')
        self.assertEqual(2, cs.count())

    def test_dup_fail(self):
        with self.assertRaises(IntegrityError):
            c1 = Course.objects.create(
                school_bulletin_prefix_code='123',
                suffix_two='11',
                subject_area_code='A101',
                course_number='123',
                course_identifier='ABC123',
                course_name='Corp. Finance',
                course_description='bla bla bla 123',
                last_mod_user_name='Fred Gary')
            c1.save()

setUp()

Source code in myapp/tests/test_models.py
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
def setUp(self):
    c1 = Course.objects.create(
        school_bulletin_prefix_code='123',
        suffix_two='11',
        subject_area_code='A101',
        course_number='123',
        course_identifier='ABC123',
        course_name='Corp. Finance',
        course_description='bla bla bla 123',
        last_mod_user_name='Fred Gary')
    c1.save()

    c2 = Course.objects.create(
        school_bulletin_prefix_code='888',
        suffix_two='22',
        subject_area_code='A102',
        course_number='456',
        course_identifier='XYZ321',
        course_name='Data Science III',
        course_description='bla bla bla 456',
        last_mod_user_name='Tom Smith')
    c2.save()

    CourseTerm.objects.create(
        term_identifier='term_id_1',
        audit_permitted_code=123,
        exam_credit_flag=True,
        last_mod_user_name='Fred Gary',
        effective_start_date=datetime.now(),
        effective_end_date=date(2019, 1, 31),
        course=c1
    ).save()

    CourseTerm.objects.create(
        term_identifier='term_id_2',
        audit_permitted_code=222,
        exam_credit_flag=False,
        last_mod_user_name='Tom Frank',
        effective_start_date=datetime.now(),
        effective_end_date=date(2019, 2, 28),
        course=c1
    ).save()

    CourseTerm.objects.create(
        term_identifier='term_id_21',
        audit_permitted_code=321,
        exam_credit_flag=True,
        last_mod_user_name='Fred Gary',
        effective_start_date=datetime.now(),
        effective_end_date=date(2020, 1, 31),
        course=c2
    ).save()

    CourseTerm.objects.create(
        term_identifier='term_id_22',
        audit_permitted_code=422,
        exam_credit_flag=False,
        last_mod_user_name='Tom Frank',
        last_mod_date=datetime.now(),
        effective_start_date=datetime.now(),
        effective_end_date=date(2020, 12, 28),
        course=c2
    ).save()

test_str()

Source code in myapp/tests/test_models.py
76
77
78
79
def test_str(self):
    courses = Course.objects.all()
    c = courses.get(course_identifier='ABC123')
    self.assertEqual(c.course_identifier, 'ABC123')

test_date()

Source code in myapp/tests/test_models.py
81
82
83
84
def test_date(self):
    courses = Course.objects.all()
    c = courses.get(course_identifier='ABC123')
    self.assertEqual(c.last_mod_date, date.today())

test_num()

Source code in myapp/tests/test_models.py
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def test_num(self):
    cs = Course.objects.all()
    self.assertEqual(2, cs.count())

    terms = CourseTerm.objects.all()
    self.assertEqual(4, terms.count())

    # field look up
    ts = CourseTerm.objects.filter(course__course_identifier='123ABC')
    self.assertEqual(0, ts.count())

    ts = CourseTerm.objects.filter(course__course_identifier='ABC123')
    self.assertEqual(2, ts.count())

    ts = CourseTerm.objects.filter(last_mod_user_name__startswith='Tom')
    self.assertEqual(2, ts.count())

    cs = Course.objects.filter(course_terms__last_mod_user_name__startswith='Tom')
    self.assertEqual(2, cs.count())

test_dup_fail()

Source code in myapp/tests/test_models.py
106
107
108
109
110
111
112
113
114
115
116
117
def test_dup_fail(self):
    with self.assertRaises(IntegrityError):
        c1 = Course.objects.create(
            school_bulletin_prefix_code='123',
            suffix_two='11',
            subject_area_code='A101',
            course_number='123',
            course_identifier='ABC123',
            course_name='Corp. Finance',
            course_description='bla bla bla 123',
            last_mod_user_name='Fred Gary')
        c1.save()

ManyToManyTestCase

Bases: TestCase

do some performance tests on large numbers of many-to-many relationships

Source code in myapp/tests/test_models.py
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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
class ManyToManyTestCase(TestCase):
    """
    do some performance tests on large numbers of many-to-many relationships
    """
    def setUp(self):
        # make 50 CourseTerms
        self.courseterms = []
        for i in range(50):
            t = CourseTerm.objects.create(term_identifier="term {}".format(i))
            t.save()
            self.courseterms.append(t)
        # make 50 instructors (one-to-one with people)
        self.people = []
        self.instructors = []
        for i in range(100):
            p = Person.objects.create(name="Person {}".format(i))
            p.save()
            self.people.append(p)
            i = Instructor.objects.create(person=p)
            # each instructor teaches all courseterms
            i.course_terms.set(self.courseterms)
            self.instructors.append(i)
            i.save()

    def test_prefetch_related(self):
        """
        Try to understand why prefetch_related does a giant WHERE IN (...) which doesn't scale well as it requires
        a parameter for each **row** of the intermediate myapp_instructor_course_terms table.  This breaks
        databases like MS SQLServer that has a limit of 2100 parameters (which is huge but fewer than mySQL supports).

        I think this is actually a _feature_ and is based on using prefetch with paginated queries.

        However, for a related child, prefetching that Model can break the database if it is not paginated and has a
        large number of objects. This is an
        `issue<https://github.com/django-json-api/django-rest-framework-json-api/issues/178>`_
        raised for DJA that is worked around by
        `skipping the related data<https://github.com/django-json-api/django-rest-framework-json-api/pull/445>`_,
        and leaving only the relationship hyperlink in place.

        According to the JSON:API spec:
        "A relationship object that represents a to-many relationship MAY also contain pagination links under the
        links member, as described below. Any pagination links in a relationship object MUST paginate the relationship
        data, not the related resources." -- https://jsonapi.org/format/#document-resource-object-relationships

        First it gets the list of instructors:

        SELECT [myapp_instructor].[id], [myapp_instructor].[effective_start_date], ...
          FROM [myapp_instructor]
          ORDER BY [myapp_instructor].[id] ASC

        Then uses the ids in a WHERE IN list:

        SELECT ([myapp_instructor_course_terms].[instructor_id])
            AS [_prefetch_related_val_instructor_id], [myapp_courseterm].[id], ...
          FROM [myapp_courseterm] INNER JOIN [myapp_instructor_course_terms]
            ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id])
         WHERE [myapp_instructor_course_terms].[instructor_id] IN (%s, %s, ...)

        But if you paginate the query (page size 10), then a 'SELECT TOP 10 ...'
        or 'SELECT ... OFFSET 10 FETCH FIRST 10 ROWS ONLY' happens

        To test this with different database engines, set `DJANGO_MYSQL=true` or `DJANGO_SQLSERVER=true` (see settings)

        Right now the SQL database debug messages just go to the console.
        TODO: Find a way to capture the SQL queries here in the test case so we can do something with them.

        See also https://medium.com/@hansonkd/performance-problems-in-the-django-orm-1f62b3d04785
        https://docs.djangoproject.com/en/2.1/ref/models/querysets/#django.db.models.Prefetch

        :return:
        """
        # TestCase turns off DEBUG, so turn it back on so we get the SQL queries logged
        olddebug = settings.DEBUG
        settings.DEBUG = True
        instructors = Instructor.objects.all().prefetch_related('course_terms')
        ####
        # For the "master" table (instructors), the prefetch list of parameters is limited to the "page" size.
        # Make a queryset of a "page" of 5 instructors.
        # In the comments below, the SQL debug log output has been cleaned up to be a little more readable.
        # (for added readability I've replaced the list of all selected column names with '*'):
        ####
        first_page = instructors[0:5]
        ####
        # Reference the instructors values. Iterating over the queryset values is what triggers the prefetch query to
        # execute.
        #
        # 1. The first 5 Instructor IDs are selected:
        #
        # SELECT TOP 5 [myapp_instructor].* FROM [myapp_instructor] ORDER BY [myapp_instructor].[id] ASC
        #
        # 2. Prefetch the (Instructor ID, associated CourseTerm.*) tuples for the given 5 Instructor IDs via the
        #    intermediate many-to-many table: instructor_course_terms:
        #
        # SELECT([myapp_instructor_course_terms].[instructor_id]) AS [_prefetch_related_val_instructor_id],
        #        [myapp_courseterm].*
        #   FROM [myapp_courseterm]
        #   INNER JOIN [myapp_instructor_course_terms]
        #           ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id])
        #        WHERE [myapp_instructor_course_terms].[instructor_id] IN (...5 instructor ids...)
        #        ORDER BY [myapp_courseterm].[term_identifier] ASC
        #
        ####
        for i in first_page:
            first_instr = i  # noqa F841
            first_instr_terms = i.course_terms.all().prefetch_related('instructors')
            break
        ####
        # Here's a non-paginated child relationship:
        #
        # 1. Get all the CourseTerm IDs for the first instructor:
        #
        # SELECT [myapp_courseterm].* FROM [myapp_courseterm]
        #  INNER JOIN [myapp_instructor_course_terms]
        #          ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id])
        #       WHERE [myapp_instructor_course_terms].[instructor_id] = <first instructor id>
        #  ORDER BY [myapp_courseterm].[term_identifier] ASC
        #
        # 2. Prefetch the (CourseTerm ID, associated Instructor.*) tuples for that list of CourseTerm IDs:
        #
        # SELECT ([myapp_instructor_course_terms].[courseterm_id]) AS [_prefetch_related_val_courseterm_id],
        #        [myapp_instructor].*
        #   FROM [myapp_instructor]
        #   INNER JOIN [myapp_instructor_course_terms]
        #      ON ([myapp_instructor].[id] = [myapp_instructor_course_terms].[instructor_id])
        #   WHERE [myapp_instructor_course_terms].[courseterm_id] IN (...long course_term list...)
        ####
        for t in first_instr_terms:
            first_termid = t.term_identifier  # noqa F841
            break
        ####
        # select the next page of Instructors
        ####
        next_page = instructors[5:10]
        ####
        #
        # SELECT [myapp_instructor].* FROM [myapp_instructor] ORDER BY [myapp_instructor].[id] ASC
        #        OFFSET 5 ROWS FETCH FIRST 5 ROWS ONLY
        #
        # SELECT ([myapp_instructor_course_terms].[instructor_id]) AS [_prefetch_related_val_instructor_id],
        #         [myapp_courseterm].*
        #   FROM [myapp_courseterm]
        #   INNER JOIN [myapp_instructor_course_terms]
        #      ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id])
        #   WHERE [myapp_instructor_course_terms].[instructor_id] IN (...5 ids...)
        #   ORDER BY [myapp_courseterm].[term_identifier] ASC
        #
        ####
        for i in next_page:
            instr = i  # noqa F841
            break
        settings.DEBUG = olddebug

setUp()

Source code in myapp/tests/test_models.py
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def setUp(self):
    # make 50 CourseTerms
    self.courseterms = []
    for i in range(50):
        t = CourseTerm.objects.create(term_identifier="term {}".format(i))
        t.save()
        self.courseterms.append(t)
    # make 50 instructors (one-to-one with people)
    self.people = []
    self.instructors = []
    for i in range(100):
        p = Person.objects.create(name="Person {}".format(i))
        p.save()
        self.people.append(p)
        i = Instructor.objects.create(person=p)
        # each instructor teaches all courseterms
        i.course_terms.set(self.courseterms)
        self.instructors.append(i)
        i.save()

Try to understand why prefetch_related does a giant WHERE IN (...) which doesn't scale well as it requires a parameter for each row of the intermediate myapp_instructor_course_terms table. This breaks databases like MS SQLServer that has a limit of 2100 parameters (which is huge but fewer than mySQL supports).

I think this is actually a feature and is based on using prefetch with paginated queries.

However, for a related child, prefetching that Model can break the database if it is not paginated and has a large number of objects. This is an issue<https://github.com/django-json-api/django-rest-framework-json-api/issues/178>_ raised for DJA that is worked around by skipping the related data<https://github.com/django-json-api/django-rest-framework-json-api/pull/445>_, and leaving only the relationship hyperlink in place.

According to the JSON:API spec: "A relationship object that represents a to-many relationship MAY also contain pagination links under the links member, as described below. Any pagination links in a relationship object MUST paginate the relationship data, not the related resources." -- https://jsonapi.org/format/#document-resource-object-relationships

First it gets the list of instructors:

SELECT [myapp_instructor].[id], [myapp_instructor].[effective_start_date], ... FROM [myapp_instructor] ORDER BY [myapp_instructor].[id] ASC

Then uses the ids in a WHERE IN list:

SELECT ([myapp_instructor_course_terms].[instructor_id]) AS [_prefetch_related_val_instructor_id], [myapp_courseterm].[id], ... FROM [myapp_courseterm] INNER JOIN [myapp_instructor_course_terms] ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id]) WHERE [myapp_instructor_course_terms].[instructor_id] IN (%s, %s, ...)

But if you paginate the query (page size 10), then a 'SELECT TOP 10 ...' or 'SELECT ... OFFSET 10 FETCH FIRST 10 ROWS ONLY' happens

To test this with different database engines, set DJANGO_MYSQL=true or DJANGO_SQLSERVER=true (see settings)

Right now the SQL database debug messages just go to the console. TODO: Find a way to capture the SQL queries here in the test case so we can do something with them.

See also https://medium.com/@hansonkd/performance-problems-in-the-django-orm-1f62b3d04785 https://docs.djangoproject.com/en/2.1/ref/models/querysets/#django.db.models.Prefetch

Returns:

Type Description
Source code in myapp/tests/test_models.py
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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def test_prefetch_related(self):
    """
    Try to understand why prefetch_related does a giant WHERE IN (...) which doesn't scale well as it requires
    a parameter for each **row** of the intermediate myapp_instructor_course_terms table.  This breaks
    databases like MS SQLServer that has a limit of 2100 parameters (which is huge but fewer than mySQL supports).

    I think this is actually a _feature_ and is based on using prefetch with paginated queries.

    However, for a related child, prefetching that Model can break the database if it is not paginated and has a
    large number of objects. This is an
    `issue<https://github.com/django-json-api/django-rest-framework-json-api/issues/178>`_
    raised for DJA that is worked around by
    `skipping the related data<https://github.com/django-json-api/django-rest-framework-json-api/pull/445>`_,
    and leaving only the relationship hyperlink in place.

    According to the JSON:API spec:
    "A relationship object that represents a to-many relationship MAY also contain pagination links under the
    links member, as described below. Any pagination links in a relationship object MUST paginate the relationship
    data, not the related resources." -- https://jsonapi.org/format/#document-resource-object-relationships

    First it gets the list of instructors:

    SELECT [myapp_instructor].[id], [myapp_instructor].[effective_start_date], ...
      FROM [myapp_instructor]
      ORDER BY [myapp_instructor].[id] ASC

    Then uses the ids in a WHERE IN list:

    SELECT ([myapp_instructor_course_terms].[instructor_id])
        AS [_prefetch_related_val_instructor_id], [myapp_courseterm].[id], ...
      FROM [myapp_courseterm] INNER JOIN [myapp_instructor_course_terms]
        ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id])
     WHERE [myapp_instructor_course_terms].[instructor_id] IN (%s, %s, ...)

    But if you paginate the query (page size 10), then a 'SELECT TOP 10 ...'
    or 'SELECT ... OFFSET 10 FETCH FIRST 10 ROWS ONLY' happens

    To test this with different database engines, set `DJANGO_MYSQL=true` or `DJANGO_SQLSERVER=true` (see settings)

    Right now the SQL database debug messages just go to the console.
    TODO: Find a way to capture the SQL queries here in the test case so we can do something with them.

    See also https://medium.com/@hansonkd/performance-problems-in-the-django-orm-1f62b3d04785
    https://docs.djangoproject.com/en/2.1/ref/models/querysets/#django.db.models.Prefetch

    :return:
    """
    # TestCase turns off DEBUG, so turn it back on so we get the SQL queries logged
    olddebug = settings.DEBUG
    settings.DEBUG = True
    instructors = Instructor.objects.all().prefetch_related('course_terms')
    ####
    # For the "master" table (instructors), the prefetch list of parameters is limited to the "page" size.
    # Make a queryset of a "page" of 5 instructors.
    # In the comments below, the SQL debug log output has been cleaned up to be a little more readable.
    # (for added readability I've replaced the list of all selected column names with '*'):
    ####
    first_page = instructors[0:5]
    ####
    # Reference the instructors values. Iterating over the queryset values is what triggers the prefetch query to
    # execute.
    #
    # 1. The first 5 Instructor IDs are selected:
    #
    # SELECT TOP 5 [myapp_instructor].* FROM [myapp_instructor] ORDER BY [myapp_instructor].[id] ASC
    #
    # 2. Prefetch the (Instructor ID, associated CourseTerm.*) tuples for the given 5 Instructor IDs via the
    #    intermediate many-to-many table: instructor_course_terms:
    #
    # SELECT([myapp_instructor_course_terms].[instructor_id]) AS [_prefetch_related_val_instructor_id],
    #        [myapp_courseterm].*
    #   FROM [myapp_courseterm]
    #   INNER JOIN [myapp_instructor_course_terms]
    #           ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id])
    #        WHERE [myapp_instructor_course_terms].[instructor_id] IN (...5 instructor ids...)
    #        ORDER BY [myapp_courseterm].[term_identifier] ASC
    #
    ####
    for i in first_page:
        first_instr = i  # noqa F841
        first_instr_terms = i.course_terms.all().prefetch_related('instructors')
        break
    ####
    # Here's a non-paginated child relationship:
    #
    # 1. Get all the CourseTerm IDs for the first instructor:
    #
    # SELECT [myapp_courseterm].* FROM [myapp_courseterm]
    #  INNER JOIN [myapp_instructor_course_terms]
    #          ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id])
    #       WHERE [myapp_instructor_course_terms].[instructor_id] = <first instructor id>
    #  ORDER BY [myapp_courseterm].[term_identifier] ASC
    #
    # 2. Prefetch the (CourseTerm ID, associated Instructor.*) tuples for that list of CourseTerm IDs:
    #
    # SELECT ([myapp_instructor_course_terms].[courseterm_id]) AS [_prefetch_related_val_courseterm_id],
    #        [myapp_instructor].*
    #   FROM [myapp_instructor]
    #   INNER JOIN [myapp_instructor_course_terms]
    #      ON ([myapp_instructor].[id] = [myapp_instructor_course_terms].[instructor_id])
    #   WHERE [myapp_instructor_course_terms].[courseterm_id] IN (...long course_term list...)
    ####
    for t in first_instr_terms:
        first_termid = t.term_identifier  # noqa F841
        break
    ####
    # select the next page of Instructors
    ####
    next_page = instructors[5:10]
    ####
    #
    # SELECT [myapp_instructor].* FROM [myapp_instructor] ORDER BY [myapp_instructor].[id] ASC
    #        OFFSET 5 ROWS FETCH FIRST 5 ROWS ONLY
    #
    # SELECT ([myapp_instructor_course_terms].[instructor_id]) AS [_prefetch_related_val_instructor_id],
    #         [myapp_courseterm].*
    #   FROM [myapp_courseterm]
    #   INNER JOIN [myapp_instructor_course_terms]
    #      ON ([myapp_courseterm].[id] = [myapp_instructor_course_terms].[courseterm_id])
    #   WHERE [myapp_instructor_course_terms].[instructor_id] IN (...5 ids...)
    #   ORDER BY [myapp_courseterm].[term_identifier] ASC
    #
    ####
    for i in next_page:
        instr = i  # noqa F841
        break
    settings.DEBUG = olddebug

myapp.tests.test_views

HEADERS = {'HTTP_ACCEPT': 'application/vnd.api+json', 'content_type': 'application/vnd.api+json'} module-attribute

COURSE_POST = {'data': {'type': 'courses', 'attributes': {'school_bulletin_prefix_code': 'CEFKGRUXI', 'suffix_two': '00', 'subject_area_code': 'IF', 'course_number': '61044', 'course_identifier': 'AHIS2321W', 'course_name': 'ROME BEYOND ROME-DISC 4', 'course_description': 'blah blah', 'effective_start_date': None, 'effective_end_date': None, 'last_mod_user_name': 'alan', 'last_mod_date': '2018-03-10'}}} module-attribute

COURSE_POST_WITH_REL = {'data': {'type': 'courses', 'attributes': {'school_bulletin_prefix_code': 'CEFKGRUXI', 'suffix_two': '00', 'subject_area_code': 'IF', 'course_number': '61044', 'course_identifier': 'AHIS2321W', 'course_name': 'ROME BEYOND ROME-DISC 4', 'course_description': 'blah blah', 'effective_start_date': None, 'effective_end_date': None, 'last_mod_user_name': 'alan', 'last_mod_date': '2018-03-10'}, 'relationships': {'course_terms': {'data': [{'type': 'course_terms', 'id': 'e2060fb7-bc0e-4259-8077-2b678fff0c5f'}]}}}} module-attribute

COURSE_TERM_POST = {'data': {'type': 'course_terms', 'attributes': {'term_identifier': '20181FILM3119X', 'audit_permitted_code': 0, 'exam_credit_flag': False, 'effective_start_date': None, 'effective_end_date': None, 'last_mod_user_name': 'alan', 'last_mod_date': '2018-03-10'}}} module-attribute

COURSE_TERM_REL_PATCH = {'data': [{'type': 'course_terms', 'id': 'e724c43c-f443-4246-8947-a8bb8953699c'}]} module-attribute

DJATestCase

Bases: APITestCase

test cases using Django REST Framework drf-json-api (DJA)

Source code in myapp/tests/test_views.py
 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
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
class DJATestCase(APITestCase):
    """
    test cases using Django REST Framework drf-json-api (DJA)
    """
    # TODO: add tests of query parameters: page, filter, fields, sort, include, and combinations thereof
    # TODO: add failure test cases (e.g. get of related where the id is invalid returns 500 instead of 404).

    fixtures = ('auth', 'oauth2', 'testcases',)

    def setUp(self):
        """
        Set up some test data for the tests.
        !!! Note
            These users are defined in the 'auth' fixture:
            ```
            | username | password       | staff   | su        | group memberships |
            | -------- | -------------- | -----   | --------- | ----------------- |
            | admin    | admin123       | X       | X         | (none)            |
            | user1    | user1password1 | X       |           | team-a, team-c    |
            | user2    | user2password2 | X       |           | team-a, team-b    |
            | user3    | user3password3 | X       |           | (none)            |
            ```
        """
        self.read_write_user = User.objects.filter(username='user1').first()
        # `somebody` can view course but not anything else
        self.someuser = User.objects.create_user('somebody', is_superuser=False)
        self.someuser.user_permissions.add(Permission.objects.get(codename='view_course').id)
        # `nobody` has no permissions
        self.noneuser = User.objects.create_user('nobody', is_superuser=False)
        # most tests just use the read_write_user
        self.user1_token = oauth_models.MyAccessToken(  # nosec B106
            token='User1Token',
            user=self.read_write_user,
            expires=datetime.isoformat(datetime.now(tz=timezone.utc)+timedelta(seconds=3600)),
            scope='auth-columbia demo-djt-sla-bronze read create update openid '
                  'profile email https://api.columbia.edu/scope/group',
            userinfo='{"sub": "user1", "given_name": "First", "family_name": "User", "name": "First User", '
                     '"email": "user1@example.com", "https://api.columbia.edu/claim/group": "team-a team-c"}'
        )
        self.user1_token.save()
        # self.client.force_authenticate(user=self.read_write_user)
        HEADERS['Authorization'] = 'Bearer User1Token'
        self.courses = Course.objects.all()
        self.courses_url = reverse('course-list')
        self.course_terms = CourseTerm.objects.all()
        self.course_terms_url = reverse('courseterm-list')

    def test_post_course_term(self):
        response = self.client.post(self.courses_url,
                                    data=json.dumps(COURSE_POST),
                                    **HEADERS)
        self.assertEqual(response.status_code, 201, msg=response.content)
        course_id = json.loads(response.content)['data']['id']
        # violate database constraint: get a course_identifier uniqueness error
        response = self.client.post(
            self.courses_url, data=json.dumps(COURSE_POST), **HEADERS)
        self.assertEqual(response.status_code, 400, msg=response.content)
        # expect this error: {"errors":[{"detail": "course with this course identifier already exists."}]}
        self.assertEqual('course with this course identifier already exists.',
                         json.loads(response.content)['errors'][0]['detail'])
        response = self.client.post(self.course_terms_url,
                                    data=json.dumps(COURSE_TERM_POST),
                                    **HEADERS)
        self.assertEqual(response.status_code, 201, msg=response.content)
        term_id = json.loads(response.content)['data']['id']
        # now patch in the relationship
        COURSE_TERM_REL_PATCH['data'][0]['id'] = term_id
        response = self.client.patch(
            self.courses_url + course_id + "/relationships/course_terms/",
            data=json.dumps(COURSE_TERM_REL_PATCH),
            **HEADERS)
        self.assertEqual(response.status_code, 200, msg=response.content)

    def test_post_primary_rel(self):
        """
        I should be able to POST the primary data and relationships together.
        """
        response = self.client.post(
            self.course_terms_url,
            data=json.dumps(COURSE_TERM_POST),
            **HEADERS)
        self.assertEqual(response.status_code, 201, msg=response.content)
        term_id = json.loads(response.content)['data']['id']
        COURSE_POST_WITH_REL['data']['relationships']['course_terms']['data'][
            0]['id'] = term_id
        # print("posting:")
        # pprint(COURSE_POST_WITH_REL)
        response = self.client.post(
            self.courses_url,
            data=json.dumps(COURSE_POST_WITH_REL),
            **HEADERS)
        self.assertEqual(response.status_code, 201, msg=response.content)
        # course_id = json.loads(response.content)['data']['id']
        # print("course_id: {}".format(course_id))
        # print("response:")
        # pprint(json.loads(response.content))
        j = json.loads(
            response.content)['data']['relationships']['course_terms']['data']
        self.assertEqual(len(j), 1, msg="missing relationships")
        self.assertEqual(
            j[0]['id'],
            term_id,
            msg="missing relationship data for {}".format(term_id))

    @skip("test_patch_primary_rel not yet implemented")
    def test_patch_primary_rel(self):
        """
        Make sure we can PATCH the attributes *and* relationships
        """
        # TODO: I should be able to PATCH the primary data and updated relationships.
        pass

    @skip("test_patch_rel not yet implemented")
    def test_patch_rel(self):
        """
        See https://jsonapi.org/format/#crud-updating-resource-relationships
        """
        # TODO: I should be able to PATCH the relationships.

        pass

    def test_page_size(self):
        """
        test rest_framework_json_api.pagination.JsonApiPageNumberPagination: page[size] and page[number]
        """
        response = self.client.get(self.courses_url,
                                   data={"page[size]": 3, "page[number]": 2},
                                   **HEADERS)
        j = json.loads(response.content)
        # pprint(j)
        self.assertEqual(len(j['data']), 3)
        # pprint(j['meta']['pagination'])
        self.assertEqual(j['meta']['pagination']['count'], len(self.courses))
        self.assertEqual(j['meta']['pagination']['page'], 2)
        self.assertEqual(j['meta']['pagination']['pages'],
                         math.ceil(len(self.courses) / 3))
        self.assertEqual(
            j['links']['next'],
            'http://testserver/v1/courses/?page%5Bnumber%5D=3&page%5Bsize%5D=3'
        )

    def test_filter_search(self):
        """
        test keyword search (rest_framework.filters.SearchFilter): filter[all]=keywords
        """
        response = self.client.get(self.courses_url,
                                   data={"filter[search]": "research seminar"},
                                   **HEADERS)
        self.assertEqual(response.status_code, 200, msg=response.content)
        j = json.loads(response.content)
        self.assertGreater(len(j['data']), 0)
        for c in j['data']:
            attr = c['attributes']
            self.assertTrue(
                'research' in (attr['course_name'] + ' ' +
                               attr['course_description']).lower()
                and 'seminar' in (attr['course_name'] + ' ' +
                                  attr['course_description']).lower())

        response = self.client.get(self.courses_url,
                                   data={"filter[search]": "nonesuch"},
                                   **HEADERS)
        self.assertEqual(response.status_code, 200, msg=response.content)
        j = json.loads(response.content)
        self.assertEqual(len(j['data']), 0)

    def test_filter_fields(self):
        """
        test field search (django_filters.rest_framework.DjangoFilterBackend): filter[<field>]=values
        """
        response = self.client.get(self.courses_url,
                                   data={"filter[subject_area_code]": "ANTB"},
                                   **HEADERS)
        self.assertEqual(response.status_code, 200, msg=response.content)
        j = json.loads(response.content)
        self.assertEqual(
            len(j['data']),
            len([
                k for k in self.courses
                if k.subject_area_code == 'ANTB'
            ]))

    def test_filter_fields_union_list(self):
        """
        test field for a list of values (ORed): ?filter[field.in]=ANTB,BIOB,XXXX
        """
        response = self.client.get(self.courses_url,
                                   data={"filter[subject_area_code.in]": "ANTB,BIOB,XXXX"},
                                   **HEADERS)
        j = response.json()
        self.assertEqual(
            len(j['data']),
            len([
                k for k in self.courses
                if k.subject_area_code == 'ANTB'
            ]) + len([
                k for k in self.courses
                if k.subject_area_code == 'BIOB'
            ]) + len([
                k for k in self.courses
                if k.subject_area_code == 'XXXX'
            ]),
            msg="filter field list (union)")

    def test_filter_fields_intersection(self):
        """
        test fields (ANDed): ?filter[subject_area_code]=ANTB&filter[course_number]=1234
        """
        response = self.client.get(self.courses_url,
                                   data={"filter[subject_area_code]": "ANTB",
                                         "filter[school_bulletin_prefix_code]": "XCEFK9"},
                                   **HEADERS)
        self.assertEqual(response.status_code, 200)
        j = json.loads(response.content)
        self.assertEqual(
            len(j['data']),
            len([
                k for k in self.courses
                if k.subject_area_code == 'ANTB'
                and k.school_bulletin_prefix_code == 'XCEFK9'
            ]))

    def test_sparse_fieldsets(self):
        """
        test sparse fieldsets
        """
        response = self.client.get("{}{}/".format(self.courses_url, self.courses[5].id),
                                   data={"fields[courses]": "course_name,course_description"},
                                   **HEADERS)
        self.assertEqual(response.status_code, 200)
        j = json.loads(response.content)
        self.assertEqual(len(j['data']['attributes']), 2)
        self.assertIn('course_name', j['data']['attributes'])
        self.assertIn('course_description', j['data']['attributes'])

    def test_sort(self):
        """
        test sort
        """
        response = self.client.get(self.courses_url,
                                   data={"sort": "subject_area_code,-course_number"},
                                   **HEADERS)
        j = json.loads(response.content)
        areas = [c['attributes']['subject_area_code'] for c in j['data']]
        sorted_areas = [
            c['attributes']['subject_area_code'] for c in j['data']
        ]
        sorted_areas.sort()
        self.assertEqual(areas, sorted_areas)
        prev_area = None
        prev_code = None
        for c in j['data']:
            area = c['attributes']['subject_area_code']
            code = c['attributes']['course_number']
            if area == prev_area:
                self.assertLess(code, prev_code)
            prev_code = code
            prev_area = area

    def test_sort_badfield(self):
        """
        test sort of nonexistent field
        """
        response = self.client.get(self.courses_url,
                                   data={"sort": "nonesuch,-not_a_field,subject_area_code"},
                                   **HEADERS)
        self.assertEqual(response.status_code, 400, msg=response.content)

    def test_include(self):
        """
        test include
        """
        response = self.client.get(self.courses_url,
                                   data={"include": "course_terms"},
                                   **HEADERS)
        j = response.json()
        self.assertIn('included', j)
        self.assertEqual(len(j['included']), sum([len(k.course_terms.all()) for k in self.courses]))
        kids = [str(k.id) for k in self.courses]
        for i in j['included']:
            self.assertIn(i['relationships']['course']['data']['id'], kids)

    def test_related_course_course_terms(self):
        """
        test toMany relationship and related links for courses.related.course_terms
        """
        # look up a random course
        course_response = self.client.get("{}{}/".format(self.courses_url, self.courses[5].id),
                                          **HEADERS)
        self.assertEqual(course_response.status_code, 200, msg=course_response.content)
        course = course_response.json()

        # check the relationship link /courses/<id>/relationships/course_terms/
        relationship_link = course['data']['relationships']['course_terms']['links']['self']
        relationship_response = self.client.get(relationship_link, **HEADERS)
        self.assertEqual(relationship_response.status_code, 200, msg=relationship_response.content)
        relationship = relationship_response.json()
        # check the self link:
        self.assertEqual(relationship_link, relationship['links']['self'])
        # confirm that the list of relationships returned for URL /courses/<id>/ matches the list
        # for URL /courses/<id>/relationships/course_terms/
        self.assertEqual(course['data']['relationships']['course_terms']['data'],
                         relationship['data'],
                         msg="course relationships data and self link data mismatch")

        # check the related link /courses/<id>/course_terms/
        related_link = course['data']['relationships']['course_terms']['links']['related']
        related_response = self.client.get(related_link, **HEADERS)
        self.assertEqual(related_response.status_code, 200, msg=related_response.content)
        related = related_response.json()
        # compare resource identifiers from course_response with those from the related link URL
        # N.B. the `data` for the former is only [{"type": <type>, "id": <id>}, ...] while the
        # later includes `attributes` and so on. Slice out just the type,id into course_term_res_ids:
        course_term_res_ids = [{'type': k['type'], 'id': k['id']} for k in related['data']]
        self.assertEqual(course['data']['relationships']['course_terms']['data'],
                         course_term_res_ids)
        # self link is not present. Should it be?
        # self.assertEqual(related_link, related['links']['self'])

        # look up the /course_terms/<id>. It should be the same content as the related link's
        course_terms_data = []
        for res_id in course_term_res_ids:
            course_term_response = self.client.get("{}{}/".format(self.course_terms_url, res_id['id']),
                                                   **HEADERS)
            self.assertEqual(course_term_response.status_code, 200, msg=course_term_response.content)
            course_term = course_term_response.json()
            course_terms_data.append(course_term['data'])
        self.assertEqual(course_terms_data, related['data'])

    def test_related_course_terms_course(self):
        """
        test toOne relationship and related links for course_terms.related.course
        """
        # look up a random course_term
        course_term_response = self.client.get("{}{}/".format(self.course_terms_url, self.course_terms[5].id),
                                               **HEADERS)
        self.assertEqual(course_term_response.status_code, 200, msg=course_term_response.content)
        course_term = course_term_response.json()

        # check the relationship link /course_terms/<id>/relationships/course/
        relationship_link = course_term['data']['relationships']['course']['links']['self']
        relationship_response = self.client.get(relationship_link, **HEADERS)
        self.assertEqual(relationship_response.status_code, 200, msg=relationship_response.content)
        relationship = relationship_response.json()
        # check the self link:
        self.assertEqual(relationship_link, relationship['links']['self'])
        # confirm that the list of relationships returned for URL /course_terms/<id>/ matches the list
        # for URL /course_terms/<id>/relationships/course/
        self.assertEqual(course_term['data']['relationships']['course']['data'],
                         relationship['data'],
                         msg="course_term relationships data and self link data mismatch")

        # check the related link /course_terms/<id>/course/
        related_link = course_term['data']['relationships']['course']['links']['related']
        related_response = self.client.get(related_link, **HEADERS)
        self.assertEqual(related_response.status_code, 200, msg=related_response.content)
        related = related_response.json()
        # N.B. data is singular for a toOne relationship
        # Compare resource identifier from course_term_response with those from the related link URL
        # The later includes `attributes` and so on. Slice out just the type,id into course_res_id:
        course_res_id = {'type': related['data']['type'], 'id': related['data']['id']}
        self.assertEqual(course_term['data']['relationships']['course']['data'],
                         course_res_id)

        # look up the /courses/<id>. It should be the same content as the related link's
        course_response = self.client.get("{}{}/".format(self.courses_url, course_res_id['id']),
                                          **HEADERS)
        self.assertEqual(course_response.status_code, 200, msg=course_response.content)
        course = course_response.json()
        self.assertEqual(course['data'], related['data'])

    @expectedFailure
    def test_permission_course_course_terms(self):
        """
        See if permissions are correctly implemented.
        """
        # authenticate as user with no permissions:
        self.client.force_authenticate(user=self.noneuser)
        course_response = self.client.get("{}{}/".format(self.courses_url, self.courses[5].id),
                                          **HEADERS)
        self.assertEqual(course_response.status_code, 403, msg=course_response.content)
        self.assertIn("You do not have permission", course_response.json()['errors'][0]['detail'])

        # authenticate as user with model permission to view course but not course_term
        self.client.force_authenticate(user=self.someuser)
        # Look up a random course. In theory the course_terms should be suppressed. In practice, not so.
        course_response = self.client.get("{}{}/".format(self.courses_url, self.courses[5].id),
                                          data={"include": "course_terms"},
                                          **HEADERS)
        self.assertEqual(course_response.status_code, 200, msg=course_response.content)
        course = course_response.json()
        # this should return zero (I think) but instead course_terms inherits permissions from course.
        self.assertEqual(len(course['data']['relationships']['course_terms']['data']), 0)

        # put back the default user
        self.client.force_authenticate(user=self.read_write_user)

    def test_permission_course_terms(self):
        """
        confirm that `somebody` lacks course_terms view permission
        """
        self.client.force_authenticate(user=self.someuser)
        # Look up a random course. In theory the course_terms should be suppressed. In practice, not so.
        term_response = self.client.get("{}{}/".format(self.course_terms_url, self.course_terms[5].id),
                                        **HEADERS)
        self.assertEqual(term_response.status_code, 403, msg=term_response.content)
        term = term_response.json()
        self.assertIn("You do not have permission", term['errors'][0]['detail'])

        # put back the default user
        self.client.force_authenticate(user=self.read_write_user)

fixtures = ('auth', 'oauth2', 'testcases') class-attribute instance-attribute

setUp()

Set up some test data for the tests.

Note

These users are defined in the 'auth' fixture:

1
2
3
4
5
6
| username | password       | staff   | su        | group memberships |
| -------- | -------------- | -----   | --------- | ----------------- |
| admin    | admin123       | X       | X         | (none)            |
| user1    | user1password1 | X       |           | team-a, team-c    |
| user2    | user2password2 | X       |           | team-a, team-b    |
| user3    | user3password3 | X       |           | (none)            |

Source code in myapp/tests/test_views.py
 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
def setUp(self):
    """
    Set up some test data for the tests.
    !!! Note
        These users are defined in the 'auth' fixture:
        ```
        | username | password       | staff   | su        | group memberships |
        | -------- | -------------- | -----   | --------- | ----------------- |
        | admin    | admin123       | X       | X         | (none)            |
        | user1    | user1password1 | X       |           | team-a, team-c    |
        | user2    | user2password2 | X       |           | team-a, team-b    |
        | user3    | user3password3 | X       |           | (none)            |
        ```
    """
    self.read_write_user = User.objects.filter(username='user1').first()
    # `somebody` can view course but not anything else
    self.someuser = User.objects.create_user('somebody', is_superuser=False)
    self.someuser.user_permissions.add(Permission.objects.get(codename='view_course').id)
    # `nobody` has no permissions
    self.noneuser = User.objects.create_user('nobody', is_superuser=False)
    # most tests just use the read_write_user
    self.user1_token = oauth_models.MyAccessToken(  # nosec B106
        token='User1Token',
        user=self.read_write_user,
        expires=datetime.isoformat(datetime.now(tz=timezone.utc)+timedelta(seconds=3600)),
        scope='auth-columbia demo-djt-sla-bronze read create update openid '
              'profile email https://api.columbia.edu/scope/group',
        userinfo='{"sub": "user1", "given_name": "First", "family_name": "User", "name": "First User", '
                 '"email": "user1@example.com", "https://api.columbia.edu/claim/group": "team-a team-c"}'
    )
    self.user1_token.save()
    # self.client.force_authenticate(user=self.read_write_user)
    HEADERS['Authorization'] = 'Bearer User1Token'
    self.courses = Course.objects.all()
    self.courses_url = reverse('course-list')
    self.course_terms = CourseTerm.objects.all()
    self.course_terms_url = reverse('courseterm-list')

test_post_course_term()

Source code in myapp/tests/test_views.py
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
def test_post_course_term(self):
    response = self.client.post(self.courses_url,
                                data=json.dumps(COURSE_POST),
                                **HEADERS)
    self.assertEqual(response.status_code, 201, msg=response.content)
    course_id = json.loads(response.content)['data']['id']
    # violate database constraint: get a course_identifier uniqueness error
    response = self.client.post(
        self.courses_url, data=json.dumps(COURSE_POST), **HEADERS)
    self.assertEqual(response.status_code, 400, msg=response.content)
    # expect this error: {"errors":[{"detail": "course with this course identifier already exists."}]}
    self.assertEqual('course with this course identifier already exists.',
                     json.loads(response.content)['errors'][0]['detail'])
    response = self.client.post(self.course_terms_url,
                                data=json.dumps(COURSE_TERM_POST),
                                **HEADERS)
    self.assertEqual(response.status_code, 201, msg=response.content)
    term_id = json.loads(response.content)['data']['id']
    # now patch in the relationship
    COURSE_TERM_REL_PATCH['data'][0]['id'] = term_id
    response = self.client.patch(
        self.courses_url + course_id + "/relationships/course_terms/",
        data=json.dumps(COURSE_TERM_REL_PATCH),
        **HEADERS)
    self.assertEqual(response.status_code, 200, msg=response.content)

test_post_primary_rel()

I should be able to POST the primary data and relationships together.

Source code in myapp/tests/test_views.py
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def test_post_primary_rel(self):
    """
    I should be able to POST the primary data and relationships together.
    """
    response = self.client.post(
        self.course_terms_url,
        data=json.dumps(COURSE_TERM_POST),
        **HEADERS)
    self.assertEqual(response.status_code, 201, msg=response.content)
    term_id = json.loads(response.content)['data']['id']
    COURSE_POST_WITH_REL['data']['relationships']['course_terms']['data'][
        0]['id'] = term_id
    # print("posting:")
    # pprint(COURSE_POST_WITH_REL)
    response = self.client.post(
        self.courses_url,
        data=json.dumps(COURSE_POST_WITH_REL),
        **HEADERS)
    self.assertEqual(response.status_code, 201, msg=response.content)
    # course_id = json.loads(response.content)['data']['id']
    # print("course_id: {}".format(course_id))
    # print("response:")
    # pprint(json.loads(response.content))
    j = json.loads(
        response.content)['data']['relationships']['course_terms']['data']
    self.assertEqual(len(j), 1, msg="missing relationships")
    self.assertEqual(
        j[0]['id'],
        term_id,
        msg="missing relationship data for {}".format(term_id))

test_patch_primary_rel()

Make sure we can PATCH the attributes and relationships

Source code in myapp/tests/test_views.py
194
195
196
197
198
199
200
@skip("test_patch_primary_rel not yet implemented")
def test_patch_primary_rel(self):
    """
    Make sure we can PATCH the attributes *and* relationships
    """
    # TODO: I should be able to PATCH the primary data and updated relationships.
    pass

test_patch_rel()

See https://jsonapi.org/format/#crud-updating-resource-relationships

Source code in myapp/tests/test_views.py
202
203
204
205
206
207
208
209
@skip("test_patch_rel not yet implemented")
def test_patch_rel(self):
    """
    See https://jsonapi.org/format/#crud-updating-resource-relationships
    """
    # TODO: I should be able to PATCH the relationships.

    pass

test_page_size()

test rest_framework_json_api.pagination.JsonApiPageNumberPagination: page[size] and page[number]

Source code in myapp/tests/test_views.py
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def test_page_size(self):
    """
    test rest_framework_json_api.pagination.JsonApiPageNumberPagination: page[size] and page[number]
    """
    response = self.client.get(self.courses_url,
                               data={"page[size]": 3, "page[number]": 2},
                               **HEADERS)
    j = json.loads(response.content)
    # pprint(j)
    self.assertEqual(len(j['data']), 3)
    # pprint(j['meta']['pagination'])
    self.assertEqual(j['meta']['pagination']['count'], len(self.courses))
    self.assertEqual(j['meta']['pagination']['page'], 2)
    self.assertEqual(j['meta']['pagination']['pages'],
                     math.ceil(len(self.courses) / 3))
    self.assertEqual(
        j['links']['next'],
        'http://testserver/v1/courses/?page%5Bnumber%5D=3&page%5Bsize%5D=3'
    )

test keyword search (rest_framework.filters.SearchFilter): filter[all]=keywords

Source code in myapp/tests/test_views.py
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def test_filter_search(self):
    """
    test keyword search (rest_framework.filters.SearchFilter): filter[all]=keywords
    """
    response = self.client.get(self.courses_url,
                               data={"filter[search]": "research seminar"},
                               **HEADERS)
    self.assertEqual(response.status_code, 200, msg=response.content)
    j = json.loads(response.content)
    self.assertGreater(len(j['data']), 0)
    for c in j['data']:
        attr = c['attributes']
        self.assertTrue(
            'research' in (attr['course_name'] + ' ' +
                           attr['course_description']).lower()
            and 'seminar' in (attr['course_name'] + ' ' +
                              attr['course_description']).lower())

    response = self.client.get(self.courses_url,
                               data={"filter[search]": "nonesuch"},
                               **HEADERS)
    self.assertEqual(response.status_code, 200, msg=response.content)
    j = json.loads(response.content)
    self.assertEqual(len(j['data']), 0)

test_filter_fields()

test field search (django_filters.rest_framework.DjangoFilterBackend): filter[]=values

Source code in myapp/tests/test_views.py
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def test_filter_fields(self):
    """
    test field search (django_filters.rest_framework.DjangoFilterBackend): filter[<field>]=values
    """
    response = self.client.get(self.courses_url,
                               data={"filter[subject_area_code]": "ANTB"},
                               **HEADERS)
    self.assertEqual(response.status_code, 200, msg=response.content)
    j = json.loads(response.content)
    self.assertEqual(
        len(j['data']),
        len([
            k for k in self.courses
            if k.subject_area_code == 'ANTB'
        ]))

test_filter_fields_union_list()

test field for a list of values (ORed): ?filter[field.in]=ANTB,BIOB,XXXX

Source code in myapp/tests/test_views.py
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
def test_filter_fields_union_list(self):
    """
    test field for a list of values (ORed): ?filter[field.in]=ANTB,BIOB,XXXX
    """
    response = self.client.get(self.courses_url,
                               data={"filter[subject_area_code.in]": "ANTB,BIOB,XXXX"},
                               **HEADERS)
    j = response.json()
    self.assertEqual(
        len(j['data']),
        len([
            k for k in self.courses
            if k.subject_area_code == 'ANTB'
        ]) + len([
            k for k in self.courses
            if k.subject_area_code == 'BIOB'
        ]) + len([
            k for k in self.courses
            if k.subject_area_code == 'XXXX'
        ]),
        msg="filter field list (union)")

test_filter_fields_intersection()

test fields (ANDed): ?filter[subject_area_code]=ANTB&filter[course_number]=1234

Source code in myapp/tests/test_views.py
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def test_filter_fields_intersection(self):
    """
    test fields (ANDed): ?filter[subject_area_code]=ANTB&filter[course_number]=1234
    """
    response = self.client.get(self.courses_url,
                               data={"filter[subject_area_code]": "ANTB",
                                     "filter[school_bulletin_prefix_code]": "XCEFK9"},
                               **HEADERS)
    self.assertEqual(response.status_code, 200)
    j = json.loads(response.content)
    self.assertEqual(
        len(j['data']),
        len([
            k for k in self.courses
            if k.subject_area_code == 'ANTB'
            and k.school_bulletin_prefix_code == 'XCEFK9'
        ]))

test_sparse_fieldsets()

test sparse fieldsets

Source code in myapp/tests/test_views.py
312
313
314
315
316
317
318
319
320
321
322
323
def test_sparse_fieldsets(self):
    """
    test sparse fieldsets
    """
    response = self.client.get("{}{}/".format(self.courses_url, self.courses[5].id),
                               data={"fields[courses]": "course_name,course_description"},
                               **HEADERS)
    self.assertEqual(response.status_code, 200)
    j = json.loads(response.content)
    self.assertEqual(len(j['data']['attributes']), 2)
    self.assertIn('course_name', j['data']['attributes'])
    self.assertIn('course_description', j['data']['attributes'])

test_sort()

test sort

Source code in myapp/tests/test_views.py
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
def test_sort(self):
    """
    test sort
    """
    response = self.client.get(self.courses_url,
                               data={"sort": "subject_area_code,-course_number"},
                               **HEADERS)
    j = json.loads(response.content)
    areas = [c['attributes']['subject_area_code'] for c in j['data']]
    sorted_areas = [
        c['attributes']['subject_area_code'] for c in j['data']
    ]
    sorted_areas.sort()
    self.assertEqual(areas, sorted_areas)
    prev_area = None
    prev_code = None
    for c in j['data']:
        area = c['attributes']['subject_area_code']
        code = c['attributes']['course_number']
        if area == prev_area:
            self.assertLess(code, prev_code)
        prev_code = code
        prev_area = area

test_sort_badfield()

test sort of nonexistent field

Source code in myapp/tests/test_views.py
349
350
351
352
353
354
355
356
def test_sort_badfield(self):
    """
    test sort of nonexistent field
    """
    response = self.client.get(self.courses_url,
                               data={"sort": "nonesuch,-not_a_field,subject_area_code"},
                               **HEADERS)
    self.assertEqual(response.status_code, 400, msg=response.content)

test_include()

test include

Source code in myapp/tests/test_views.py
358
359
360
361
362
363
364
365
366
367
368
369
370
def test_include(self):
    """
    test include
    """
    response = self.client.get(self.courses_url,
                               data={"include": "course_terms"},
                               **HEADERS)
    j = response.json()
    self.assertIn('included', j)
    self.assertEqual(len(j['included']), sum([len(k.course_terms.all()) for k in self.courses]))
    kids = [str(k.id) for k in self.courses]
    for i in j['included']:
        self.assertIn(i['relationships']['course']['data']['id'], kids)

test toMany relationship and related links for courses.related.course_terms

Source code in myapp/tests/test_views.py
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
def test_related_course_course_terms(self):
    """
    test toMany relationship and related links for courses.related.course_terms
    """
    # look up a random course
    course_response = self.client.get("{}{}/".format(self.courses_url, self.courses[5].id),
                                      **HEADERS)
    self.assertEqual(course_response.status_code, 200, msg=course_response.content)
    course = course_response.json()

    # check the relationship link /courses/<id>/relationships/course_terms/
    relationship_link = course['data']['relationships']['course_terms']['links']['self']
    relationship_response = self.client.get(relationship_link, **HEADERS)
    self.assertEqual(relationship_response.status_code, 200, msg=relationship_response.content)
    relationship = relationship_response.json()
    # check the self link:
    self.assertEqual(relationship_link, relationship['links']['self'])
    # confirm that the list of relationships returned for URL /courses/<id>/ matches the list
    # for URL /courses/<id>/relationships/course_terms/
    self.assertEqual(course['data']['relationships']['course_terms']['data'],
                     relationship['data'],
                     msg="course relationships data and self link data mismatch")

    # check the related link /courses/<id>/course_terms/
    related_link = course['data']['relationships']['course_terms']['links']['related']
    related_response = self.client.get(related_link, **HEADERS)
    self.assertEqual(related_response.status_code, 200, msg=related_response.content)
    related = related_response.json()
    # compare resource identifiers from course_response with those from the related link URL
    # N.B. the `data` for the former is only [{"type": <type>, "id": <id>}, ...] while the
    # later includes `attributes` and so on. Slice out just the type,id into course_term_res_ids:
    course_term_res_ids = [{'type': k['type'], 'id': k['id']} for k in related['data']]
    self.assertEqual(course['data']['relationships']['course_terms']['data'],
                     course_term_res_ids)
    # self link is not present. Should it be?
    # self.assertEqual(related_link, related['links']['self'])

    # look up the /course_terms/<id>. It should be the same content as the related link's
    course_terms_data = []
    for res_id in course_term_res_ids:
        course_term_response = self.client.get("{}{}/".format(self.course_terms_url, res_id['id']),
                                               **HEADERS)
        self.assertEqual(course_term_response.status_code, 200, msg=course_term_response.content)
        course_term = course_term_response.json()
        course_terms_data.append(course_term['data'])
    self.assertEqual(course_terms_data, related['data'])

test toOne relationship and related links for course_terms.related.course

Source code in myapp/tests/test_views.py
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
def test_related_course_terms_course(self):
    """
    test toOne relationship and related links for course_terms.related.course
    """
    # look up a random course_term
    course_term_response = self.client.get("{}{}/".format(self.course_terms_url, self.course_terms[5].id),
                                           **HEADERS)
    self.assertEqual(course_term_response.status_code, 200, msg=course_term_response.content)
    course_term = course_term_response.json()

    # check the relationship link /course_terms/<id>/relationships/course/
    relationship_link = course_term['data']['relationships']['course']['links']['self']
    relationship_response = self.client.get(relationship_link, **HEADERS)
    self.assertEqual(relationship_response.status_code, 200, msg=relationship_response.content)
    relationship = relationship_response.json()
    # check the self link:
    self.assertEqual(relationship_link, relationship['links']['self'])
    # confirm that the list of relationships returned for URL /course_terms/<id>/ matches the list
    # for URL /course_terms/<id>/relationships/course/
    self.assertEqual(course_term['data']['relationships']['course']['data'],
                     relationship['data'],
                     msg="course_term relationships data and self link data mismatch")

    # check the related link /course_terms/<id>/course/
    related_link = course_term['data']['relationships']['course']['links']['related']
    related_response = self.client.get(related_link, **HEADERS)
    self.assertEqual(related_response.status_code, 200, msg=related_response.content)
    related = related_response.json()
    # N.B. data is singular for a toOne relationship
    # Compare resource identifier from course_term_response with those from the related link URL
    # The later includes `attributes` and so on. Slice out just the type,id into course_res_id:
    course_res_id = {'type': related['data']['type'], 'id': related['data']['id']}
    self.assertEqual(course_term['data']['relationships']['course']['data'],
                     course_res_id)

    # look up the /courses/<id>. It should be the same content as the related link's
    course_response = self.client.get("{}{}/".format(self.courses_url, course_res_id['id']),
                                      **HEADERS)
    self.assertEqual(course_response.status_code, 200, msg=course_response.content)
    course = course_response.json()
    self.assertEqual(course['data'], related['data'])

test_permission_course_course_terms()

See if permissions are correctly implemented.

Source code in myapp/tests/test_views.py
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
@expectedFailure
def test_permission_course_course_terms(self):
    """
    See if permissions are correctly implemented.
    """
    # authenticate as user with no permissions:
    self.client.force_authenticate(user=self.noneuser)
    course_response = self.client.get("{}{}/".format(self.courses_url, self.courses[5].id),
                                      **HEADERS)
    self.assertEqual(course_response.status_code, 403, msg=course_response.content)
    self.assertIn("You do not have permission", course_response.json()['errors'][0]['detail'])

    # authenticate as user with model permission to view course but not course_term
    self.client.force_authenticate(user=self.someuser)
    # Look up a random course. In theory the course_terms should be suppressed. In practice, not so.
    course_response = self.client.get("{}{}/".format(self.courses_url, self.courses[5].id),
                                      data={"include": "course_terms"},
                                      **HEADERS)
    self.assertEqual(course_response.status_code, 200, msg=course_response.content)
    course = course_response.json()
    # this should return zero (I think) but instead course_terms inherits permissions from course.
    self.assertEqual(len(course['data']['relationships']['course_terms']['data']), 0)

    # put back the default user
    self.client.force_authenticate(user=self.read_write_user)

test_permission_course_terms()

confirm that somebody lacks course_terms view permission

Source code in myapp/tests/test_views.py
487
488
489
490
491
492
493
494
495
496
497
498
499
500
def test_permission_course_terms(self):
    """
    confirm that `somebody` lacks course_terms view permission
    """
    self.client.force_authenticate(user=self.someuser)
    # Look up a random course. In theory the course_terms should be suppressed. In practice, not so.
    term_response = self.client.get("{}{}/".format(self.course_terms_url, self.course_terms[5].id),
                                    **HEADERS)
    self.assertEqual(term_response.status_code, 403, msg=term_response.content)
    term = term_response.json()
    self.assertIn("You do not have permission", term['errors'][0]['detail'])

    # put back the default user
    self.client.force_authenticate(user=self.read_write_user)