def test_open_dois(self, test_data):
(doi, fulltext_url, license) = test_data
my_pub = pub.lookup_product_by_doi(doi)
my_pub.recalculate()
logger.info(u"was looking for {}, got {}\n\n".format(fulltext_url, my_pub.fulltext_url))
logger.info(u"doi: http://doi.org/{}".format(doi))
logger.info(u"title: {}".format(my_pub.best_title))
logger.info(u"evidence: {}\n\n".format(my_pub.evidence))
if my_pub.error:
logger.info(my_pub.error)
assert_not_equals(my_pub.fulltext_url, None)
# @data(*closed_dois)
# def test_closed_dois(self, test_data):
# (doi, fulltext_url, license) = test_data
# my_pub = pub.lookup_product_by_doi(doi)
# my_pub.recalculate()
#
# logger.info(u"was looking for {}, got {}\n\n".format(fulltext_url, my_pub.fulltext_url))
# logger.info(u"doi: http://doi.org/{}".format(doi))
# logger.info(u"title: {}".format(my_pub.best_title))
# logger.info(u"evidence: {}\n\n".format(my_pub.evidence))
# if my_pub.error:
# logger.info(my_pub.error)
#
# assert_equals(my_pub.fulltext_url, None)
#
# have to scrape the publisher pages to find these
python类data()的实例源码
def test_context(self):
"""
Assert context values for logged in user
"""
with mute_signals(post_save):
profile = ProfileFactory.create()
self.client.force_login(profile.user)
# ProgramFaculty and ProgramCourse are asserted via ProgramPageSerializer below
FacultyFactory.create_batch(3, program_page=self.program_page)
courses = self.program_page.program.course_set.all()
ProgramCourseFactory.create_batch(
len(courses), program_page=self.program_page, course=Iterator(courses)
)
ga_tracking_id = FuzzyText().fuzz()
with self.settings(
GA_TRACKING_ID=ga_tracking_id,
ENVIRONMENT='environment',
VERSION='version',
):
response = self.client.get(self.program_page.url)
assert response.context['authenticated'] is True
assert response.context['username'] is None
assert response.context['title'] == self.program_page.title
assert response.context['is_public'] is True
assert response.context['has_zendesk_widget'] is True
assert response.context['is_staff'] is False
self.assertContains(response, 'Share this page')
js_settings = json.loads(response.context['js_settings_json'])
assert js_settings == {
'gaTrackingID': ga_tracking_id,
'environment': 'environment',
'sentry_dsn': None,
'host': 'testserver',
'release_version': 'version',
'user': serialize_maybe_user(profile.user),
'program': ProgramPageSerializer(self.program_page).data,
}
def test_semester_date(self):
"""
If present, semester data should be displayed
"""
semester_dates = SemesterDateFactory.create_batch(3, program_page=self.program_page)
response = self.client.get(self.program_page.url)
for semester_date in semester_dates:
self.assertContains(response, semester_date.semester_name)
def test_edx_data_fetch_and_set(self):
"""Test that a user's edX data is properly fetched and set onto object properties"""
edx_user_data = CachedEdxUserData(self.user)
assert isinstance(edx_user_data.enrollments, Enrollments)
assert isinstance(edx_user_data.certificates, Certificates)
assert isinstance(edx_user_data.current_grades, CurrentGrades)
self.assert_edx_data_has_given_ids(edx_user_data, self.p1_course_run_keys + self.p2_course_run_keys)
def test_edx_data_with_program(self):
"""Test that a user's edX data is filtered by program when specified"""
p1_course_run_program = self.p1_course_run.course.program
edx_user_data = CachedEdxUserData(self.user, program=p1_course_run_program)
self.assert_edx_data_has_given_ids(edx_user_data, self.p1_course_run_keys)
p2_course_run_program = self.p2_course_run_1.course.program
edx_user_data = CachedEdxUserData(self.user, program=p2_course_run_program)
self.assert_edx_data_has_given_ids(edx_user_data, self.p2_course_run_keys)
def test_update_cached_enrollment(self, mocked_index):
"""Test for update_cached_enrollment"""
course_id = list(self.enrollment_ids)[0]
enrollment = self.enrollments.get_enrollment_for_course(course_id)
self.assert_cache_in_db()
# normal update that creates also the entry
CachedEdxDataApi.update_cached_enrollment(self.user, enrollment, course_id, False)
self.assert_cache_in_db(enrollment_keys=[course_id])
cached_enr = CachedEnrollment.objects.get(user=self.user, course_run__edx_course_key=course_id)
assert cached_enr.data == enrollment.json
assert mocked_index.delay.called is False
# update of different data with indexing
enr_json = {
"course_details": {
"course_id": course_id,
},
"is_active": True,
"mode": "verified",
"user": self.user.username
}
enrollment_new = Enrollment(enr_json)
CachedEdxDataApi.update_cached_enrollment(self.user, enrollment_new, course_id, True)
self.assert_cache_in_db(enrollment_keys=[course_id])
cached_enr.refresh_from_db()
assert cached_enr.data == enr_json
mocked_index.delay.assert_any_call([self.user.id], check_if_changed=True)
def test_when_edx_client_is_none(self, mock_cache_refresh):
"""Test that the edx data is not refreshed"""
api.get_user_program_info(self.user, None)
assert mock_cache_refresh.call_count == 0
def test_get_dashboard(self):
"""Test for GET"""
with patch('backends.utils.refresh_user_token', autospec=True):
result = self.client.get(self.url)
assert 'programs' in result.data
assert 'is_edx_data_fresh' in result.data
assert result.data['is_edx_data_fresh'] is True
assert len(result.data['programs']) == 2
assert [self.program_1.id, self.program_2.id] == [res_item['id'] for res_item in result.data['programs']]
def test_refresh_token_fails(self, refr_token):
"""
Test if the refresh_user_token raises any other kind of exception
"""
refr_token.side_effect = ZeroDivisionError
result = self.client.get(self.url)
assert result.status_code == status.HTTP_200_OK
assert 'programs' in result.data
assert 'is_edx_data_fresh' in result.data
def update_social_extra_data(self, data):
"""Helper function to update the python social auth extra data"""
self.social_auth.extra_data.update(data)
self.social_auth.save()
def test_course_id_mandatory(self):
"""
The request data must contain course_id
"""
resp = self.client.post(self.url, {}, format='json')
assert resp.status_code == status.HTTP_400_BAD_REQUEST
def test_enrollment_fails(self, mock_refresh, mock_edx_enr): # pylint: disable=unused-argument
"""
Test error when backend raises an exception
"""
error = HTTPError()
error.response = MagicMock()
error.response.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
mock_edx_enr.side_effect = error
resp = self.client.post(self.url, {'course_id': self.course_id}, format='json')
assert resp.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
# the response has a structure like {"error": "<message>"}
assert isinstance(resp.data, dict)
assert 'error' in resp.data
assert mock_edx_enr.call_count == 1
# assert just the second argument, since the first is `self`
assert mock_edx_enr.call_args[0][1] == self.course_id
# if instead edX returns a 400 error, an exception is raised by
# the view and the user gets a different error message
error.response.status_code = status.HTTP_400_BAD_REQUEST
mock_edx_enr.side_effect = error
resp = self.client.post(self.url, {'course_id': self.course_id}, format='json')
assert resp.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert isinstance(resp.data, list)
assert len(resp.data) == 1
assert PossiblyImproperlyConfigured.__name__ in resp.data[0]
# if the error from the call to edX is is not HTTPError, the user gets a normal json error
mock_edx_enr.side_effect = ValueError()
resp = self.client.post(self.url, {'course_id': self.course_id}, format='json')
assert resp.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
# the response has a structure like {"error": "<message>"}
assert isinstance(resp.data, dict)
assert 'error' in resp.data
def setUp(self):
super().setUp()
self.request_url = reverse("financial_aid_request")
self.client.force_login(self.profile.user)
self.data = {
"original_currency": "USD",
"program_id": self.program.id,
"original_income": 50000
}
def test_income_validation(self, original_income, original_currency, income_threshold, auto_approved):
"""
Tests FinancialAidRequestView post endpoint
"""
CountryIncomeThreshold.objects.filter(country_code=self.profile.country).update(
income_threshold=income_threshold
)
data = {
"original_income": original_income,
"original_currency": original_currency,
"program_id": self.program.id,
}
assert FinancialAid.objects.exclude(status=FinancialAidStatus.RESET).count() == 0
assert FinancialAidAudit.objects.count() == 0
self.make_http_request(self.client.post, self.request_url, status.HTTP_201_CREATED, data=data)
assert FinancialAid.objects.exclude(status=FinancialAidStatus.RESET).count() == 1
assert FinancialAidAudit.objects.count() == 1
financial_aid = FinancialAid.objects.exclude(status=FinancialAidStatus.RESET).first()
income_usd = determine_income_usd(original_income, original_currency)
assert financial_aid.tier_program == determine_tier_program(self.program, income_usd)
if not auto_approved:
assert financial_aid.status == FinancialAidStatus.PENDING_DOCS
else:
assert financial_aid.status == FinancialAidStatus.AUTO_APPROVED
self.assertAlmostEqual(financial_aid.income_usd, income_usd)
assert financial_aid.user == self.profile.user
self.assertAlmostEqual(financial_aid.original_income, original_income)
assert financial_aid.original_currency == original_currency
assert financial_aid.country_of_income == self.profile.country
assert financial_aid.country_of_residence == self.profile.country
assert is_near_now(financial_aid.date_exchange_rate)
def test_income_validation_missing_args(self):
"""
Tests FinancialAidRequestView post with missing args
"""
for missing_key in self.data:
data = {key: value for key, value in self.data.items() if key != missing_key}
self.make_http_request(self.client.post, self.request_url, status.HTTP_400_BAD_REQUEST, data=data)
def test_income_validation_no_financial_aid_availability(self):
"""
Tests FinancialAidRequestView post when financial aid not available for program
"""
self.program.financial_aid_availability = False
self.program.save()
self.make_http_request(self.client.post, self.request_url, status.HTTP_400_BAD_REQUEST, data=self.data)
def test_income_validation_user_not_enrolled(self):
"""
Tests FinancialAidRequestView post when User not enrolled in program
"""
ProgramEnrollment.objects.all().delete()
self.make_http_request(self.client.post, self.request_url, status.HTTP_400_BAD_REQUEST, data=self.data)
