replaced unittest assertions pytest assertions (#26544)

This commit is contained in:
Aarif
2021-02-19 11:59:44 +05:00
committed by GitHub
parent 0112339b20
commit a8b9733654
21 changed files with 1209 additions and 1639 deletions

View File

@@ -5,7 +5,7 @@ Tests for Discussion API internal interface
import itertools
from datetime import datetime, timedelta
import pytest
import ddt
import httpretty
import mock
@@ -153,32 +153,28 @@ class GetCourseTest(ForumsEnableMixin, UrlResetMixin, SharedModuleStoreTestCase)
self.request.user = self.user
def test_nonexistent_course(self):
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
get_course(self.request, CourseLocator.from_string("non/existent/course"))
def test_not_enrolled(self):
unenrolled_user = UserFactory.create()
self.request.user = unenrolled_user
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
get_course(self.request, self.course.id)
def test_discussions_disabled(self):
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
get_course(self.request, _discussion_disabled_course_for(self.user).id)
def test_basic(self):
self.assertEqual(
get_course(self.request, self.course.id),
{
"id": six.text_type(self.course.id),
"blackouts": [],
"thread_list_url": "http://testserver/api/discussion/v1/threads/?course_id=x%2Fy%2Fz",
"following_thread_list_url": (
"http://testserver/api/discussion/v1/threads/?course_id=x%2Fy%2Fz&following=True"
),
"topics_url": "http://testserver/api/discussion/v1/course_topics/x/y/z",
}
)
assert get_course(self.request, self.course.id) == {
'id': six.text_type(self.course.id),
'blackouts': [],
'thread_list_url': 'http://testserver/api/discussion/v1/threads/?course_id=x%2Fy%2Fz',
'following_thread_list_url':
'http://testserver/api/discussion/v1/threads/?course_id=x%2Fy%2Fz&following=True',
'topics_url': 'http://testserver/api/discussion/v1/course_topics/x/y/z'
}
@ddt.ddt
@@ -205,13 +201,10 @@ class GetCourseTestBlackouts(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCa
]
modulestore().update_item(self.course, self.user.id)
result = get_course(self.request, self.course.id)
self.assertEqual(
result["blackouts"],
[
{"start": "2015-06-09T00:00:00Z", "end": "2015-06-10T00:00:00Z"},
{"start": "2015-06-11T00:00:00Z", "end": "2015-06-12T00:00:00Z"},
]
)
assert result['blackouts'] == [
{'start': '2015-06-09T00:00:00Z', 'end': '2015-06-10T00:00:00Z'},
{'start': '2015-06-11T00:00:00Z', 'end': '2015-06-12T00:00:00Z'}
]
@ddt.data(None, "not a datetime", "2015", [])
def test_blackout_errors(self, bad_value):
@@ -221,7 +214,7 @@ class GetCourseTestBlackouts(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCa
]
modulestore().update_item(self.course, self.user.id)
result = get_course(self.request, self.course.id)
self.assertEqual(result["blackouts"], [])
assert result['blackouts'] == []
@mock.patch.dict("django.conf.settings.FEATURES", {"DISABLE_START_DATES": False})
@@ -300,18 +293,18 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
return node
def test_nonexistent_course(self):
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
get_course_topics(self.request, CourseLocator.from_string("non/existent/course"))
def test_not_enrolled(self):
unenrolled_user = UserFactory.create()
self.request.user = unenrolled_user
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
self.get_course_topics()
def test_discussions_disabled(self):
_remove_discussion_tab(self.course, self.user.id)
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
self.get_course_topics()
def test_without_courseware(self):
@@ -322,7 +315,7 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
self.make_expected_tree("non-courseware-topic-id", "Test Topic")
],
}
self.assertEqual(actual, expected)
assert actual == expected
def test_with_courseware(self):
self.make_discussion_xblock("courseware-topic-id", "Foo", "Bar")
@@ -339,7 +332,7 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
self.make_expected_tree("non-courseware-topic-id", "Test Topic")
],
}
self.assertEqual(actual, expected)
assert actual == expected
def test_many(self):
with self.store.bulk_operations(self.course.id, emit_signals=False):
@@ -383,7 +376,7 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
self.make_expected_tree("non-courseware-2", "B"),
],
}
self.assertEqual(actual, expected)
assert actual == expected
def test_sort_key(self):
with self.store.bulk_operations(self.course.id, emit_signals=False):
@@ -432,7 +425,7 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
self.make_expected_tree("non-courseware-1", "W"),
],
}
self.assertEqual(actual, expected)
assert actual == expected
def test_access_control(self):
"""
@@ -499,7 +492,7 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
self.make_expected_tree("non-courseware-topic-id", "Test Topic"),
],
}
self.assertEqual(student_actual, student_expected)
assert student_actual == student_expected
self.request.user = beta_tester
beta_actual = self.get_course_topics()
beta_expected = {
@@ -522,7 +515,7 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
self.make_expected_tree("non-courseware-topic-id", "Test Topic"),
],
}
self.assertEqual(beta_actual, beta_expected)
assert beta_actual == beta_expected
self.request.user = staff
staff_actual = self.get_course_topics()
@@ -550,7 +543,7 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
self.make_expected_tree("non-courseware-topic-id", "Test Topic"),
],
}
self.assertEqual(staff_actual, staff_expected)
assert staff_actual == staff_expected
def test_discussion_topic(self):
"""
@@ -561,41 +554,35 @@ class GetCourseTopicsTest(ForumsEnableMixin, UrlResetMixin, ModuleStoreTestCase)
self.make_discussion_xblock(topic_id_1, "test_category_1", "test_target_1")
self.make_discussion_xblock(topic_id_2, "test_category_2", "test_target_2")
actual = get_course_topics(self.request, self.course.id, {"topic_id_1", "topic_id_2"})
self.assertEqual(
actual,
{
"non_courseware_topics": [],
"courseware_topics": [
{
"children": [{
"children": [],
"id": "topic_id_1",
"thread_list_url": "http://testserver/api/discussion/v1/threads/?"
"course_id=x%2Fy%2Fz&topic_id=topic_id_1",
"name": "test_target_1"
}],
"id": None,
"thread_list_url": "http://testserver/api/discussion/v1/threads/?"
"course_id=x%2Fy%2Fz&topic_id=topic_id_1",
"name": "test_category_1"
},
{
"children":
[{
"children": [],
"id": "topic_id_2",
"thread_list_url": "http://testserver/api/discussion/v1/threads/?"
"course_id=x%2Fy%2Fz&topic_id=topic_id_2",
"name": "test_target_2"
}],
"id": None,
"thread_list_url": "http://testserver/api/discussion/v1/threads/?"
"course_id=x%2Fy%2Fz&topic_id=topic_id_2",
"name": "test_category_2"
}
]
}
)
assert actual == {
'non_courseware_topics': [],
'courseware_topics': [
{'children': [
{'children': [],
'id': 'topic_id_1',
'thread_list_url':
'http://testserver/api/discussion/v1/threads/?course_id=x%2Fy%2Fz&topic_id=topic_id_1',
'name': 'test_target_1'}
],
'id': None,
'thread_list_url':
'http://testserver/api/discussion/v1/threads/?course_id=x%2Fy%2Fz&topic_id=topic_id_1',
'name': 'test_category_1'
},
{'children': [
{'children': [],
'id': 'topic_id_2',
'thread_list_url':
'http://testserver/api/discussion/v1/threads/?course_id=x%2Fy%2Fz&topic_id=topic_id_2',
'name': 'test_target_2'}
],
'id': None,
'thread_list_url':
'http://testserver/api/discussion/v1/threads/?course_id=x%2Fy%2Fz&topic_id=topic_id_2',
'name': 'test_category_2'
}
]
}
@ddt.ddt
@@ -645,36 +632,35 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
return ret
def test_nonexistent_course(self):
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
get_thread_list(self.request, CourseLocator.from_string("non/existent/course"), 1, 1)
def test_not_enrolled(self):
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
self.get_thread_list([])
def test_discussions_disabled(self):
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
self.get_thread_list([], course=_discussion_disabled_course_for(self.user))
def test_empty(self):
self.assertEqual(
self.get_thread_list([], num_pages=0).data,
{
"pagination": {
"next": None,
"previous": None,
"num_pages": 0,
"count": 0
},
"results": [],
"text_search_rewrite": None,
}
)
assert self.get_thread_list(
[], num_pages=0
).data == {
'pagination': {
'next': None,
'previous': None,
'num_pages': 0,
'count': 0
},
'results': [],
'text_search_rewrite': None
}
def test_get_threads_by_topic_id(self):
self.get_thread_list([], topic_id_list=["topic_x", "topic_meow"])
self.assertEqual(urlparse(httpretty.last_request().path).path, "/api/v1/threads") # lint-amnesty, pylint: disable=no-member
assert urlparse(httpretty.last_request().path).path == '/api/v1/threads' # lint-amnesty, pylint: disable=no-member
self.assert_last_query_params({
"user_id": [six.text_type(self.user.id)],
"course_id": [six.text_type(self.course.id)],
@@ -774,10 +760,7 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
results=expected_threads, count=2, num_pages=1, next_link=None, previous_link=None
)
expected_result.update({"text_search_rewrite": None})
self.assertEqual(
self.get_thread_list(source_threads).data,
expected_result
)
assert self.get_thread_list(source_threads).data == expected_result
@ddt.data(
*itertools.product(
@@ -799,7 +782,7 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
self.get_thread_list([], course=cohort_course)
actual_has_group = "group_id" in httpretty.last_request().querystring # lint-amnesty, pylint: disable=no-member
expected_has_group = (course_is_cohorted and role_name == FORUM_ROLE_STUDENT)
self.assertEqual(actual_has_group, expected_has_group)
assert actual_has_group == expected_has_group
def test_pagination(self):
# N.B. Empty thread list is not realistic but convenient for this test
@@ -807,10 +790,7 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
results=[], count=0, num_pages=3, next_link="http://testserver/test_path?page=2", previous_link=None
)
expected_result.update({"text_search_rewrite": None})
self.assertEqual(
self.get_thread_list([], page=1, num_pages=3).data,
expected_result
)
assert self.get_thread_list([], page=1, num_pages=3).data == expected_result
expected_result = make_paginated_api_response(
results=[],
@@ -820,23 +800,17 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
previous_link="http://testserver/test_path?page=1"
)
expected_result.update({"text_search_rewrite": None})
self.assertEqual(
self.get_thread_list([], page=2, num_pages=3).data,
expected_result
)
assert self.get_thread_list([], page=2, num_pages=3).data == expected_result
expected_result = make_paginated_api_response(
results=[], count=0, num_pages=3, next_link=None, previous_link="http://testserver/test_path?page=2"
)
expected_result.update({"text_search_rewrite": None})
self.assertEqual(
self.get_thread_list([], page=3, num_pages=3).data,
expected_result
)
assert self.get_thread_list([], page=3, num_pages=3).data == expected_result
# Test page past the last one
self.register_get_threads_response([], page=3, num_pages=3)
with self.assertRaises(PageNotFoundError):
with pytest.raises(PageNotFoundError):
get_thread_list(self.request, self.course.id, page=4, page_size=10)
@ddt.data(None, "rewritten search string")
@@ -846,16 +820,13 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
)
expected_result.update({"text_search_rewrite": text_search_rewrite})
self.register_get_threads_search_response([], text_search_rewrite, num_pages=0)
self.assertEqual(
get_thread_list(
self.request,
self.course.id,
page=1,
page_size=10,
text_search="test search string"
).data,
expected_result
)
assert get_thread_list(
self.request,
self.course.id,
page=1,
page_size=10,
text_search='test search string'
).data == expected_result
self.assert_last_query_params({
"user_id": [six.text_type(self.user.id)],
"course_id": [six.text_type(self.course.id)],
@@ -879,14 +850,10 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
results=[], count=0, num_pages=0, next_link=None, previous_link=None
)
expected_result.update({"text_search_rewrite": None})
self.assertEqual(
result,
expected_result
)
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/users/{}/subscribed_threads".format(self.user.id)
)
assert result == expected_result
assert urlparse(
httpretty.last_request().path # lint-amnesty, pylint: disable=no-member
).path == '/api/v1/users/{}/subscribed_threads'.format(self.user.id)
self.assert_last_query_params({
"user_id": [six.text_type(self.user.id)],
"course_id": [six.text_type(self.course.id)],
@@ -910,14 +877,8 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
results=[], count=0, num_pages=0, next_link=None, previous_link=None
)
expected_result.update({"text_search_rewrite": None})
self.assertEqual(
result,
expected_result
)
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/threads"
)
assert result == expected_result
assert urlparse(httpretty.last_request().path).path == '/api/v1/threads' # lint-amnesty, pylint: disable=no-member
self.assert_last_query_params({
"user_id": [six.text_type(self.user.id)],
"course_id": [six.text_type(self.course.id)],
@@ -954,11 +915,8 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
results=[], count=0, num_pages=0, next_link=None, previous_link=None
)
expected_result.update({"text_search_rewrite": None})
self.assertEqual(result, expected_result)
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/threads"
)
assert result == expected_result
assert urlparse(httpretty.last_request().path).path == '/api/v1/threads' # lint-amnesty, pylint: disable=no-member
self.assert_last_query_params({
"user_id": [six.text_type(self.user.id)],
"course_id": [six.text_type(self.course.id)],
@@ -985,11 +943,8 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
results=[], count=0, num_pages=0, next_link=None, previous_link=None
)
expected_result.update({"text_search_rewrite": None})
self.assertEqual(result, expected_result)
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/threads"
)
assert result == expected_result
assert urlparse(httpretty.last_request().path).path == '/api/v1/threads' # lint-amnesty, pylint: disable=no-member
self.assert_last_query_params({
"user_id": [six.text_type(self.user.id)],
"course_id": [six.text_type(self.course.id)],
@@ -1002,7 +957,7 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
"""
Test with invalid order_direction (e.g. "asc")
"""
with self.assertRaises(ValidationError) as assertion:
with pytest.raises(ValidationError) as assertion:
self.register_get_threads_response([], page=1, num_pages=0)
get_thread_list( # pylint: disable=expression-not-assigned
self.request,
@@ -1011,7 +966,7 @@ class GetThreadListTest(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetMix
page_size=11,
order_direction="asc",
).data
self.assertIn("order_direction", assertion.exception.message_dict)
assert 'order_direction' in assertion.value.message_dict
@ddt.ddt
@@ -1059,21 +1014,21 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
def test_nonexistent_thread(self):
thread_id = "nonexistent_thread"
self.register_get_thread_error_response(thread_id, 404)
with self.assertRaises(ThreadNotFoundError):
with pytest.raises(ThreadNotFoundError):
get_comment_list(self.request, thread_id, endorsed=False, page=1, page_size=1)
def test_nonexistent_course(self):
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
self.get_comment_list(self.make_minimal_cs_thread({"course_id": "non/existent/course"}))
def test_not_enrolled(self):
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
self.get_comment_list(self.make_minimal_cs_thread())
def test_discussions_disabled(self):
disabled_course = _discussion_disabled_course_for(self.user)
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
self.get_comment_list(
self.make_minimal_cs_thread(
overrides={"course_id": six.text_type(disabled_course.id)}
@@ -1128,41 +1083,33 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
)
try:
self.get_comment_list(thread)
self.assertFalse(expected_error)
assert not expected_error
except ThreadNotFoundError:
self.assertTrue(expected_error)
assert expected_error
@ddt.data(True, False)
def test_discussion_endorsed(self, endorsed_value):
with self.assertRaises(ValidationError) as assertion:
with pytest.raises(ValidationError) as assertion:
self.get_comment_list(
self.make_minimal_cs_thread({"thread_type": "discussion"}),
endorsed=endorsed_value
)
self.assertEqual(
assertion.exception.message_dict,
{"endorsed": ["This field may not be specified for discussion threads."]}
)
assert assertion.value.message_dict == {'endorsed': ['This field may not be specified for discussion threads.']}
def test_question_without_endorsed(self):
with self.assertRaises(ValidationError) as assertion:
with pytest.raises(ValidationError) as assertion:
self.get_comment_list(
self.make_minimal_cs_thread({"thread_type": "question"}),
endorsed=None
)
self.assertEqual(
assertion.exception.message_dict,
{"endorsed": ["This field is required for question threads."]}
)
assert assertion.value.message_dict == {'endorsed': ['This field is required for question threads.']}
def test_empty(self):
discussion_thread = self.make_minimal_cs_thread(
{"thread_type": "discussion", "children": [], "resp_total": 0}
)
self.assertEqual(
self.get_comment_list(discussion_thread).data,
make_paginated_api_response(results=[], count=0, num_pages=1, next_link=None, previous_link=None)
)
assert self.get_comment_list(discussion_thread).data == make_paginated_api_response(
results=[], count=0, num_pages=1, next_link=None, previous_link=None)
question_thread = self.make_minimal_cs_thread({
"thread_type": "question",
@@ -1170,14 +1117,10 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
"non_endorsed_responses": [],
"non_endorsed_resp_total": 0
})
self.assertEqual(
self.get_comment_list(question_thread, endorsed=False).data,
make_paginated_api_response(results=[], count=0, num_pages=1, next_link=None, previous_link=None)
)
self.assertEqual(
self.get_comment_list(question_thread, endorsed=True).data,
make_paginated_api_response(results=[], count=0, num_pages=1, next_link=None, previous_link=None)
)
assert self.get_comment_list(question_thread, endorsed=False).data == make_paginated_api_response(
results=[], count=0, num_pages=1, next_link=None, previous_link=None)
assert self.get_comment_list(question_thread, endorsed=True).data == make_paginated_api_response(
results=[], count=0, num_pages=1, next_link=None, previous_link=None)
def test_basic_query_params(self):
self.get_comment_list(
@@ -1284,7 +1227,7 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
actual_comments = self.get_comment_list(
self.make_minimal_cs_thread({"children": source_comments})
).data["results"]
self.assertEqual(actual_comments, expected_comments)
assert actual_comments == expected_comments
def test_question_content(self):
thread = self.make_minimal_cs_thread({
@@ -1297,10 +1240,10 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
})
endorsed_actual = self.get_comment_list(thread, endorsed=True).data
self.assertEqual(endorsed_actual["results"][0]["id"], "endorsed_comment")
assert endorsed_actual['results'][0]['id'] == 'endorsed_comment'
non_endorsed_actual = self.get_comment_list(thread, endorsed=False).data
self.assertEqual(non_endorsed_actual["results"][0]["id"], "non_endorsed_comment")
assert non_endorsed_actual['results'][0]['id'] == 'non_endorsed_comment'
def test_endorsed_by_anonymity(self):
"""
@@ -1317,7 +1260,7 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
]
})
actual_comments = self.get_comment_list(thread).data["results"]
self.assertIsNone(actual_comments[0]["endorsed_by"])
assert actual_comments[0]['endorsed_by'] is None
@ddt.data(
("discussion", None, "children", "resp_total"),
@@ -1345,23 +1288,23 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
# Only page
actual = self.get_comment_list(thread, endorsed=endorsed_arg, page=1, page_size=5).data
self.assertIsNone(actual["pagination"]["next"])
self.assertIsNone(actual["pagination"]["previous"])
assert actual['pagination']['next'] is None
assert actual['pagination']['previous'] is None
# First page of many
actual = self.get_comment_list(thread, endorsed=endorsed_arg, page=1, page_size=2).data
self.assertEqual(actual["pagination"]["next"], "http://testserver/test_path?page=2")
self.assertIsNone(actual["pagination"]["previous"])
assert actual['pagination']['next'] == 'http://testserver/test_path?page=2'
assert actual['pagination']['previous'] is None
# Middle page of many
actual = self.get_comment_list(thread, endorsed=endorsed_arg, page=2, page_size=2).data
self.assertEqual(actual["pagination"]["next"], "http://testserver/test_path?page=3")
self.assertEqual(actual["pagination"]["previous"], "http://testserver/test_path?page=1")
assert actual['pagination']['next'] == 'http://testserver/test_path?page=3'
assert actual['pagination']['previous'] == 'http://testserver/test_path?page=1'
# Last page of many
actual = self.get_comment_list(thread, endorsed=endorsed_arg, page=3, page_size=2).data
self.assertIsNone(actual["pagination"]["next"])
self.assertEqual(actual["pagination"]["previous"], "http://testserver/test_path?page=2")
assert actual['pagination']['next'] is None
assert actual['pagination']['previous'] == 'http://testserver/test_path?page=2'
# Page past the end
thread = self.make_minimal_cs_thread({
@@ -1369,7 +1312,7 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
response_field: [],
response_total_field: 5
})
with self.assertRaises(PageNotFoundError):
with pytest.raises(PageNotFoundError):
self.get_comment_list(thread, endorsed=endorsed_arg, page=2, page_size=5)
def test_question_endorsed_pagination(self):
@@ -1388,17 +1331,12 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
"""
actual = self.get_comment_list(thread, endorsed=True, page=page, page_size=page_size).data
result_ids = [result["id"] for result in actual["results"]]
self.assertEqual(
result_ids,
["comment_{}".format(i) for i in range(expected_start, expected_stop)]
assert result_ids == ['comment_{}'.format(i) for i in range(expected_start, expected_stop)]
assert actual['pagination']['next'] == (
'http://testserver/test_path?page={}'.format(expected_next) if expected_next else None
)
self.assertEqual(
actual["pagination"]["next"],
"http://testserver/test_path?page={}".format(expected_next) if expected_next else None
)
self.assertEqual(
actual["pagination"]["previous"],
"http://testserver/test_path?page={}".format(expected_prev) if expected_prev else None
assert actual['pagination']['previous'] == (
'http://testserver/test_path?page={}'.format(expected_prev) if expected_prev else None
)
# Only page
@@ -1442,7 +1380,7 @@ class GetCommentListTest(ForumsEnableMixin, CommentsServiceMockMixin, SharedModu
)
# Page past the end
with self.assertRaises(PageNotFoundError):
with pytest.raises(PageNotFoundError):
self.get_comment_list(thread, endorsed=True, page=2, page_size=10)
@@ -1534,39 +1472,33 @@ class CreateThreadTest(
"comment_list_url": "http://testserver/api/discussion/v1/comments/?thread_id=test_id",
"read": True,
})
self.assertEqual(actual, expected)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"commentable_id": ["test_topic"],
"thread_type": ["discussion"],
"title": ["Test Title"],
"body": ["Test body"],
"user_id": [str(self.user.id)],
}
)
assert actual == expected
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'commentable_id': ['test_topic'],
'thread_type': ['discussion'],
'title': ['Test Title'],
'body': ['Test body'],
'user_id': [str(self.user.id)]
}
event_name, event_data = mock_emit.call_args[0]
self.assertEqual(event_name, "edx.forum.thread.created")
self.assertEqual(
event_data,
{
"commentable_id": "test_topic",
"group_id": None,
"thread_type": "discussion",
"title": "Test Title",
"title_truncated": False,
"anonymous": False,
"anonymous_to_peers": False,
"options": {"followed": False},
"id": "test_id",
"truncated": False,
"body": "Test body",
"url": "",
"user_forums_roles": [FORUM_ROLE_STUDENT],
"user_course_roles": [],
}
)
assert event_name == 'edx.forum.thread.created'
assert event_data == {
'commentable_id': 'test_topic',
'group_id': None,
'thread_type': 'discussion',
'title': 'Test Title',
'title_truncated': False,
'anonymous': False,
'anonymous_to_peers': False,
'options': {'followed': False},
'id': 'test_id',
'truncated': False,
'body': 'Test body',
'url': '',
'user_forums_roles': [FORUM_ROLE_STUDENT],
'user_course_roles': []
}
def test_basic_in_blackout_period(self):
"""
@@ -1652,26 +1584,23 @@ class CreateThreadTest(
with self.assert_signal_sent(api, 'thread_created', sender=None, user=self.user, exclude_args=('post',)):
create_thread(self.request, data)
event_name, event_data = mock_emit.call_args[0]
self.assertEqual(event_name, "edx.forum.thread.created")
self.assertEqual(
event_data,
{
"commentable_id": "test_topic",
"group_id": None,
"thread_type": "discussion",
"title": self.LONG_TITLE[:1000],
"title_truncated": True,
"anonymous": False,
"anonymous_to_peers": False,
"options": {"followed": False},
"id": "test_id",
"truncated": False,
"body": "Test body",
"url": "",
"user_forums_roles": [FORUM_ROLE_STUDENT],
"user_course_roles": [],
}
)
assert event_name == 'edx.forum.thread.created'
assert event_data == {
'commentable_id': 'test_topic',
'group_id': None,
'thread_type': 'discussion',
'title': self.LONG_TITLE[:1000],
'title_truncated': True,
'anonymous': False,
'anonymous_to_peers': False,
'options': {'followed': False},
'id': 'test_id',
'truncated': False,
'body': 'Test body',
'url': '',
'user_forums_roles': [FORUM_ROLE_STUDENT],
'user_course_roles': []
}
@ddt.data(
*itertools.product(
@@ -1724,14 +1653,14 @@ class CreateThreadTest(
)
try:
create_thread(self.request, data)
self.assertFalse(expected_error)
assert not expected_error
actual_post_data = httpretty.last_request().parsed_body # lint-amnesty, pylint: disable=no-member
if data_group_state == "group_is_set":
self.assertEqual(actual_post_data["group_id"], [str(data["group_id"])])
assert actual_post_data['group_id'] == [str(data['group_id'])]
elif data_group_state == "no_group_set" and course_is_cohorted and topic_is_cohorted:
self.assertEqual(actual_post_data["group_id"], [str(cohort.id)])
assert actual_post_data['group_id'] == [str(cohort.id)]
else:
self.assertNotIn("group_id", actual_post_data)
assert 'group_id' not in actual_post_data
except ValidationError as ex:
if not expected_error:
self.fail(u"Unexpected validation error: {}".format(ex))
@@ -1742,17 +1671,11 @@ class CreateThreadTest(
data = self.minimal_data.copy()
data["following"] = "True"
result = create_thread(self.request, data)
self.assertEqual(result["following"], True)
assert result['following'] is True
cs_request = httpretty.last_request()
self.assertEqual(
urlparse(cs_request.path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/users/{}/subscriptions".format(self.user.id)
)
self.assertEqual(cs_request.method, "POST")
self.assertEqual(
cs_request.parsed_body, # lint-amnesty, pylint: disable=no-member
{"source_type": ["thread"], "source_id": ["test_id"]}
)
assert urlparse(cs_request.path).path == '/api/v1/users/{}/subscriptions'.format(self.user.id) # lint-amnesty, pylint: disable=no-member
assert cs_request.method == 'POST'
assert cs_request.parsed_body == {'source_type': ['thread'], 'source_id': ['test_id']} # lint-amnesty, pylint: disable=no-member
def test_voted(self):
self.register_post_thread_response({"id": "test_id", "username": self.user.username})
@@ -1761,14 +1684,11 @@ class CreateThreadTest(
data["voted"] = "True"
with self.assert_signal_sent(api, 'thread_voted', sender=None, user=self.user, exclude_args=('post',)):
result = create_thread(self.request, data)
self.assertEqual(result["voted"], True)
assert result['voted'] is True
cs_request = httpretty.last_request()
self.assertEqual(urlparse(cs_request.path).path, "/api/v1/threads/test_id/votes") # lint-amnesty, pylint: disable=no-member
self.assertEqual(cs_request.method, "PUT")
self.assertEqual(
cs_request.parsed_body, # lint-amnesty, pylint: disable=no-member
{"user_id": [str(self.user.id)], "value": ["up"]}
)
assert urlparse(cs_request.path).path == '/api/v1/threads/test_id/votes' # lint-amnesty, pylint: disable=no-member
assert cs_request.method == 'PUT'
assert cs_request.parsed_body == {'user_id': [str(self.user.id)], 'value': ['up']} # lint-amnesty, pylint: disable=no-member
def test_abuse_flagged(self):
self.register_post_thread_response({"id": "test_id", "username": self.user.username})
@@ -1776,41 +1696,41 @@ class CreateThreadTest(
data = self.minimal_data.copy()
data["abuse_flagged"] = "True"
result = create_thread(self.request, data)
self.assertEqual(result["abuse_flagged"], True)
assert result['abuse_flagged'] is True
cs_request = httpretty.last_request()
self.assertEqual(urlparse(cs_request.path).path, "/api/v1/threads/test_id/abuse_flag") # lint-amnesty, pylint: disable=no-member
self.assertEqual(cs_request.method, "PUT")
self.assertEqual(cs_request.parsed_body, {"user_id": [str(self.user.id)]}) # lint-amnesty, pylint: disable=no-member
assert urlparse(cs_request.path).path == '/api/v1/threads/test_id/abuse_flag' # lint-amnesty, pylint: disable=no-member
assert cs_request.method == 'PUT'
assert cs_request.parsed_body == {'user_id': [str(self.user.id)]} # lint-amnesty, pylint: disable=no-member
def test_course_id_missing(self):
with self.assertRaises(ValidationError) as assertion:
with pytest.raises(ValidationError) as assertion:
create_thread(self.request, {})
self.assertEqual(assertion.exception.message_dict, {"course_id": ["This field is required."]})
assert assertion.value.message_dict == {'course_id': ['This field is required.']}
def test_course_id_invalid(self):
with self.assertRaises(ValidationError) as assertion:
with pytest.raises(ValidationError) as assertion:
create_thread(self.request, {"course_id": "invalid!"})
self.assertEqual(assertion.exception.message_dict, {"course_id": ["Invalid value."]})
assert assertion.value.message_dict == {'course_id': ['Invalid value.']}
def test_nonexistent_course(self):
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
create_thread(self.request, {"course_id": "non/existent/course"})
def test_not_enrolled(self):
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
create_thread(self.request, self.minimal_data)
def test_discussions_disabled(self):
disabled_course = _discussion_disabled_course_for(self.user)
self.minimal_data["course_id"] = six.text_type(disabled_course.id)
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
create_thread(self.request, self.minimal_data)
def test_invalid_field(self):
data = self.minimal_data.copy()
data["type"] = "invalid_type"
with self.assertRaises(ValidationError):
with pytest.raises(ValidationError):
create_thread(self.request, data)
@@ -1893,23 +1813,17 @@ class CreateCommentTest(
"editable_fields": ["abuse_flagged", "raw_body", "voted"],
"child_count": 0,
}
self.assertEqual(actual, expected)
assert actual == expected
expected_url = (
"/api/v1/comments/{}".format(parent_id) if parent_id else
"/api/v1/threads/test_thread/comments"
)
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
expected_url
)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"body": ["Test body"],
"user_id": [str(self.user.id)]
}
)
assert urlparse(httpretty.last_request().path).path == expected_url # lint-amnesty, pylint: disable=no-member
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'body': ['Test body'],
'user_id': [str(self.user.id)]
}
expected_event_name = (
"edx.forum.comment.created" if parent_id else
"edx.forum.response.created"
@@ -1928,8 +1842,8 @@ class CreateCommentTest(
if parent_id:
expected_event_data["response"] = {"id": parent_id}
actual_event_name, actual_event_data = mock_emit.call_args[0]
self.assertEqual(actual_event_name, expected_event_name)
self.assertEqual(actual_event_data, expected_event_data)
assert actual_event_name == expected_event_name
assert actual_event_data == expected_event_data
@ddt.data(None, "test_parent")
@mock.patch("eventtracking.tracker.emit")
@@ -2060,10 +1974,10 @@ class CreateCommentTest(
)
try:
create_comment(self.request, data)
self.assertEqual(httpretty.last_request().parsed_body["endorsed"], ["True"]) # lint-amnesty, pylint: disable=no-member
self.assertFalse(expected_error)
assert httpretty.last_request().parsed_body['endorsed'] == ['True'] # lint-amnesty, pylint: disable=no-member
assert not expected_error
except ValidationError:
self.assertTrue(expected_error)
assert expected_error
def test_voted(self):
self.register_post_comment_response({"id": "test_comment", "username": self.user.username}, "test_thread")
@@ -2072,14 +1986,11 @@ class CreateCommentTest(
data["voted"] = "True"
with self.assert_signal_sent(api, 'comment_voted', sender=None, user=self.user, exclude_args=('post',)):
result = create_comment(self.request, data)
self.assertEqual(result["voted"], True)
assert result['voted'] is True
cs_request = httpretty.last_request()
self.assertEqual(urlparse(cs_request.path).path, "/api/v1/comments/test_comment/votes") # lint-amnesty, pylint: disable=no-member
self.assertEqual(cs_request.method, "PUT")
self.assertEqual(
cs_request.parsed_body, # lint-amnesty, pylint: disable=no-member
{"user_id": [str(self.user.id)], "value": ["up"]}
)
assert urlparse(cs_request.path).path == '/api/v1/comments/test_comment/votes' # lint-amnesty, pylint: disable=no-member
assert cs_request.method == 'PUT'
assert cs_request.parsed_body == {'user_id': [str(self.user.id)], 'value': ['up']} # lint-amnesty, pylint: disable=no-member
def test_abuse_flagged(self):
self.register_post_comment_response({"id": "test_comment", "username": self.user.username}, "test_thread")
@@ -2087,32 +1998,32 @@ class CreateCommentTest(
data = self.minimal_data.copy()
data["abuse_flagged"] = "True"
result = create_comment(self.request, data)
self.assertEqual(result["abuse_flagged"], True)
assert result['abuse_flagged'] is True
cs_request = httpretty.last_request()
self.assertEqual(urlparse(cs_request.path).path, "/api/v1/comments/test_comment/abuse_flag") # lint-amnesty, pylint: disable=no-member
self.assertEqual(cs_request.method, "PUT")
self.assertEqual(cs_request.parsed_body, {"user_id": [str(self.user.id)]}) # lint-amnesty, pylint: disable=no-member
assert urlparse(cs_request.path).path == '/api/v1/comments/test_comment/abuse_flag' # lint-amnesty, pylint: disable=no-member
assert cs_request.method == 'PUT'
assert cs_request.parsed_body == {'user_id': [str(self.user.id)]} # lint-amnesty, pylint: disable=no-member
def test_thread_id_missing(self):
with self.assertRaises(ValidationError) as assertion:
with pytest.raises(ValidationError) as assertion:
create_comment(self.request, {})
self.assertEqual(assertion.exception.message_dict, {"thread_id": ["This field is required."]})
assert assertion.value.message_dict == {'thread_id': ['This field is required.']}
def test_thread_id_not_found(self):
self.register_get_thread_error_response("test_thread", 404)
with self.assertRaises(ThreadNotFoundError):
with pytest.raises(ThreadNotFoundError):
create_comment(self.request, self.minimal_data)
def test_nonexistent_course(self):
self.register_get_thread_response(
make_minimal_cs_thread({"id": "test_thread", "course_id": "non/existent/course"})
)
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
create_comment(self.request, self.minimal_data)
def test_not_enrolled(self):
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
create_comment(self.request, self.minimal_data)
def test_discussions_disabled(self):
@@ -2124,7 +2035,7 @@ class CreateCommentTest(
"commentable_id": "test_topic",
})
)
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
create_comment(self.request, self.minimal_data)
@ddt.data(
@@ -2161,14 +2072,14 @@ class CreateCommentTest(
)
try:
create_comment(self.request, data)
self.assertFalse(expected_error)
assert not expected_error
except ThreadNotFoundError:
self.assertTrue(expected_error)
assert expected_error
def test_invalid_field(self):
data = self.minimal_data.copy()
del data["raw_body"]
with self.assertRaises(ValidationError):
with pytest.raises(ValidationError):
create_comment(self.request, data)
@@ -2230,57 +2141,54 @@ class UpdateThreadTest(
self.register_thread()
update_thread(self.request, "test_thread", {})
for request in httpretty.httpretty.latest_requests:
self.assertEqual(request.method, "GET")
assert request.method == 'GET'
def test_basic(self):
self.register_thread()
with self.assert_signal_sent(api, 'thread_edited', sender=None, user=self.user, exclude_args=('post',)):
actual = update_thread(self.request, "test_thread", {"raw_body": "Edited body"})
self.assertEqual(actual, self.expected_thread_data({
"raw_body": "Edited body",
"rendered_body": "<p>Edited body</p>",
"topic_id": "original_topic",
"read": True,
"title": "Original Title",
}))
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"commentable_id": ["original_topic"],
"thread_type": ["discussion"],
"title": ["Original Title"],
"body": ["Edited body"],
"user_id": [str(self.user.id)],
"anonymous": ["False"],
"anonymous_to_peers": ["False"],
"closed": ["False"],
"pinned": ["False"],
"read": ["False"],
}
)
assert actual == self.expected_thread_data({
'raw_body': 'Edited body',
'rendered_body': '<p>Edited body</p>',
'topic_id': 'original_topic',
'read': True,
'title': 'Original Title'
})
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'commentable_id': ['original_topic'],
'thread_type': ['discussion'],
'title': ['Original Title'],
'body': ['Edited body'],
'user_id': [str(self.user.id)],
'anonymous': ['False'],
'anonymous_to_peers': ['False'],
'closed': ['False'],
'pinned': ['False'],
'read': ['False']
}
def test_nonexistent_thread(self):
self.register_get_thread_error_response("test_thread", 404)
with self.assertRaises(ThreadNotFoundError):
with pytest.raises(ThreadNotFoundError):
update_thread(self.request, "test_thread", {})
def test_nonexistent_course(self):
self.register_thread({"course_id": "non/existent/course"})
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
update_thread(self.request, "test_thread", {})
def test_not_enrolled(self):
self.register_thread()
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
update_thread(self.request, "test_thread", {})
def test_discussions_disabled(self):
disabled_course = _discussion_disabled_course_for(self.user)
self.register_thread(overrides={"course_id": six.text_type(disabled_course.id)})
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
update_thread(self.request, "test_thread", {})
@ddt.data(
@@ -2313,9 +2221,9 @@ class UpdateThreadTest(
)
try:
update_thread(self.request, "test_thread", {})
self.assertFalse(expected_error)
assert not expected_error
except ThreadNotFoundError:
self.assertTrue(expected_error)
assert expected_error
@ddt.data(
FORUM_ROLE_ADMINISTRATOR,
@@ -2331,13 +2239,10 @@ class UpdateThreadTest(
expected_error = role_name == FORUM_ROLE_STUDENT
try:
update_thread(self.request, "test_thread", data)
self.assertFalse(expected_error)
assert not expected_error
except ValidationError as err:
self.assertTrue(expected_error)
self.assertEqual(
err.message_dict,
{field: ["This field is not editable."] for field in data.keys()}
)
assert expected_error
assert err.message_dict == {field: ['This field is not editable.'] for field in data.keys()}
@ddt.data(*itertools.product([True, False], [True, False]))
@ddt.unpack
@@ -2357,26 +2262,20 @@ class UpdateThreadTest(
self.register_thread()
data = {"following": new_following}
result = update_thread(self.request, "test_thread", data)
self.assertEqual(result["following"], new_following)
assert result['following'] == new_following
last_request_path = urlparse(httpretty.last_request().path).path # lint-amnesty, pylint: disable=no-member
subscription_url = "/api/v1/users/{}/subscriptions".format(self.user.id)
if old_following == new_following:
self.assertNotEqual(last_request_path, subscription_url)
assert last_request_path != subscription_url
else:
self.assertEqual(last_request_path, subscription_url)
self.assertEqual(
httpretty.last_request().method,
"POST" if new_following else "DELETE"
)
assert last_request_path == subscription_url
assert httpretty.last_request().method == ('POST' if new_following else 'DELETE')
request_data = (
httpretty.last_request().parsed_body if new_following else # lint-amnesty, pylint: disable=no-member
parse_qs(urlparse(httpretty.last_request().path).query) # lint-amnesty, pylint: disable=no-member
)
request_data.pop("request_id", None)
self.assertEqual(
request_data,
{"source_type": ["thread"], "source_id": ["test_thread"]}
)
assert request_data == {'source_type': ['thread'], 'source_id': ['test_thread']}
@ddt.data(*itertools.product([True, False], [True, False]))
@ddt.unpack
@@ -2397,17 +2296,14 @@ class UpdateThreadTest(
self.register_thread()
data = {"voted": new_vote_status}
result = update_thread(self.request, "test_thread", data)
self.assertEqual(result["voted"], new_vote_status)
assert result['voted'] == new_vote_status
last_request_path = urlparse(httpretty.last_request().path).path # lint-amnesty, pylint: disable=no-member
votes_url = "/api/v1/threads/test_thread/votes"
if current_vote_status == new_vote_status:
self.assertNotEqual(last_request_path, votes_url)
assert last_request_path != votes_url
else:
self.assertEqual(last_request_path, votes_url)
self.assertEqual(
httpretty.last_request().method,
"PUT" if new_vote_status else "DELETE"
)
assert last_request_path == votes_url
assert httpretty.last_request().method == ('PUT' if new_vote_status else 'DELETE')
actual_request_data = (
httpretty.last_request().parsed_body if new_vote_status else # lint-amnesty, pylint: disable=no-member
parse_qs(urlparse(httpretty.last_request().path).query) # lint-amnesty, pylint: disable=no-member
@@ -2416,23 +2312,20 @@ class UpdateThreadTest(
expected_request_data = {"user_id": [str(self.user.id)]}
if new_vote_status:
expected_request_data["value"] = ["up"]
self.assertEqual(actual_request_data, expected_request_data)
assert actual_request_data == expected_request_data
event_name, event_data = mock_emit.call_args[0]
self.assertEqual(event_name, "edx.forum.thread.voted")
self.assertEqual(
event_data,
{
'undo_vote': not new_vote_status,
'url': '',
'target_username': self.user.username,
'vote_value': 'up',
'user_forums_roles': [FORUM_ROLE_STUDENT],
'user_course_roles': [],
'commentable_id': 'original_topic',
'id': 'test_thread'
}
)
assert event_name == 'edx.forum.thread.voted'
assert event_data == {
'undo_vote': (not new_vote_status),
'url': '',
'target_username': self.user.username,
'vote_value': 'up',
'user_forums_roles': [FORUM_ROLE_STUDENT],
'user_course_roles': [],
'commentable_id': 'original_topic',
'id': 'test_thread'
}
@ddt.data(*itertools.product([True, False], [True, False], [True, False]))
@ddt.unpack
@@ -2452,12 +2345,12 @@ class UpdateThreadTest(
data = {"voted": first_vote}
result = update_thread(self.request, "test_thread", data)
self.register_thread(overrides={"voted": first_vote})
self.assertEqual(result["vote_count"], 1 if first_vote else 0)
assert result['vote_count'] == (1 if first_vote else 0)
#second vote
data = {"voted": second_vote}
result = update_thread(self.request, "test_thread", data)
self.assertEqual(result["vote_count"], 1 if second_vote else 0)
assert result['vote_count'] == (1 if second_vote else 0)
@ddt.data(*itertools.product([True, False], [True, False], [True, False], [True, False]))
@ddt.unpack
@@ -2496,14 +2389,14 @@ class UpdateThreadTest(
data = {"voted": user_vote}
result = update_thread(request, "test_thread", data)
if current_vote == user_vote:
self.assertEqual(result["vote_count"], vote_count)
assert result['vote_count'] == vote_count
elif user_vote:
vote_count += 1
self.assertEqual(result["vote_count"], vote_count)
assert result['vote_count'] == vote_count
self.register_get_user_response(self.user, upvoted_ids=["test_thread"])
else:
vote_count -= 1
self.assertEqual(result["vote_count"], vote_count)
assert result['vote_count'] == vote_count
self.register_get_user_response(self.user, upvoted_ids=[])
@ddt.data(*itertools.product([True, False], [True, False]))
@@ -2523,32 +2416,23 @@ class UpdateThreadTest(
self.register_thread({"abuse_flaggers": [str(self.user.id)] if old_flagged else []})
data = {"abuse_flagged": new_flagged}
result = update_thread(self.request, "test_thread", data)
self.assertEqual(result["abuse_flagged"], new_flagged)
assert result['abuse_flagged'] == new_flagged
last_request_path = urlparse(httpretty.last_request().path).path # lint-amnesty, pylint: disable=no-member
flag_url = "/api/v1/threads/test_thread/abuse_flag"
unflag_url = "/api/v1/threads/test_thread/abuse_unflag"
if old_flagged == new_flagged:
self.assertNotEqual(last_request_path, flag_url)
self.assertNotEqual(last_request_path, unflag_url)
assert last_request_path != flag_url
assert last_request_path != unflag_url
else:
self.assertEqual(
last_request_path,
flag_url if new_flagged else unflag_url
)
self.assertEqual(httpretty.last_request().method, "PUT")
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{"user_id": [str(self.user.id)]}
)
assert last_request_path == (flag_url if new_flagged else unflag_url)
assert httpretty.last_request().method == 'PUT'
assert httpretty.last_request().parsed_body == {'user_id': [str(self.user.id)]} # lint-amnesty, pylint: disable=no-member
def test_invalid_field(self):
self.register_thread()
with self.assertRaises(ValidationError) as assertion:
with pytest.raises(ValidationError) as assertion:
update_thread(self.request, "test_thread", {"raw_body": ""})
self.assertEqual(
assertion.exception.message_dict,
{"raw_body": ["This field may not be blank."]}
)
assert assertion.value.message_dict == {'raw_body': ['This field may not be blank.']}
@ddt.ddt
@@ -2618,7 +2502,7 @@ class UpdateCommentTest(
self.register_comment()
update_comment(self.request, "test_comment", {})
for request in httpretty.httpretty.latest_requests:
self.assertEqual(request.method, "GET")
assert request.method == 'GET'
@ddt.data(None, "test_parent")
def test_basic(self, parent_id):
@@ -2646,38 +2530,35 @@ class UpdateCommentTest(
"editable_fields": ["abuse_flagged", "raw_body", "voted"],
"child_count": 0,
}
self.assertEqual(actual, expected)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"body": ["Edited body"],
"course_id": [six.text_type(self.course.id)],
"user_id": [str(self.user.id)],
"anonymous": ["False"],
"anonymous_to_peers": ["False"],
"endorsed": ["False"],
}
)
assert actual == expected
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'body': ['Edited body'],
'course_id': [six.text_type(self.course.id)],
'user_id': [str(self.user.id)],
'anonymous': ['False'],
'anonymous_to_peers': ['False'],
'endorsed': ['False']
}
def test_nonexistent_comment(self):
self.register_get_comment_error_response("test_comment", 404)
with self.assertRaises(CommentNotFoundError):
with pytest.raises(CommentNotFoundError):
update_comment(self.request, "test_comment", {})
def test_nonexistent_course(self):
self.register_comment(thread_overrides={"course_id": "non/existent/course"})
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
update_comment(self.request, "test_comment", {})
def test_unenrolled(self):
self.register_comment()
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
update_comment(self.request, "test_comment", {})
def test_discussions_disabled(self):
self.register_comment(course=_discussion_disabled_course_for(self.user))
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
update_comment(self.request, "test_comment", {})
@ddt.data(
@@ -2715,9 +2596,9 @@ class UpdateCommentTest(
)
try:
update_comment(self.request, "test_comment", {})
self.assertFalse(expected_error)
assert not expected_error
except ThreadNotFoundError:
self.assertTrue(expected_error)
assert expected_error
@ddt.data(*itertools.product(
[
@@ -2741,13 +2622,10 @@ class UpdateCommentTest(
expected_error = role_name == FORUM_ROLE_STUDENT and not is_comment_author
try:
update_comment(self.request, "test_comment", {"raw_body": "edited"})
self.assertFalse(expected_error)
assert not expected_error
except ValidationError as err:
self.assertTrue(expected_error)
self.assertEqual(
err.message_dict,
{"raw_body": ["This field is not editable."]}
)
assert expected_error
assert err.message_dict == {'raw_body': ['This field is not editable.']}
@ddt.data(*itertools.product(
[
@@ -2776,13 +2654,10 @@ class UpdateCommentTest(
)
try:
update_comment(self.request, "test_comment", {"endorsed": True})
self.assertFalse(expected_error)
assert not expected_error
except ValidationError as err:
self.assertTrue(expected_error)
self.assertEqual(
err.message_dict,
{"endorsed": ["This field is not editable."]}
)
assert expected_error
assert err.message_dict == {'endorsed': ['This field is not editable.']}
@ddt.data(*itertools.product([True, False], [True, False]))
@ddt.unpack
@@ -2805,18 +2680,15 @@ class UpdateCommentTest(
self.register_comment(overrides={"votes": {"up_count": vote_count}})
data = {"voted": new_vote_status}
result = update_comment(self.request, "test_comment", data)
self.assertEqual(result["vote_count"], 1 if new_vote_status else 0)
self.assertEqual(result["voted"], new_vote_status)
assert result['vote_count'] == (1 if new_vote_status else 0)
assert result['voted'] == new_vote_status
last_request_path = urlparse(httpretty.last_request().path).path # lint-amnesty, pylint: disable=no-member
votes_url = "/api/v1/comments/test_comment/votes"
if current_vote_status == new_vote_status:
self.assertNotEqual(last_request_path, votes_url)
assert last_request_path != votes_url
else:
self.assertEqual(last_request_path, votes_url)
self.assertEqual(
httpretty.last_request().method,
"PUT" if new_vote_status else "DELETE"
)
assert last_request_path == votes_url
assert httpretty.last_request().method == ('PUT' if new_vote_status else 'DELETE')
actual_request_data = (
httpretty.last_request().parsed_body if new_vote_status else # lint-amnesty, pylint: disable=no-member
parse_qs(urlparse(httpretty.last_request().path).query) # lint-amnesty, pylint: disable=no-member
@@ -2825,24 +2697,21 @@ class UpdateCommentTest(
expected_request_data = {"user_id": [str(self.user.id)]}
if new_vote_status:
expected_request_data["value"] = ["up"]
self.assertEqual(actual_request_data, expected_request_data)
assert actual_request_data == expected_request_data
event_name, event_data = mock_emit.call_args[0]
self.assertEqual(event_name, "edx.forum.response.voted")
assert event_name == 'edx.forum.response.voted'
self.assertEqual(
event_data,
{
'undo_vote': not new_vote_status,
'url': '',
'target_username': self.user.username,
'vote_value': 'up',
'user_forums_roles': [FORUM_ROLE_STUDENT],
'user_course_roles': [],
'commentable_id': 'dummy',
'id': 'test_comment'
}
)
assert event_data == {
'undo_vote': (not new_vote_status),
'url': '',
'target_username': self.user.username,
'vote_value': 'up',
'user_forums_roles': [FORUM_ROLE_STUDENT],
'user_course_roles': [],
'commentable_id': 'dummy',
'id': 'test_comment'
}
@ddt.data(*itertools.product([True, False], [True, False], [True, False]))
@ddt.unpack
@@ -2862,12 +2731,12 @@ class UpdateCommentTest(
data = {"voted": first_vote}
result = update_comment(self.request, "test_comment", data)
self.register_comment(overrides={"voted": first_vote})
self.assertEqual(result["vote_count"], 1 if first_vote else 0)
assert result['vote_count'] == (1 if first_vote else 0)
#second vote
data = {"voted": second_vote}
result = update_comment(self.request, "test_comment", data)
self.assertEqual(result["vote_count"], 1 if second_vote else 0)
assert result['vote_count'] == (1 if second_vote else 0)
@ddt.data(*itertools.product([True, False], [True, False], [True, False], [True, False]))
@ddt.unpack
@@ -2905,14 +2774,14 @@ class UpdateCommentTest(
data = {"voted": user_vote}
result = update_comment(request, "test_comment", data)
if current_vote == user_vote:
self.assertEqual(result["vote_count"], vote_count)
assert result['vote_count'] == vote_count
elif user_vote:
vote_count += 1
self.assertEqual(result["vote_count"], vote_count)
assert result['vote_count'] == vote_count
self.register_get_user_response(self.user, upvoted_ids=["test_comment"])
else:
vote_count -= 1
self.assertEqual(result["vote_count"], vote_count)
assert result['vote_count'] == vote_count
self.register_get_user_response(self.user, upvoted_ids=[])
@ddt.data(*itertools.product([True, False], [True, False]))
@@ -2932,23 +2801,17 @@ class UpdateCommentTest(
self.register_comment({"abuse_flaggers": [str(self.user.id)] if old_flagged else []})
data = {"abuse_flagged": new_flagged}
result = update_comment(self.request, "test_comment", data)
self.assertEqual(result["abuse_flagged"], new_flagged)
assert result['abuse_flagged'] == new_flagged
last_request_path = urlparse(httpretty.last_request().path).path # lint-amnesty, pylint: disable=no-member
flag_url = "/api/v1/comments/test_comment/abuse_flag"
unflag_url = "/api/v1/comments/test_comment/abuse_unflag"
if old_flagged == new_flagged:
self.assertNotEqual(last_request_path, flag_url)
self.assertNotEqual(last_request_path, unflag_url)
assert last_request_path != flag_url
assert last_request_path != unflag_url
else:
self.assertEqual(
last_request_path,
flag_url if new_flagged else unflag_url
)
self.assertEqual(httpretty.last_request().method, "PUT")
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{"user_id": [str(self.user.id)]}
)
assert last_request_path == (flag_url if new_flagged else unflag_url)
assert httpretty.last_request().method == 'PUT'
assert httpretty.last_request().parsed_body == {'user_id': [str(self.user.id)]} # lint-amnesty, pylint: disable=no-member
@ddt.ddt
@@ -2999,33 +2862,30 @@ class DeleteThreadTest(
def test_basic(self):
self.register_thread()
with self.assert_signal_sent(api, 'thread_deleted', sender=None, user=self.user, exclude_args=('post',)):
self.assertIsNone(delete_thread(self.request, self.thread_id))
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/threads/{}".format(self.thread_id)
)
self.assertEqual(httpretty.last_request().method, "DELETE")
assert delete_thread(self.request, self.thread_id) is None
assert urlparse(httpretty.last_request().path).path == '/api/v1/threads/{}'.format(self.thread_id) # lint-amnesty, pylint: disable=no-member
assert httpretty.last_request().method == 'DELETE'
def test_thread_id_not_found(self):
self.register_get_thread_error_response("missing_thread", 404)
with self.assertRaises(ThreadNotFoundError):
with pytest.raises(ThreadNotFoundError):
delete_thread(self.request, "missing_thread")
def test_nonexistent_course(self):
self.register_thread({"course_id": "non/existent/course"})
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
delete_thread(self.request, self.thread_id)
def test_not_enrolled(self):
self.register_thread()
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
delete_thread(self.request, self.thread_id)
def test_discussions_disabled(self):
disabled_course = _discussion_disabled_course_for(self.user)
self.register_thread(overrides={"course_id": six.text_type(disabled_course.id)})
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
delete_thread(self.request, self.thread_id)
@ddt.data(
@@ -3040,9 +2900,9 @@ class DeleteThreadTest(
expected_error = role_name == FORUM_ROLE_STUDENT
try:
delete_thread(self.request, self.thread_id)
self.assertFalse(expected_error)
assert not expected_error
except PermissionDenied:
self.assertTrue(expected_error)
assert expected_error
@ddt.data(
*itertools.product(
@@ -3082,9 +2942,9 @@ class DeleteThreadTest(
)
try:
delete_thread(self.request, self.thread_id)
self.assertFalse(expected_error)
assert not expected_error
except ThreadNotFoundError:
self.assertTrue(expected_error)
assert expected_error
@ddt.ddt
@@ -3144,29 +3004,26 @@ class DeleteCommentTest(
def test_basic(self):
self.register_comment_and_thread()
with self.assert_signal_sent(api, 'comment_deleted', sender=None, user=self.user, exclude_args=('post',)):
self.assertIsNone(delete_comment(self.request, self.comment_id))
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/comments/{}".format(self.comment_id)
)
self.assertEqual(httpretty.last_request().method, "DELETE")
assert delete_comment(self.request, self.comment_id) is None
assert urlparse(httpretty.last_request().path).path == '/api/v1/comments/{}'.format(self.comment_id) # lint-amnesty, pylint: disable=no-member
assert httpretty.last_request().method == 'DELETE'
def test_comment_id_not_found(self):
self.register_get_comment_error_response("missing_comment", 404)
with self.assertRaises(CommentNotFoundError):
with pytest.raises(CommentNotFoundError):
delete_comment(self.request, "missing_comment")
def test_nonexistent_course(self):
self.register_comment_and_thread(
thread_overrides={"course_id": "non/existent/course"}
)
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
delete_comment(self.request, self.comment_id)
def test_not_enrolled(self):
self.register_comment_and_thread()
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
delete_comment(self.request, self.comment_id)
def test_discussions_disabled(self):
@@ -3175,7 +3032,7 @@ class DeleteCommentTest(
thread_overrides={"course_id": six.text_type(disabled_course.id)},
overrides={"course_id": six.text_type(disabled_course.id)}
)
with self.assertRaises(DiscussionDisabledError):
with pytest.raises(DiscussionDisabledError):
delete_comment(self.request, self.comment_id)
@ddt.data(
@@ -3192,9 +3049,9 @@ class DeleteCommentTest(
expected_error = role_name == FORUM_ROLE_STUDENT
try:
delete_comment(self.request, self.comment_id)
self.assertFalse(expected_error)
assert not expected_error
except PermissionDenied:
self.assertTrue(expected_error)
assert expected_error
@ddt.data(
*itertools.product(
@@ -3237,9 +3094,9 @@ class DeleteCommentTest(
)
try:
delete_comment(self.request, self.comment_id)
self.assertFalse(expected_error)
assert not expected_error
except ThreadNotFoundError:
self.assertTrue(expected_error)
assert expected_error
@ddt.ddt
@@ -3292,15 +3149,15 @@ class RetrieveThreadTest(
def test_basic(self):
self.register_thread({"resp_total": 2})
self.assertEqual(get_thread(self.request, self.thread_id), self.expected_thread_data({
"response_count": 2,
"unread_comment_count": 1,
}))
self.assertEqual(httpretty.last_request().method, "GET")
assert get_thread(self.request, self.thread_id) == self.expected_thread_data({
'response_count': 2,
'unread_comment_count': 1
})
assert httpretty.last_request().method == 'GET'
def test_thread_id_not_found(self):
self.register_get_thread_error_response("missing_thread", 404)
with self.assertRaises(ThreadNotFoundError):
with pytest.raises(ThreadNotFoundError):
get_thread(self.request, "missing_thread")
def test_nonauthor_enrolled_in_course(self):
@@ -3309,16 +3166,16 @@ class RetrieveThreadTest(
CourseEnrollmentFactory.create(user=non_author_user, course_id=self.course.id)
self.register_thread()
self.request.user = non_author_user
self.assertEqual(get_thread(self.request, self.thread_id), self.expected_thread_data({
"editable_fields": ["abuse_flagged", "following", "read", "voted"],
"unread_comment_count": 1,
}))
self.assertEqual(httpretty.last_request().method, "GET")
assert get_thread(self.request, self.thread_id) == self.expected_thread_data({
'editable_fields': ['abuse_flagged', 'following', 'read', 'voted'],
'unread_comment_count': 1
})
assert httpretty.last_request().method == 'GET'
def test_not_enrolled_in_course(self):
self.register_thread()
self.request.user = UserFactory.create()
with self.assertRaises(CourseNotFoundError):
with pytest.raises(CourseNotFoundError):
get_thread(self.request, self.thread_id)
@ddt.data(
@@ -3359,6 +3216,6 @@ class RetrieveThreadTest(
)
try:
get_thread(self.request, self.thread_id)
self.assertFalse(expected_error)
assert not expected_error
except ThreadNotFoundError:
self.assertTrue(expected_error)
assert expected_error

View File

@@ -59,37 +59,28 @@ class ThreadListGetFormTest(FormTestMixin, PaginationTestMixin, TestCase):
def test_basic(self):
form = self.get_form(expected_valid=True)
self.assertEqual(
form.cleaned_data,
{
"course_id": CourseLocator.from_string("Foo/Bar/Baz"),
"page": 2,
"page_size": 13,
"topic_id": set(),
"text_search": "",
"following": None,
"view": "",
"order_by": "last_activity_at",
"order_direction": "desc",
"requested_fields": set(),
}
)
assert form.cleaned_data == {
'course_id': CourseLocator.from_string('Foo/Bar/Baz'),
'page': 2,
'page_size': 13,
'topic_id': set(),
'text_search': '',
'following': None,
'view': '',
'order_by': 'last_activity_at',
'order_direction': 'desc',
'requested_fields': set()
}
def test_topic_id(self):
self.form_data.setlist("topic_id", ["example topic_id", "example 2nd topic_id"])
form = self.get_form(expected_valid=True)
self.assertEqual(
form.cleaned_data["topic_id"],
{"example topic_id", "example 2nd topic_id"},
)
assert form.cleaned_data['topic_id'] == {'example topic_id', 'example 2nd topic_id'}
def test_text_search(self):
self.form_data["text_search"] = "test search string"
form = self.get_form(expected_valid=True)
self.assertEqual(
form.cleaned_data["text_search"],
"test search string",
)
assert form.cleaned_data['text_search'] == 'test search string'
def test_missing_course_id(self):
self.form_data.pop("course_id")
@@ -159,10 +150,7 @@ class ThreadListGetFormTest(FormTestMixin, PaginationTestMixin, TestCase):
def test_requested_fields(self):
self.form_data["requested_fields"] = "profile_image"
form = self.get_form(expected_valid=True)
self.assertEqual(
form.cleaned_data["requested_fields"],
{"profile_image"},
)
assert form.cleaned_data['requested_fields'] == {'profile_image'}
@ddt.ddt
@@ -181,16 +169,13 @@ class CommentListGetFormTest(FormTestMixin, PaginationTestMixin, TestCase):
def test_basic(self):
form = self.get_form(expected_valid=True)
self.assertEqual(
form.cleaned_data,
{
"thread_id": "deadbeef",
"endorsed": False,
"page": 2,
"page_size": 13,
"requested_fields": set(),
}
)
assert form.cleaned_data == {
'thread_id': 'deadbeef',
'endorsed': False,
'page': 2,
'page_size': 13,
'requested_fields': set()
}
def test_missing_thread_id(self):
self.form_data.pop("thread_id")
@@ -217,7 +202,4 @@ class CommentListGetFormTest(FormTestMixin, PaginationTestMixin, TestCase):
def test_requested_fields(self):
self.form_data["requested_fields"] = {"profile_image"}
form = self.get_form(expected_valid=True)
self.assertEqual(
form.cleaned_data["requested_fields"],
{"profile_image"},
)
assert form.cleaned_data['requested_fields'] == {'profile_image'}

View File

@@ -22,7 +22,7 @@ class PaginationSerializerTest(TestCase):
request = RequestFactory().get("/test")
paginator = DiscussionAPIPagination(request, page_num, num_pages)
actual = paginator.get_paginated_response(objects)
self.assertEqual(actual.data, expected)
assert actual.data == expected
def test_empty(self):
self.do_case(

View File

@@ -48,7 +48,7 @@ class GetInitializableFieldsTest(ModuleStoreTestCase):
}
if is_privileged and is_cohorted:
expected |= {"group_id"}
self.assertEqual(actual, expected)
assert actual == expected
@ddt.data(*itertools.product([True, False], ["question", "discussion"], [True, False]))
@ddt.unpack
@@ -64,7 +64,7 @@ class GetInitializableFieldsTest(ModuleStoreTestCase):
}
if (is_thread_author and thread_type == "question") or is_privileged:
expected |= {"endorsed"}
self.assertEqual(actual, expected)
assert actual == expected
@ddt.ddt
@@ -85,7 +85,7 @@ class GetEditableFieldsTest(ModuleStoreTestCase):
expected |= {"topic_id", "type", "title", "raw_body"}
if is_privileged and is_cohorted:
expected |= {"group_id"}
self.assertEqual(actual, expected)
assert actual == expected
@ddt.data(*itertools.product([True, False], [True, False], ["question", "discussion"], [True, False]))
@ddt.unpack
@@ -102,7 +102,7 @@ class GetEditableFieldsTest(ModuleStoreTestCase):
expected |= {"raw_body"}
if (is_thread_author and thread_type == "question") or is_privileged:
expected |= {"endorsed"}
self.assertEqual(actual, expected)
assert actual == expected
@ddt.ddt
@@ -113,7 +113,7 @@ class CanDeleteTest(ModuleStoreTestCase):
def test_thread(self, is_author, is_privileged):
thread = Thread(user_id="5" if is_author else "6")
context = _get_context(requester_id="5", is_requester_privileged=is_privileged)
self.assertEqual(can_delete(thread, context), is_author or is_privileged)
assert can_delete(thread, context) == (is_author or is_privileged)
@ddt.data(*itertools.product([True, False], [True, False], [True, False]))
@ddt.unpack
@@ -124,4 +124,4 @@ class CanDeleteTest(ModuleStoreTestCase):
is_requester_privileged=is_privileged,
thread=Thread(user_id="5" if is_thread_author else "6")
)
self.assertEqual(can_delete(comment, context), is_author or is_privileged)
assert can_delete(comment, context) == (is_author or is_privileged)

View File

@@ -20,7 +20,7 @@ class RenderBodyTest(TestCase):
"""Tests for render_body"""
def test_empty(self):
self.assertEqual(render_body(""), "")
assert render_body('') == ''
@ddt.data(
("*", "em"),
@@ -29,10 +29,8 @@ class RenderBodyTest(TestCase):
)
@ddt.unpack
def test_markdown_inline(self, delimiter, tag):
self.assertEqual(
render_body(u"{delimiter}some text{delimiter}".format(delimiter=delimiter)),
u"<p><{tag}>some text</{tag}></p>".format(tag=tag)
)
assert render_body(u'{delimiter}some text{delimiter}'.format(delimiter=delimiter)) == \
u'<p><{tag}>some text</{tag}></p>'.format(tag=tag)
@ddt.data(
"b", "blockquote", "code", "del", "dd", "dl", "dt", "em", "h1", "h2", "h3", "i", "kbd",
@@ -42,52 +40,49 @@ class RenderBodyTest(TestCase):
raw_body = "<{tag}>some text</{tag}>".format(tag=tag)
is_inline_tag = tag in ["b", "code", "del", "em", "i", "kbd", "s", "sup", "sub", "strong", "strike"]
rendered_body = _add_p_tags(raw_body) if is_inline_tag else raw_body
self.assertEqual(render_body(raw_body), rendered_body)
assert render_body(raw_body) == rendered_body
@ddt.data("br", "hr")
def test_selfclosing_tag(self, tag):
raw_body = "<{tag}>".format(tag=tag)
is_inline_tag = tag == "br"
rendered_body = _add_p_tags(raw_body) if is_inline_tag else raw_body
self.assertEqual(render_body(raw_body), rendered_body)
assert render_body(raw_body) == rendered_body
@ddt.data("http", "https", "ftp")
def test_allowed_a_tag(self, protocol):
raw_body = '<a href="{protocol}://foo" title="bar">baz</a>'.format(protocol=protocol)
self.assertEqual(render_body(raw_body), _add_p_tags(raw_body))
assert render_body(raw_body) == _add_p_tags(raw_body)
def test_disallowed_a_tag(self):
raw_body = '<a href="gopher://foo">link content</a>'
self.assertEqual(render_body(raw_body), "<p>link content</p>")
assert render_body(raw_body) == '<p>link content</p>'
@ddt.data("http", "https")
def test_allowed_img_tag(self, protocol):
raw_body = '<img src="{protocol}://foo" width="111" height="222" alt="bar" title="baz">'.format(
protocol=protocol
)
self.assertEqual(render_body(raw_body), _add_p_tags(raw_body))
assert render_body(raw_body) == _add_p_tags(raw_body)
def test_disallowed_img_tag(self):
raw_body = '<img src="gopher://foo">'
self.assertEqual(render_body(raw_body), "<p></p>")
assert render_body(raw_body) == '<p></p>'
def test_script_tag(self):
raw_body = '<script type="text/javascript">alert("evil script");</script>'
self.assertEqual(render_body(raw_body), 'alert("evil script");')
assert render_body(raw_body) == 'alert("evil script");'
@ddt.data("p", "br", "li", "hr") # img is tested above
def test_allowed_unpaired_tags(self, tag):
raw_body = "foo<{tag}>bar".format(tag=tag)
self.assertEqual(render_body(raw_body), _add_p_tags(raw_body))
assert render_body(raw_body) == _add_p_tags(raw_body)
def test_unpaired_start_tag(self):
self.assertEqual(render_body("foo<i>bar"), "<p>foobar</p>")
assert render_body('foo<i>bar') == '<p>foobar</p>'
def test_unpaired_end_tag(self):
self.assertEqual(render_body("foo</i>bar"), "<p>foobar</p>")
assert render_body('foo</i>bar') == '<p>foobar</p>'
def test_interleaved_tags(self):
self.assertEqual(
render_body("foo<i>bar<b>baz</i>quux</b>greg"),
"<p>foo<i>barbaz</i>quuxgreg</p>"
)
assert render_body('foo<i>bar<b>baz</i>quux</b>greg') == '<p>foo<i>barbaz</i>quuxgreg</p>'

View File

@@ -100,7 +100,7 @@ class SerializerTestMixin(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetM
self.make_cs_content({"anonymous": anonymous, "anonymous_to_peers": anonymous_to_peers})
)
actual_serialized_anonymous = serialized["author"] is None
self.assertEqual(actual_serialized_anonymous, expected_serialized_anonymous)
assert actual_serialized_anonymous == expected_serialized_anonymous
@ddt.data(
(FORUM_ROLE_ADMINISTRATOR, False, "Staff"),
@@ -128,17 +128,17 @@ class SerializerTestMixin(ForumsEnableMixin, CommentsServiceMockMixin, UrlResetM
"""
self.create_role(role_name, [self.author])
serialized = self.serialize(self.make_cs_content({"anonymous": anonymous}))
self.assertEqual(serialized["author_label"], expected_label)
assert serialized['author_label'] == expected_label
def test_abuse_flagged(self):
serialized = self.serialize(self.make_cs_content({"abuse_flaggers": [str(self.user.id)]}))
self.assertEqual(serialized["abuse_flagged"], True)
assert serialized['abuse_flagged'] is True
def test_voted(self):
thread_id = "test_thread"
self.register_get_user_response(self.user, upvoted_ids=[thread_id])
serialized = self.serialize(self.make_cs_content({"id": thread_id}))
self.assertEqual(serialized["voted"], True)
assert serialized['voted'] is True
@ddt.ddt
@@ -188,7 +188,7 @@ class ThreadSerializerSerializationTest(SerializerTestMixin, SharedModuleStoreTe
"pinned": True,
"editable_fields": ["abuse_flagged", "following", "read", "voted"],
})
self.assertEqual(self.serialize(thread), expected)
assert self.serialize(thread) == expected
thread["thread_type"] = "question"
expected.update({
@@ -201,7 +201,7 @@ class ThreadSerializerSerializationTest(SerializerTestMixin, SharedModuleStoreTe
"http://testserver/api/discussion/v1/comments/?thread_id=test_thread&endorsed=False"
),
})
self.assertEqual(self.serialize(thread), expected)
assert self.serialize(thread) == expected
def test_pinned_missing(self):
"""
@@ -212,34 +212,34 @@ class ThreadSerializerSerializationTest(SerializerTestMixin, SharedModuleStoreTe
del thread_data["pinned"]
self.register_get_thread_response(thread_data)
serialized = self.serialize(thread_data)
self.assertEqual(serialized["pinned"], False)
assert serialized['pinned'] is False
def test_group(self):
self.course.cohort_config = {"cohorted": True}
modulestore().update_item(self.course, ModuleStoreEnum.UserID.test)
cohort = CohortFactory.create(course_id=self.course.id)
serialized = self.serialize(self.make_cs_content({"group_id": cohort.id}))
self.assertEqual(serialized["group_id"], cohort.id)
self.assertEqual(serialized["group_name"], cohort.name)
assert serialized['group_id'] == cohort.id
assert serialized['group_name'] == cohort.name
def test_following(self):
thread_id = "test_thread"
self.register_get_user_response(self.user, subscribed_thread_ids=[thread_id])
serialized = self.serialize(self.make_cs_content({"id": thread_id}))
self.assertEqual(serialized["following"], True)
assert serialized['following'] is True
def test_response_count(self):
thread_data = self.make_cs_content({"resp_total": 2})
self.register_get_thread_response(thread_data)
serialized = self.serialize(thread_data)
self.assertEqual(serialized["response_count"], 2)
assert serialized['response_count'] == 2
def test_response_count_missing(self):
thread_data = self.make_cs_content({})
del thread_data["resp_total"]
self.register_get_thread_response(thread_data)
serialized = self.serialize(thread_data)
self.assertNotIn("response_count", serialized)
assert 'response_count' not in serialized
@ddt.ddt
@@ -313,7 +313,7 @@ class CommentSerializerTest(SerializerTestMixin, SharedModuleStoreTestCase):
"editable_fields": ["abuse_flagged", "voted"],
"child_count": 0,
}
self.assertEqual(self.serialize(comment), expected)
assert self.serialize(comment) == expected
@ddt.data(
*itertools.product(
@@ -344,7 +344,7 @@ class CommentSerializerTest(SerializerTestMixin, SharedModuleStoreTestCase):
)
actual_endorser_anonymous = serialized["endorsed_by"] is None
expected_endorser_anonymous = endorser_role_name == FORUM_ROLE_STUDENT and thread_anonymous
self.assertEqual(actual_endorser_anonymous, expected_endorser_anonymous)
assert actual_endorser_anonymous == expected_endorser_anonymous
@ddt.data(
(FORUM_ROLE_ADMINISTRATOR, "Staff"),
@@ -366,11 +366,11 @@ class CommentSerializerTest(SerializerTestMixin, SharedModuleStoreTestCase):
"""
self.create_role(role_name, [self.endorser])
serialized = self.serialize(self.make_cs_content(with_endorsement=True))
self.assertEqual(serialized["endorsed_by_label"], expected_label)
assert serialized['endorsed_by_label'] == expected_label
def test_endorsed_at(self):
serialized = self.serialize(self.make_cs_content(with_endorsement=True))
self.assertEqual(serialized["endorsed_at"], self.endorsed_at)
assert serialized['endorsed_at'] == self.endorsed_at
def test_children(self):
comment = self.make_cs_content({
@@ -393,12 +393,12 @@ class CommentSerializerTest(SerializerTestMixin, SharedModuleStoreTestCase):
],
})
serialized = self.serialize(comment)
self.assertEqual(serialized["children"][0]["id"], "test_child_1")
self.assertEqual(serialized["children"][0]["parent_id"], "test_root")
self.assertEqual(serialized["children"][1]["id"], "test_child_2")
self.assertEqual(serialized["children"][1]["parent_id"], "test_root")
self.assertEqual(serialized["children"][1]["children"][0]["id"], "test_grandchild")
self.assertEqual(serialized["children"][1]["children"][0]["parent_id"], "test_child_2")
assert serialized['children'][0]['id'] == 'test_child_1'
assert serialized['children'][0]['parent_id'] == 'test_root'
assert serialized['children'][1]['id'] == 'test_child_2'
assert serialized['children'][1]['parent_id'] == 'test_root'
assert serialized['children'][1]['children'][0]['id'] == 'test_grandchild'
assert serialized['children'][1]['children'][0]['parent_id'] == 'test_child_2'
@ddt.ddt
@@ -458,69 +458,57 @@ class ThreadSerializerDeserializationTest(
partial=(instance is not None),
context=get_context(self.course, self.request)
)
self.assertTrue(serializer.is_valid())
assert serializer.is_valid()
serializer.save()
return serializer.data
def test_create_minimal(self):
self.register_post_thread_response({"id": "test_id", "username": self.user.username})
saved = self.save_and_reserialize(self.minimal_data)
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/test_topic/threads"
)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"commentable_id": ["test_topic"],
"thread_type": ["discussion"],
"title": ["Test Title"],
"body": ["Test body"],
"user_id": [str(self.user.id)],
}
)
self.assertEqual(saved["id"], "test_id")
assert urlparse(httpretty.last_request().path).path ==\
'/api/v1/test_topic/threads' # lint-amnesty, pylint: disable=no-member
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'commentable_id': ['test_topic'],
'thread_type': ['discussion'],
'title': ['Test Title'],
'body': ['Test body'],
'user_id': [str(self.user.id)]
}
assert saved['id'] == 'test_id'
def test_create_all_fields(self):
self.register_post_thread_response({"id": "test_id", "username": self.user.username})
data = self.minimal_data.copy()
data["group_id"] = 42
self.save_and_reserialize(data)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"commentable_id": ["test_topic"],
"thread_type": ["discussion"],
"title": ["Test Title"],
"body": ["Test body"],
"user_id": [str(self.user.id)],
"group_id": ["42"],
}
)
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'commentable_id': ['test_topic'],
'thread_type': ['discussion'],
'title': ['Test Title'],
'body': ['Test body'],
'user_id': [str(self.user.id)],
'group_id': ['42']
}
def test_create_missing_field(self):
for field in self.minimal_data:
data = self.minimal_data.copy()
data.pop(field)
serializer = ThreadSerializer(data=data)
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{field: ["This field is required."]}
)
assert not serializer.is_valid()
assert serializer.errors == {field: ['This field is required.']}
@ddt.data("", " ")
def test_create_empty_string(self, value):
data = self.minimal_data.copy()
data.update({field: value for field in ["topic_id", "title", "raw_body"]})
serializer = ThreadSerializer(data=data, context=get_context(self.course, self.request))
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{field: ["This field may not be blank."] for field in ["topic_id", "title", "raw_body"]}
)
assert not serializer.is_valid()
assert serializer.errors == {
field: ['This field may not be blank.'] for field in ['topic_id', 'title', 'raw_body']
}
def test_create_type(self):
self.register_post_thread_response({"id": "test_id", "username": self.user.username})
@@ -530,27 +518,24 @@ class ThreadSerializerDeserializationTest(
data["type"] = "invalid_type"
serializer = ThreadSerializer(data=data)
self.assertFalse(serializer.is_valid())
assert not serializer.is_valid()
def test_update_empty(self):
self.register_put_thread_response(self.existing_thread.attributes)
self.save_and_reserialize({}, self.existing_thread)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"commentable_id": ["original_topic"],
"thread_type": ["discussion"],
"title": ["Original Title"],
"body": ["Original body"],
"anonymous": ["False"],
"anonymous_to_peers": ["False"],
"closed": ["False"],
"pinned": ["False"],
"user_id": [str(self.user.id)],
"read": ["False"],
}
)
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'commentable_id': ['original_topic'],
'thread_type': ['discussion'],
'title': ['Original Title'],
'body': ['Original body'],
'anonymous': ['False'],
'anonymous_to_peers': ['False'],
'closed': ['False'],
'pinned': ['False'],
'user_id': [str(self.user.id)],
'read': ['False']
}
@ddt.data(True, False)
def test_update_all(self, read):
@@ -563,24 +548,21 @@ class ThreadSerializerDeserializationTest(
"read": read,
}
saved = self.save_and_reserialize(data, self.existing_thread)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"commentable_id": ["edited_topic"],
"thread_type": ["question"],
"title": ["Edited Title"],
"body": ["Edited body"],
"anonymous": ["False"],
"anonymous_to_peers": ["False"],
"closed": ["False"],
"pinned": ["False"],
"user_id": [str(self.user.id)],
"read": [str(read)],
}
)
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'commentable_id': ['edited_topic'],
'thread_type': ['question'],
'title': ['Edited Title'],
'body': ['Edited body'],
'anonymous': ['False'],
'anonymous_to_peers': ['False'],
'closed': ['False'],
'pinned': ['False'],
'user_id': [str(self.user.id)],
'read': [str(read)]
}
for key in data:
self.assertEqual(saved[key], data[key])
assert saved[key] == data[key]
@ddt.data("", " ")
def test_update_empty_string(self, value):
@@ -590,11 +572,10 @@ class ThreadSerializerDeserializationTest(
partial=True,
context=get_context(self.course, self.request)
)
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{field: ["This field may not be blank."] for field in ["topic_id", "title", "raw_body"]}
)
assert not serializer.is_valid()
assert serializer.errors == {
field: ['This field may not be blank.'] for field in ['topic_id', 'title', 'raw_body']
}
def test_update_course_id(self):
serializer = ThreadSerializer(
@@ -603,11 +584,8 @@ class ThreadSerializerDeserializationTest(
partial=True,
context=get_context(self.course, self.request)
)
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{"course_id": ["This field is not allowed in an update."]}
)
assert not serializer.is_valid()
assert serializer.errors == {'course_id': ['This field is not allowed in an update.']}
@ddt.ddt
@@ -657,7 +635,7 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
partial=(instance is not None),
context=context
)
self.assertTrue(serializer.is_valid())
assert serializer.is_valid()
serializer.save()
return serializer.data
@@ -677,17 +655,14 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
"/api/v1/comments/{}".format(parent_id) if parent_id else
"/api/v1/threads/test_thread/comments"
)
self.assertEqual(urlparse(httpretty.last_request().path).path, expected_url) # lint-amnesty, pylint: disable=no-member
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"body": ["Test body"],
"user_id": [str(self.user.id)],
}
)
self.assertEqual(saved["id"], "test_comment")
self.assertEqual(saved["parent_id"], parent_id)
assert urlparse(httpretty.last_request().path).path == expected_url # lint-amnesty, pylint: disable=no-member
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'body': ['Test body'],
'user_id': [str(self.user.id)]
}
assert saved['id'] == 'test_comment'
assert saved['parent_id'] == parent_id
def test_create_all_fields(self):
data = self.minimal_data.copy()
@@ -700,15 +675,12 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
parent_id="test_parent"
)
self.save_and_reserialize(data)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"body": ["Test body"],
"user_id": [str(self.user.id)],
"endorsed": ["True"],
}
)
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'body': ['Test body'],
'user_id': [str(self.user.id)],
'endorsed': ['True']
}
def test_create_parent_id_nonexistent(self):
self.register_get_comment_error_response("bad_parent", 404)
@@ -716,15 +688,10 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
data["parent_id"] = "bad_parent"
context = get_context(self.course, self.request, make_minimal_cs_thread())
serializer = CommentSerializer(data=data, context=context)
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{
"non_field_errors": [
"parent_id does not identify a comment in the thread identified by thread_id."
]
}
)
assert not serializer.is_valid()
assert serializer.errors == {
'non_field_errors': ['parent_id does not identify a comment in the thread identified by thread_id.']
}
def test_create_parent_id_wrong_thread(self):
self.register_get_comment_response({"thread_id": "different_thread", "id": "test_parent"})
@@ -732,15 +699,10 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
data["parent_id"] = "test_parent"
context = get_context(self.course, self.request, make_minimal_cs_thread())
serializer = CommentSerializer(data=data, context=context)
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{
"non_field_errors": [
"parent_id does not identify a comment in the thread identified by thread_id."
]
}
)
assert not serializer.is_valid()
assert serializer.errors == {
'non_field_errors': ['parent_id does not identify a comment in the thread identified by thread_id.']
}
@ddt.data(None, -1, 0, 2, 5)
def test_create_parent_id_too_deep(self, max_depth):
@@ -758,7 +720,7 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
else:
data["parent_id"] = None
serializer = CommentSerializer(data=data, context=context)
self.assertTrue(serializer.is_valid(), serializer.errors)
assert serializer.is_valid(), serializer.errors
if max_depth is not None:
if max_depth >= 0:
self.register_get_comment_response({
@@ -770,8 +732,8 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
else:
data["parent_id"] = None
serializer = CommentSerializer(data=data, context=context)
self.assertFalse(serializer.is_valid())
self.assertEqual(serializer.errors, {"non_field_errors": ["Comment level is too deep."]})
assert not serializer.is_valid()
assert serializer.errors == {'non_field_errors': ['Comment level is too deep.']}
def test_create_missing_field(self):
for field in self.minimal_data:
@@ -781,11 +743,8 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
data=data,
context=get_context(self.course, self.request, make_minimal_cs_thread())
)
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{field: ["This field is required."]}
)
assert not serializer.is_valid()
assert serializer.errors == {field: ['This field is required.']}
def test_create_endorsed(self):
# TODO: The comments service doesn't populate the endorsement field on
@@ -794,34 +753,28 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
data = self.minimal_data.copy()
data["endorsed"] = True
saved = self.save_and_reserialize(data)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [six.text_type(self.course.id)],
"body": ["Test body"],
"user_id": [str(self.user.id)],
"endorsed": ["True"],
}
)
self.assertTrue(saved["endorsed"])
self.assertIsNone(saved["endorsed_by"])
self.assertIsNone(saved["endorsed_by_label"])
self.assertIsNone(saved["endorsed_at"])
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [six.text_type(self.course.id)],
'body': ['Test body'],
'user_id': [str(self.user.id)],
'endorsed': ['True']
}
assert saved['endorsed']
assert saved['endorsed_by'] is None
assert saved['endorsed_by_label'] is None
assert saved['endorsed_at'] is None
def test_update_empty(self):
self.register_put_comment_response(self.existing_comment.attributes)
self.save_and_reserialize({}, instance=self.existing_comment)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"body": ["Original body"],
"course_id": [six.text_type(self.course.id)],
"user_id": [str(self.user.id)],
"anonymous": ["False"],
"anonymous_to_peers": ["False"],
"endorsed": ["False"],
}
)
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'body': ['Original body'],
'course_id': [six.text_type(self.course.id)],
'user_id': [str(self.user.id)],
'anonymous': ['False'],
'anonymous_to_peers': ['False'],
'endorsed': ['False']
}
def test_update_all(self):
cs_response_data = self.existing_comment.attributes.copy()
@@ -832,22 +785,19 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
self.register_put_comment_response(cs_response_data)
data = {"raw_body": "Edited body", "endorsed": True}
saved = self.save_and_reserialize(data, instance=self.existing_comment)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"body": ["Edited body"],
"course_id": [six.text_type(self.course.id)],
"user_id": [str(self.user.id)],
"anonymous": ["False"],
"anonymous_to_peers": ["False"],
"endorsed": ["True"],
"endorsement_user_id": [str(self.user.id)],
}
)
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'body': ['Edited body'],
'course_id': [six.text_type(self.course.id)],
'user_id': [str(self.user.id)],
'anonymous': ['False'],
'anonymous_to_peers': ['False'],
'endorsed': ['True'],
'endorsement_user_id': [str(self.user.id)]
}
for key in data:
self.assertEqual(saved[key], data[key])
self.assertEqual(saved["endorsed_by"], self.user.username)
self.assertEqual(saved["endorsed_at"], "2015-06-05T00:00:00Z")
assert saved[key] == data[key]
assert saved['endorsed_by'] == self.user.username
assert saved['endorsed_at'] == '2015-06-05T00:00:00Z'
@ddt.data("", " ")
def test_update_empty_raw_body(self, value):
@@ -857,11 +807,8 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
partial=True,
context=get_context(self.course, self.request)
)
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{"raw_body": ["This field may not be blank."]}
)
assert not serializer.is_valid()
assert serializer.errors == {'raw_body': ['This field may not be blank.']}
@ddt.data("thread_id", "parent_id")
def test_update_non_updatable(self, field):
@@ -871,8 +818,5 @@ class CommentSerializerDeserializationTest(ForumsEnableMixin, CommentsServiceMoc
partial=True,
context=get_context(self.course, self.request)
)
self.assertFalse(serializer.is_valid())
self.assertEqual(
serializer.errors,
{field: ["This field is not allowed in an update."]}
)
assert not serializer.is_valid()
assert serializer.errors == {field: ['This field is not allowed in an update.']}

View File

@@ -82,9 +82,9 @@ class DiscussionAPIViewTestMixin(ForumsEnableMixin, CommentsServiceMockMixin, Ur
"""
Assert that the response has the given status code and parsed content
"""
self.assertEqual(response.status_code, expected_status)
assert response.status_code == expected_status
parsed_content = json.loads(response.content.decode('utf-8'))
self.assertEqual(parsed_content, expected_content)
assert parsed_content == expected_content
def register_thread(self, overrides=None):
"""
@@ -190,10 +190,10 @@ class RetireViewTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
"""
Assert that the response has the given status code and content
"""
self.assertEqual(response.status_code, expected_status)
assert response.status_code == expected_status
if expected_content:
self.assertEqual(response.content.decode('utf-8'), expected_content)
assert response.content.decode('utf-8') == expected_content
def build_jwt_headers(self, user):
"""
@@ -258,10 +258,10 @@ class ReplaceUsernamesViewTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
"""
Assert that the response has the given status code and content
"""
self.assertEqual(response.status_code, expected_status)
assert response.status_code == expected_status
if expected_content:
self.assertEqual(text_type(response.content), expected_content)
assert text_type(response.content) == expected_content
def build_jwt_headers(self, user):
"""
@@ -288,7 +288,7 @@ class ReplaceUsernamesViewTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
"username_mappings": mapping_data
}
response = self.call_api(self.client_user, data)
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
def test_auth(self):
""" Verify the endpoint only works with the service worker """
@@ -301,16 +301,16 @@ class ReplaceUsernamesViewTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
# Test unauthenticated
response = self.client.post(self.url, data)
self.assertEqual(response.status_code, 401)
assert response.status_code == 401
# Test non-service worker
random_user = UserFactory()
response = self.call_api(random_user, data)
self.assertEqual(response.status_code, 403)
assert response.status_code == 403
# Test service worker
response = self.call_api(self.client_user, data)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
def test_basic(self):
""" Check successful replacement """
@@ -325,8 +325,8 @@ class ReplaceUsernamesViewTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
}
self.register_get_username_replacement_response(self.user)
response = self.call_api(self.client_user, data)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data, expected_response)
assert response.status_code == 200
assert response.data == expected_response
def test_not_authenticated(self):
"""
@@ -663,10 +663,9 @@ class ThreadViewSetListTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase, Pro
200,
expected_response
)
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/users/{}/subscribed_threads".format(self.user.id)
)
assert urlparse(
httpretty.last_request().path # lint-amnesty, pylint: disable=no-member
).path == '/api/v1/users/{}/subscribed_threads'.format(self.user.id)
@ddt.data(False, "false", "0")
def test_following_false(self, following):
@@ -798,13 +797,13 @@ class ThreadViewSetListTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase, Pro
self.url,
{"course_id": text_type(self.course.id), "requested_fields": "profile_image"},
)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_threads = json.loads(response.content.decode('utf-8'))['results']
for response_thread in response_threads:
expected_profile_data = self.get_expected_user_profile(response_thread['author'])
response_users = response_thread['users']
self.assertEqual(expected_profile_data, response_users[response_thread['author']])
assert expected_profile_data == response_users[response_thread['author']]
def test_profile_image_requested_field_anonymous_user(self):
"""
@@ -823,10 +822,10 @@ class ThreadViewSetListTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase, Pro
self.url,
{"course_id": text_type(self.course.id), "requested_fields": "profile_image"},
)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_thread = json.loads(response.content.decode('utf-8'))['results'][0]
self.assertIsNone(response_thread['author'])
self.assertEqual({}, response_thread['users'])
assert response_thread['author'] is None
assert {} == response_thread['users']
@httpretty.activate
@@ -858,20 +857,17 @@ class ThreadViewSetCreateTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
json.dumps(request_data),
content_type="application/json"
)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(response_data, self.expected_thread_data({"read": True}))
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [text_type(self.course.id)],
"commentable_id": ["test_topic"],
"thread_type": ["discussion"],
"title": ["Test Title"],
"body": ["Test body"],
"user_id": [str(self.user.id)],
}
)
assert response_data == self.expected_thread_data({'read': True})
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [text_type(self.course.id)],
'commentable_id': ['test_topic'],
'thread_type': ['discussion'],
'title': ['Test Title'],
'body': ['Test body'],
'user_id': [str(self.user.id)]
}
def test_error(self):
request_data = {
@@ -888,9 +884,9 @@ class ThreadViewSetCreateTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
expected_response_data = {
"field_errors": {"course_id": {"developer_message": "This field is required."}}
}
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(response_data, expected_response_data)
assert response_data == expected_response_data
@ddt.ddt
@@ -914,39 +910,31 @@ class ThreadViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTest
})
request_data = {"raw_body": "Edited body"}
response = self.request_patch(request_data)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(
response_data,
self.expected_thread_data({
"raw_body": "Edited body",
"rendered_body": "<p>Edited body</p>",
"editable_fields": [
"abuse_flagged", "following", "raw_body", "read", "title", "topic_id", "type", "voted"
],
"created_at": "Test Created Date",
"updated_at": "Test Updated Date",
"comment_count": 1,
"read": True,
"response_count": 2,
})
)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [text_type(self.course.id)],
"commentable_id": ["test_topic"],
"thread_type": ["discussion"],
"title": ["Test Title"],
"body": ["Edited body"],
"user_id": [str(self.user.id)],
"anonymous": ["False"],
"anonymous_to_peers": ["False"],
"closed": ["False"],
"pinned": ["False"],
"read": ["True"],
}
)
assert response_data == self.expected_thread_data({
'raw_body': 'Edited body',
'rendered_body': '<p>Edited body</p>',
'editable_fields': ['abuse_flagged', 'following', 'raw_body', 'read', 'title', 'topic_id', 'type', 'voted'],
'created_at': 'Test Created Date',
'updated_at': 'Test Updated Date',
'comment_count': 1,
'read': True,
'response_count': 2
})
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [text_type(self.course.id)],
'commentable_id': ['test_topic'],
'thread_type': ['discussion'],
'title': ['Test Title'],
'body': ['Edited body'],
'user_id': [str(self.user.id)],
'anonymous': ['False'],
'anonymous_to_peers': ['False'],
'closed': ['False'],
'pinned': ['False'],
'read': ['True']
}
def test_error(self):
self.register_get_user_response(self.user)
@@ -956,9 +944,9 @@ class ThreadViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTest
expected_response_data = {
"field_errors": {"title": {"developer_message": "This field may not be blank."}}
}
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(response_data, expected_response_data)
assert response_data == expected_response_data
@ddt.data(
("abuse_flagged", True),
@@ -971,19 +959,15 @@ class ThreadViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTest
self.register_flag_response("thread", "test_thread")
request_data = {field: value}
response = self.request_patch(request_data)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(
response_data,
self.expected_thread_data({
"read": True,
"closed": True,
"abuse_flagged": value,
"editable_fields": ["abuse_flagged", "read"],
"comment_count": 1,
"unread_comment_count": 0,
})
)
assert response_data == self.expected_thread_data({
'read': True,
'closed': True,
'abuse_flagged': value,
'editable_fields': ['abuse_flagged', 'read'],
'comment_count': 1, 'unread_comment_count': 0
})
@ddt.data(
("raw_body", "Edited body"),
@@ -997,7 +981,7 @@ class ThreadViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTest
self.register_flag_response("thread", "test_thread")
request_data = {field: value}
response = self.request_patch(request_data)
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
def test_patch_read_owner_user(self):
self.register_get_user_response(self.user)
@@ -1006,19 +990,14 @@ class ThreadViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTest
request_data = {"read": True}
response = self.request_patch(request_data)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(
response_data,
self.expected_thread_data({
"comment_count": 1,
"read": True,
"editable_fields": [
"abuse_flagged", "following", "raw_body", "read", "title", "topic_id", "type", "voted"
],
"response_count": 2,
})
)
assert response_data == self.expected_thread_data({
'comment_count': 1,
'read': True,
'editable_fields': ['abuse_flagged', 'following', 'raw_body', 'read', 'title', 'topic_id', 'type', 'voted'],
'response_count': 2
})
def test_patch_read_non_owner_user(self):
self.register_get_user_response(self.user)
@@ -1034,20 +1013,15 @@ class ThreadViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTest
request_data = {"read": True}
response = self.request_patch(request_data)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(
response_data,
self.expected_thread_data({
"author": str(thread_owner_user.username),
"comment_count": 1,
"read": True,
"editable_fields": [
"abuse_flagged", "following", "read", "voted"
],
"response_count": 2,
})
)
assert response_data == self.expected_thread_data({
'author': str(thread_owner_user.username),
'comment_count': 1,
'read': True,
'editable_fields': ['abuse_flagged', 'following', 'read', 'voted'],
'response_count': 2
})
@httpretty.activate
@@ -1071,18 +1045,15 @@ class ThreadViewSetDeleteTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
self.register_get_thread_response(cs_thread)
self.register_delete_thread_response(self.thread_id)
response = self.client.delete(self.url)
self.assertEqual(response.status_code, 204)
self.assertEqual(response.content, b"")
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/threads/{}".format(self.thread_id)
)
self.assertEqual(httpretty.last_request().method, "DELETE")
assert response.status_code == 204
assert response.content == b''
assert urlparse(httpretty.last_request().path).path == '/api/v1/threads/{}'.format(self.thread_id) # lint-amnesty, pylint: disable=no-member
assert httpretty.last_request().method == 'DELETE'
def test_delete_nonexistent_thread(self):
self.register_get_thread_error_response(self.thread_id, 404)
response = self.client.delete(self.url)
self.assertEqual(response.status_code, 404)
assert response.status_code == 404
@ddt.ddt
@@ -1277,7 +1248,7 @@ class CommentViewSetListTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase, Pr
"endorsed": endorsed,
})
parsed_content = json.loads(response.content.decode('utf-8'))
self.assertEqual(parsed_content["results"][0]["id"], comment_id)
assert parsed_content['results'][0]['id'] == comment_id
def test_question_invalid_endorsed(self):
response = self.client.get(self.url, {
@@ -1374,12 +1345,12 @@ class CommentViewSetListTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase, Pr
self.create_profile_image(self.user, get_profile_image_storage())
response = self.client.get(self.url, {"thread_id": self.thread_id, "requested_fields": "profile_image"})
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_comments = json.loads(response.content.decode('utf-8'))['results']
for response_comment in response_comments:
expected_profile_data = self.get_expected_user_profile(response_comment['author'])
response_users = response_comment['users']
self.assertEqual(expected_profile_data, response_users[response_comment['author']])
assert expected_profile_data == response_users[response_comment['author']]
def test_profile_image_requested_field_endorsed_comments(self):
"""
@@ -1417,14 +1388,14 @@ class CommentViewSetListTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase, Pr
"endorsed": True,
"requested_fields": "profile_image",
})
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_comments = json.loads(response.content.decode('utf-8'))['results']
for response_comment in response_comments:
expected_author_profile_data = self.get_expected_user_profile(response_comment['author'])
expected_endorser_profile_data = self.get_expected_user_profile(response_comment['endorsed_by'])
response_users = response_comment['users']
self.assertEqual(expected_author_profile_data, response_users[response_comment['author']])
self.assertEqual(expected_endorser_profile_data, response_users[response_comment['endorsed_by']])
assert expected_author_profile_data == response_users[response_comment['author']]
assert expected_endorser_profile_data == response_users[response_comment['endorsed_by']]
def test_profile_image_request_for_null_endorsed_by(self):
"""
@@ -1450,13 +1421,13 @@ class CommentViewSetListTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase, Pr
"endorsed": True,
"requested_fields": "profile_image",
})
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_comments = json.loads(response.content.decode('utf-8'))['results']
for response_comment in response_comments:
expected_author_profile_data = self.get_expected_user_profile(response_comment['author'])
response_users = response_comment['users']
self.assertEqual(expected_author_profile_data, response_users[response_comment['author']])
self.assertNotIn(response_comment['endorsed_by'], response_users)
assert expected_author_profile_data == response_users[response_comment['author']]
assert response_comment['endorsed_by'] not in response_users
@httpretty.activate
@@ -1487,18 +1458,15 @@ class CommentViewSetDeleteTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
self.register_get_comment_response(cs_comment)
self.register_delete_comment_response(self.comment_id)
response = self.client.delete(self.url)
self.assertEqual(response.status_code, 204)
self.assertEqual(response.content, b"")
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/comments/{}".format(self.comment_id)
)
self.assertEqual(httpretty.last_request().method, "DELETE")
assert response.status_code == 204
assert response.content == b''
assert urlparse(httpretty.last_request().path).path == '/api/v1/comments/{}'.format(self.comment_id) # lint-amnesty, pylint: disable=no-member
assert httpretty.last_request().method == 'DELETE'
def test_delete_nonexistent_comment(self):
self.register_get_comment_error_response(self.comment_id, 404)
response = self.client.delete(self.url)
self.assertEqual(response.status_code, 404)
assert response.status_code == 404
@httpretty.activate
@@ -1544,21 +1512,15 @@ class CommentViewSetCreateTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
json.dumps(request_data),
content_type="application/json"
)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(response_data, expected_response_data)
self.assertEqual(
urlparse(httpretty.last_request().path).path, # lint-amnesty, pylint: disable=no-member
"/api/v1/threads/test_thread/comments"
)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"course_id": [text_type(self.course.id)],
"body": ["Test body"],
"user_id": [str(self.user.id)],
}
)
assert response_data == expected_response_data
assert urlparse(httpretty.last_request().path).path == '/api/v1/threads/test_thread/comments' # lint-amnesty, pylint: disable=no-member
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'course_id': [text_type(self.course.id)],
'body': ['Test body'],
'user_id': [str(self.user.id)]
}
def test_error(self):
response = self.client.post(
@@ -1569,9 +1531,9 @@ class CommentViewSetCreateTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
expected_response_data = {
"field_errors": {"thread_id": {"developer_message": "This field is required."}}
}
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(response_data, expected_response_data)
assert response_data == expected_response_data
def test_closed_thread(self):
self.register_get_user_response(self.user)
@@ -1586,7 +1548,7 @@ class CommentViewSetCreateTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase):
json.dumps(request_data),
content_type="application/json"
)
self.assertEqual(response.status_code, 403)
assert response.status_code == 403
@ddt.ddt
@@ -1637,29 +1599,23 @@ class CommentViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTes
self.register_comment({"created_at": "Test Created Date", "updated_at": "Test Updated Date"})
request_data = {"raw_body": "Edited body"}
response = self.request_patch(request_data)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(
response_data,
self.expected_response_data({
"raw_body": "Edited body",
"rendered_body": "<p>Edited body</p>",
"editable_fields": ["abuse_flagged", "raw_body", "voted"],
"created_at": "Test Created Date",
"updated_at": "Test Updated Date",
})
)
self.assertEqual(
httpretty.last_request().parsed_body, # lint-amnesty, pylint: disable=no-member
{
"body": ["Edited body"],
"course_id": [text_type(self.course.id)],
"user_id": [str(self.user.id)],
"anonymous": ["False"],
"anonymous_to_peers": ["False"],
"endorsed": ["False"],
}
)
assert response_data == self.expected_response_data({
'raw_body': 'Edited body',
'rendered_body': '<p>Edited body</p>',
'editable_fields': ['abuse_flagged', 'raw_body', 'voted'],
'created_at': 'Test Created Date',
'updated_at': 'Test Updated Date'
})
assert httpretty.last_request().parsed_body == { # lint-amnesty, pylint: disable=no-member
'body': ['Edited body'],
'course_id': [text_type(self.course.id)],
'user_id': [str(self.user.id)],
'anonymous': ['False'],
'anonymous_to_peers': ['False'],
'endorsed': ['False']
}
def test_error(self):
self.register_thread()
@@ -1669,9 +1625,9 @@ class CommentViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTes
expected_response_data = {
"field_errors": {"raw_body": {"developer_message": "This field may not be blank."}}
}
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(response_data, expected_response_data)
assert response_data == expected_response_data
@ddt.data(
("abuse_flagged", True),
@@ -1684,15 +1640,12 @@ class CommentViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTes
self.register_flag_response("comment", "test_comment")
request_data = {field: value}
response = self.request_patch(request_data)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_data = json.loads(response.content.decode('utf-8'))
self.assertEqual(
response_data,
self.expected_response_data({
"abuse_flagged": value,
"editable_fields": ["abuse_flagged"],
})
)
assert response_data == self.expected_response_data({
'abuse_flagged': value,
'editable_fields': ['abuse_flagged']
})
@ddt.data(
("raw_body", "Edited body"),
@@ -1705,7 +1658,7 @@ class CommentViewSetPartialUpdateTest(DiscussionAPIViewTestMixin, ModuleStoreTes
self.register_comment()
request_data = {field: value}
response = self.request_patch(request_data)
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
@httpretty.activate
@@ -1730,17 +1683,14 @@ class ThreadViewSetRetrieveTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase,
})
self.register_get_thread_response(cs_thread)
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertEqual(
json.loads(response.content.decode('utf-8')),
self.expected_thread_data({"unread_comment_count": 1})
)
self.assertEqual(httpretty.last_request().method, "GET")
assert response.status_code == 200
assert json.loads(response.content.decode('utf-8')) == self.expected_thread_data({'unread_comment_count': 1})
assert httpretty.last_request().method == 'GET'
def test_retrieve_nonexistent_thread(self):
self.register_get_thread_error_response(self.thread_id, 404)
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
assert response.status_code == 404
def test_profile_image_requested_field(self):
"""
@@ -1756,10 +1706,10 @@ class ThreadViewSetRetrieveTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase,
self.register_get_thread_response(cs_thread)
self.create_profile_image(self.user, get_profile_image_storage())
response = self.client.get(self.url, {"requested_fields": "profile_image"})
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
expected_profile_data = self.get_expected_user_profile(self.user.username)
response_users = json.loads(response.content.decode('utf-8'))['users']
self.assertEqual(expected_profile_data, response_users[self.user.username])
assert expected_profile_data == response_users[self.user.username]
@httpretty.activate
@@ -1825,13 +1775,13 @@ class CommentViewSetRetrieveTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase
}
response = self.client.get(self.url)
self.assertEqual(response.status_code, 200)
self.assertEqual(json.loads(response.content.decode('utf-8'))['results'][0], expected_response_data)
assert response.status_code == 200
assert json.loads(response.content.decode('utf-8'))['results'][0] == expected_response_data
def test_retrieve_nonexistent_comment(self):
self.register_get_comment_error_response(self.comment_id, 404)
response = self.client.get(self.url)
self.assertEqual(response.status_code, 404)
assert response.status_code == 404
def test_pagination(self):
"""
@@ -1876,13 +1826,13 @@ class CommentViewSetRetrieveTest(DiscussionAPIViewTestMixin, ModuleStoreTestCase
self.create_profile_image(self.user, get_profile_image_storage())
response = self.client.get(self.url, {'requested_fields': 'profile_image'})
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
response_comments = json.loads(response.content.decode('utf-8'))['results']
for response_comment in response_comments:
expected_profile_data = self.get_expected_user_profile(response_comment['author'])
response_users = response_comment['users']
self.assertEqual(expected_profile_data, response_users[response_comment['author']])
assert expected_profile_data == response_users[response_comment['author']]
@ddt.ddt
@@ -1959,14 +1909,14 @@ class CourseDiscussionSettingsAPIViewTest(APITestCase, UrlResetMixin, ModuleStor
def _assert_current_settings(self, expected_response):
"""Validate the current discussion settings against the expected response."""
response = self.client.get(self.path)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
content = json.loads(response.content.decode('utf-8'))
self.assertEqual(content, expected_response)
assert content == expected_response
def _assert_patched_settings(self, data, expected_response):
"""Validate the patched settings against the expected response."""
response = self.patch_request(data)
self.assertEqual(response.status_code, 204)
assert response.status_code == 204
self._assert_current_settings(expected_response)
@ddt.data('get', 'patch')
@@ -1974,7 +1924,7 @@ class CourseDiscussionSettingsAPIViewTest(APITestCase, UrlResetMixin, ModuleStor
"""Test and verify that authentication is required for this endpoint."""
self.client.logout()
response = getattr(self.client, method)(self.path)
self.assertEqual(response.status_code, 401)
assert response.status_code == 401
@ddt.data(
{'is_staff': False, 'get_status': 403, 'put_status': 403},
@@ -1988,12 +1938,12 @@ class CourseDiscussionSettingsAPIViewTest(APITestCase, UrlResetMixin, ModuleStor
self.client.logout()
response = self.client.get(self.path, **headers)
self.assertEqual(response.status_code, get_status)
assert response.status_code == get_status
response = self.patch_request(
{'always_divide_inline_discussions': True}, headers
)
self.assertEqual(response.status_code, put_status)
assert response.status_code == put_status
def test_non_existent_course_id(self):
"""Test the response when this endpoint is passed a non-existent course id."""
@@ -2003,14 +1953,14 @@ class CourseDiscussionSettingsAPIViewTest(APITestCase, UrlResetMixin, ModuleStor
'course_id': 'a/b/c'
})
)
self.assertEqual(response.status_code, 404)
assert response.status_code == 404
def test_get_settings(self):
"""Test the current discussion settings against the expected response."""
divided_inline_discussions, divided_course_wide_discussions = self._create_divided_discussions()
self._login_as_staff()
response = self.client.get(self.path)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
expected_response = self._get_expected_response()
expected_response['divided_course_wide_discussions'] = [
topic_name_to_id(self.course, name) for name in divided_course_wide_discussions
@@ -2019,7 +1969,7 @@ class CourseDiscussionSettingsAPIViewTest(APITestCase, UrlResetMixin, ModuleStor
topic_name_to_id(self.course, name) for name in divided_inline_discussions
]
content = json.loads(response.content.decode('utf-8'))
self.assertEqual(content, expected_response)
assert content == expected_response
def test_available_schemes(self):
"""Test the available division schemes against the expected response."""
@@ -2045,10 +1995,10 @@ class CourseDiscussionSettingsAPIViewTest(APITestCase, UrlResetMixin, ModuleStor
"""Test the response status code on sending a PATCH request with an empty body or missing fields."""
self._login_as_staff()
response = self.patch_request("")
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
response = self.patch_request({})
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
@ddt.data(
{'abc': 123},
@@ -2061,7 +2011,7 @@ class CourseDiscussionSettingsAPIViewTest(APITestCase, UrlResetMixin, ModuleStor
"""Test the response status code on sending a PATCH request with parameters having incorrect types."""
self._login_as_staff()
response = self.patch_request(body)
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
def test_update_always_divide_inline_discussion_settings(self):
"""Test whether the 'always_divide_inline_discussions' setting is updated."""
@@ -2194,17 +2144,17 @@ class CourseDiscussionRolesAPIViewTest(APITestCase, UrlResetMixin, ModuleStoreTe
"""Test and verify that authentication is required for this endpoint."""
self.client.logout()
response = getattr(self.client, method)(self.path())
self.assertEqual(response.status_code, 401)
assert response.status_code == 401
def test_oauth(self):
"""Test that OAuth authentication works for this endpoint."""
oauth_headers = self._get_oauth_headers(self.user)
self.client.logout()
response = self.client.get(self.path(), **oauth_headers)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
body = {'user_id': 'staff', 'action': 'allow'}
response = self.client.post(self.path(), body, format='json', **oauth_headers)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
@ddt.data(
{'username': 'u1', 'is_staff': False, 'expected_status': 403},
@@ -2216,10 +2166,10 @@ class CourseDiscussionRolesAPIViewTest(APITestCase, UrlResetMixin, ModuleStoreTe
UserFactory(username=username, password='edx', is_staff=is_staff)
self.client.login(username=username, password='edx')
response = self.client.get(self.path())
self.assertEqual(response.status_code, expected_status)
assert response.status_code == expected_status
response = self.client.post(self.path(), {'user_id': username, 'action': 'allow'}, format='json')
self.assertEqual(response.status_code, expected_status)
assert response.status_code == expected_status
def test_non_existent_course_id(self):
"""Test the response when the endpoint URL contains a non-existent course id."""
@@ -2227,10 +2177,10 @@ class CourseDiscussionRolesAPIViewTest(APITestCase, UrlResetMixin, ModuleStoreTe
path = self.path(course_id='a/b/c')
response = self.client.get(path)
self.assertEqual(response.status_code, 404)
assert response.status_code == 404
response = self.client.post(path)
self.assertEqual(response.status_code, 404)
assert response.status_code == 404
def test_non_existent_course_role(self):
"""Test the response when the endpoint URL contains a non-existent role."""
@@ -2238,10 +2188,10 @@ class CourseDiscussionRolesAPIViewTest(APITestCase, UrlResetMixin, ModuleStoreTe
path = self.path(role='A')
response = self.client.get(path)
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
response = self.client.post(path)
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
@ddt.data(
{'role': 'Moderator', 'count': 0},
@@ -2259,22 +2209,22 @@ class CourseDiscussionRolesAPIViewTest(APITestCase, UrlResetMixin, ModuleStoreTe
self._login_as_staff()
response = self.client.get(self.path(role=role))
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
content = json.loads(response.content.decode('utf-8'))
self.assertEqual(content['course_id'], 'x/y/z')
self.assertEqual(len(content['results']), count)
assert content['course_id'] == 'x/y/z'
assert len(content['results']) == count
expected_fields = ('username', 'email', 'first_name', 'last_name', 'group_name')
for item in content['results']:
for expected_field in expected_fields:
self.assertIn(expected_field, item)
self.assertEqual(content['division_scheme'], 'cohort')
assert expected_field in item
assert content['division_scheme'] == 'cohort'
def test_post_missing_body(self):
"""Test the response with a POST request without a body."""
self._login_as_staff()
response = self.client.post(self.path())
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
@ddt.data(
{'a': 1},
@@ -2288,10 +2238,10 @@ class CourseDiscussionRolesAPIViewTest(APITestCase, UrlResetMixin, ModuleStoreTe
"""
self._login_as_staff()
response = self.client.post(self.path(), body)
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
response = self.client.post(self.path(), body, format='json')
self.assertEqual(response.status_code, 400)
assert response.status_code == 400
@ddt.data(
{'action': 'allow', 'user_in_role': False},
@@ -2309,7 +2259,7 @@ class CourseDiscussionRolesAPIViewTest(APITestCase, UrlResetMixin, ModuleStoreTe
self._add_users_to_role(users, role)
response = self.post(role, user.username, action)
self.assertEqual(response.status_code, 200)
assert response.status_code == 200
content = json.loads(response.content.decode('utf-8'))
assertion = self.assertTrue if action == 'allow' else self.assertFalse
assertion(any(user.username in x['username'] for x in content['results']))

View File

@@ -359,7 +359,7 @@ class CommentsServiceMockMixin(object):
"""
actual_params = dict(httpretty_request.querystring)
actual_params.pop("request_id") # request_id is random
self.assertEqual(actual_params, expected_params)
assert actual_params == expected_params
def assert_last_query_params(self, expected_params):
"""
@@ -519,12 +519,12 @@ class ProfileImageTestMixin(object):
"""
for size, name in get_profile_image_names(user.username).items():
if exist:
self.assertTrue(storage.exists(name))
assert storage.exists(name)
with closing(Image.open(storage.path(name))) as img:
self.assertEqual(img.size, (size, size))
self.assertEqual(img.format, 'JPEG')
assert img.size == (size, size)
assert img.format == 'JPEG'
else:
self.assertFalse(storage.exists(name))
assert not storage.exists(name)
def get_expected_user_profile(self, username):
"""