我们从Python开源项目中,提取了以下48个代码示例,用于说明如何使用django.db.models.signals.post_save()。
def test_attach_avatars(self): """ It should attach robotic avatars given a username prefix """ with mute_signals(post_save): for i in range(3): ProfileFactory.create(user__username="user_{}".format(i)) for i in range(3): ProfileFactory.create(user__username="fake_{}".format(i)) # clear all images for profile in Profile.objects.all(): profile.image = None profile.image_medium = None profile.image_small = None profile.save() self.command.handle("attach_avatars", username_prefix="fake") for profile in Profile.objects.all(): has_image = profile.user.username.startswith("fake") assert bool(profile.image) == has_image assert bool(profile.image_medium) == has_image assert bool(profile.image_small) == has_image
def deserialize_user_data(user_data, programs): """ Deserializes a dict of mixed User/Profile data and returns the newly-inserted User """ username = FAKE_USER_USERNAME_PREFIX + user_data['email'].split('@')[0] user = deserialize_model_data(User, user_data, username=username) # Create social username user.social_auth.create( provider=EdxOrgOAuth2.name, uid=user.username, ) # The user data is generated in this script with mute_signals(post_save) # so we need to create the profile explicitly. profile = deserialize_model_data(Profile, user_data, user=user, fake_user=True) deserialize_profile_detail_data(profile, Employment, user_data['work_history']) deserialize_profile_detail_data(profile, Education, user_data['education']) deserialize_dashboard_data(user, user_data, programs) ensure_cached_data_freshness(user) return user
def test_context_has_role(self, role): """ Assert context values when staff or instructor which are different than for regular logged in user """ with mute_signals(post_save): profile = ProfileFactory.create() Role.objects.create( role=role, program=self.program_page.program, user=profile.user, ) self.client.force_login(profile.user) response = self.client.get(self.program_page.url) assert response.context['authenticated'] is True assert response.context['username'] is None assert response.context['is_staff'] is True
def test_get_token_for_user(settings, mocker): """ Assert that get_token_for_user returns a token for a valid DiscussionUser """ with mute_signals(post_save): user = UserFactory.create() settings.OPEN_DISCUSSIONS_JWT_SECRET = 'secret' settings.OPEN_DISCUSSIONS_JWT_EXPIRES_DELTA = 3600 mock_get_token = mocker.patch('open_discussions_api.utils.get_token') DiscussionUser.objects.create(user=user, username='username') assert get_token_for_user(user) is not None mock_get_token.assert_called_once_with( 'secret', 'username', [], expires_delta=3600, extra_payload={ 'auth_url': None, 'session_url': None } )
def test_create_discussion_user(mock_staff_client): """Verify create_discussion_user makes the correct API calls""" mock_response = mock_staff_client.users.create.return_value mock_response.status_code = 201 mock_response.json.return_value = { 'username': 'username' } with mute_signals(post_save): profile = ProfileFactory.create() discussion_user = DiscussionUser.objects.create(user=profile.user) api.create_discussion_user(discussion_user) assert discussion_user.username == 'username' mock_staff_client.users.create.assert_called_once_with( name=profile.full_name, image=profile.image.url, image_small=profile.image_small.url, image_medium=profile.image_medium.url, )
def test_update_discussion_user(mock_staff_client): """Verify update_discussion_user makes the correct API calls""" mock_response = mock_staff_client.users.update.return_value mock_response.status_code = 200 mock_response.json.return_value = { 'username': 'username' } with mute_signals(post_save): profile = ProfileFactory.create() discussion_user = DiscussionUser.objects.create(user=profile.user, username='username') api.update_discussion_user(discussion_user) mock_staff_client.users.update.assert_called_once_with( discussion_user.username, name=profile.full_name, image=profile.image.url, image_small=profile.image_small.url, image_medium=profile.image_medium.url, )
def setUpTestData(cls): with mute_signals(post_save): cls.learner1 = UserFactory.create() cls.learner1_username = 'learner1_username' cls.learner1.social_auth.create( provider=EdxOrgOAuth2.name, uid=cls.learner1_username ) cls.learner2 = UserFactory.create() cls.learner2_username = 'learner2_username' cls.learner2.social_auth.create( provider=EdxOrgOAuth2.name, uid=cls.learner2_username ) cls.program = ProgramFactory.create() cls.staff = UserFactory.create() cls.staff_username = 'staff_username' for learner in (cls.learner1, cls.learner2): ProgramEnrollment.objects.create( program=cls.program, user=learner, )
def setUpTestData(cls): with mute_signals(post_save): cls.profile = ProfileFactory.create() cls.program, _ = create_program(past=True) cls.course_run = cls.program.course_set.first().courserun_set.first() CachedCurrentGradeFactory.create( user=cls.profile.user, course_run=cls.course_run, data={ "passed": True, "percent": 0.9, "course_key": cls.course_run.edx_course_key, "username": cls.profile.user.username } ) CachedCertificateFactory.create(user=cls.profile.user, course_run=cls.course_run) cls.exam_run = ExamRunFactory.create( course=cls.course_run.course, date_first_schedulable=now_in_utc() - timedelta(days=1), )
def test_write_ead_file(self): """ Tests that write_ead_file outputs correctly """ kwargs = { 'id': 143, 'operation': 'add', 'exam_run__exam_series_code': 'MM-DEDP', 'exam_run__date_first_eligible': date(2016, 5, 15), 'exam_run__date_last_eligible': date(2016, 10, 15), } with mute_signals(post_save): profile = ProfileFactory(id=14879) exam_auths = [ExamAuthorizationFactory.create(user=profile.user, **kwargs)] exam_auths[0].updated_on = FIXED_DATETIME self.ead_writer.write(self.tsv_file, exam_auths) assert self.tsv_rows[0] == ( "add\t143\t" "14879\tMM-DEDP\t\t" "\t2016/05/15\t2016/10/15\t" # accommodation blank intentionally "2016/05/15 15:02:55" )
def setUpTestData(cls): with mute_signals(post_save): profile = ProfileFactory.create() cls.program, _ = create_program(past=True) cls.user = profile.user cls.course_run = cls.program.course_set.first().courserun_set.first() with mute_signals(post_save): CachedEnrollmentFactory.create(user=cls.user, course_run=cls.course_run) cls.exam_run = ExamRunFactory.create(course=cls.course_run.course) with mute_signals(post_save): cls.final_grade = FinalGradeFactory.create( user=cls.user, course_run=cls.course_run, passed=True, course_run_paid_on_edx=False, )
def test_successful_program_certificate_generation(self): """ Test has final grade and a certificate """ final_grade = FinalGradeFactory.create( user=self.user, course_run=self.run_1, passed=True, status='complete', grade=0.8 ) CourseRunGradingStatus.objects.create(course_run=self.run_1, status='complete') with mute_signals(post_save): MicromastersCourseCertificate.objects.create(final_grade=final_grade) cert_qset = MicromastersProgramCertificate.objects.filter(user=self.user, program=self.program) assert cert_qset.exists() is False api.generate_program_certificate(self.user, self.program) assert cert_qset.exists() is True
def test_upload_to(self): """ Image upload_to should have a function which creates a path """ # pin the timestamps used in creating the URL with patch('profiles.util.now_in_utc', autospec=True) as mocked_now_in_utc: mocked_now_in_utc.return_value = now_in_utc() with mute_signals(post_save): profile = ProfileFactory.create() assert profile.image.url.endswith(profile_image_upload_uri(None, "example.jpg").replace("+", "")) assert profile.image_small.url.endswith( profile_image_upload_uri_small(None, "example.jpg").replace("+", "") ) assert profile.image_medium.url.endswith( profile_image_upload_uri_medium(None, "example.jpg").replace("+", "") )
def test_update_education(self): """ Test that we handle updating an Education correctly """ with mute_signals(post_save): education = EducationFactory.create() education_data = EducationSerializer(education).data education_data['degree_name'] = BACHELORS serializer = ProfileSerializer(instance=education.profile, data={ 'education': [education_data], 'work_history': [] }) serializer.is_valid(raise_exception=True) serializer.save() assert education.profile.education.count() == 1 education = education.profile.education.first() assert EducationSerializer(education).data == education_data
def test_delete_education(self): """ Test that we delete Educations which aren't specified in the PATCH """ with mute_signals(post_save): education1 = EducationFactory.create() EducationFactory.create(profile=education1.profile) # has a different profile education3 = EducationFactory.create() assert education1.profile.education.count() == 2 education_object1 = EducationSerializer(education1).data serializer = ProfileSerializer(instance=education1.profile, data={ 'education': [education_object1], 'work_history': [] }) serializer.is_valid(raise_exception=True) serializer.save() assert education1.profile.education.count() == 1 assert education1.profile.education.first() == education1 # Other profile is unaffected assert education3.profile.education.count() == 1
def test_update_employment(self): """ Test that we handle updating an employment correctly """ with mute_signals(post_save): employment = EmploymentFactory.create() employment_object = EmploymentSerializer(employment).data employment_object['position'] = "SE" serializer = ProfileSerializer(instance=employment.profile, data={ 'work_history': [employment_object], 'education': [] }) serializer.is_valid(raise_exception=True) serializer.save() assert employment.profile.work_history.count() == 1 employment = employment.profile.work_history.first() assert EmploymentSerializer(employment).data == employment_object
def test_update_employment_different_profile(self): """ Make sure we can't edit an employment for a different profile """ with mute_signals(post_save): employment1 = EmploymentFactory.create() employment2 = EmploymentFactory.create() employment_object = EmploymentSerializer(employment1).data employment_object['id'] = employment2.id serializer = ProfileSerializer(instance=employment1.profile, data={ 'work_history': [employment_object], 'education': [] }) serializer.is_valid(raise_exception=True) with self.assertRaises(ValidationError) as ex: serializer.save() assert ex.exception.detail == ["Work history {} does not exist".format(employment2.id)]
def test_delete_employment(self): """ Test that we delete employments which aren't specified in the PATCH """ with mute_signals(post_save): employment1 = EmploymentFactory.create() EmploymentFactory.create(profile=employment1.profile) # has a different profile employment3 = EmploymentFactory.create() assert employment1.profile.work_history.count() == 2 employment_object1 = EmploymentSerializer(employment1).data serializer = ProfileSerializer(instance=employment1.profile, data={ 'work_history': [employment_object1], 'education': [] }) serializer.is_valid(raise_exception=True) serializer.save() assert employment1.profile.work_history.count() == 1 assert employment1.profile.work_history.first() == employment1 # Other profile is unaffected assert employment3.profile.work_history.count() == 1
def test_vermm_user_get_public_to_mm_profile(self): """ A verified mm user gets user's public_to_mm profile. """ with mute_signals(post_save): profile = ProfileFactory.create(user=self.user2, account_privacy=Profile.PUBLIC_TO_MM) ProfileFactory.create(user=self.user1, verified_micromaster_user=True) program = ProgramFactory.create() ProgramEnrollment.objects.create( program=program, user=self.user2, ) ProgramEnrollment.objects.create( program=program, user=self.user1, ) profile_data = ProfileLimitedSerializer(profile).data self.client.force_login(self.user1) resp = self.client.get(self.url2) assert resp.json() == format_image_expectation(profile_data)
def test_instructor_sees_entire_profile(self): """ An instructor should be able to see the entire profile despite the account privacy """ with mute_signals(post_save): profile = ProfileFactory.create(user=self.user2, account_privacy=Profile.PRIVATE) ProfileFactory.create(user=self.user1, verified_micromaster_user=False) program = ProgramFactory.create() ProgramEnrollment.objects.create( program=program, user=profile.user, ) Role.objects.create( program=program, role=Instructor.ROLE_ID, user=self.user1, ) self.client.force_login(self.user1) profile_data = ProfileSerializer(profile).data resp = self.client.get(self.url2) assert resp.json() == format_image_expectation(profile_data)
def test_staff_sees_entire_profile(self): """ Staff should be able to see the entire profile despite the account privacy """ with mute_signals(post_save): profile = ProfileFactory.create(user=self.user2, account_privacy=Profile.PRIVATE) ProfileFactory.create(user=self.user1, verified_micromaster_user=False) program = ProgramFactory.create() ProgramEnrollment.objects.create( program=program, user=profile.user, ) Role.objects.create( program=program, role=Staff.ROLE_ID, user=self.user1, ) self.client.force_login(self.user1) resp = self.client.get(self.url2) profile_data = ProfileSerializer(profile).data assert resp.json() == format_image_expectation(profile_data)
def test_serializer(self): """ Get a user's own profile, ensure that we used ProfileSerializer and not ProfileFilledOutSerializer """ with mute_signals(post_save): profile = ProfileFactory.create(user=self.user1, filled_out=False) self.client.force_login(self.user1) patch_data = ProfileSerializer(profile).data # PATCH may not succeed, we just care that the right serializer was used with patch( 'profiles.views.ProfileFilledOutSerializer.__new__', autospec=True, side_effect=ProfileFilledOutSerializer.__new__ ) as mocked_filled_out, patch( 'profiles.views.ProfileSerializer.__new__', autospec=True, side_effect=ProfileSerializer.__new__ ) as mocked: self.client.patch(self.url1, content_type="application/json", data=json.dumps(patch_data)) assert mocked.called assert not mocked_filled_out.called
def test_filled_out_serializer(self): """ Get a user's own profile, ensure that we used ProfileFilledOutSerializer """ with mute_signals(post_save): profile = ProfileFactory.create(user=self.user1, filled_out=True) self.client.force_login(self.user1) patch_data = ProfileSerializer(profile).data # PATCH may not succeed, we just care that the right serializer was used with patch( 'profiles.views.ProfileFilledOutSerializer.__new__', autospec=True, side_effect=ProfileFilledOutSerializer.__new__ ) as mocked: self.client.patch(self.url1, content_type="application/json", data=json.dumps(patch_data)) assert mocked.called
def test_upload_image(self): """ An image upload should not delete education or work history entries """ with mute_signals(post_save): profile = ProfileFactory.create(user=self.user1) EducationFactory.create(profile=profile) EmploymentFactory.create(profile=profile) self.client.force_login(self.user1) # create a dummy image file in memory for upload image_file = BytesIO() image = Image.new('RGBA', size=(50, 50), color=(256, 0, 0)) image.save(image_file, 'png') image_file.seek(0) # format patch using multipart upload resp = self.client.patch(self.url1, data={ 'image': image_file }, format='multipart') assert resp.status_code == 200, resp.content.decode('utf-8') assert profile.education.count() == 1 assert profile.work_history.count() == 1
def test_presave_removes_current_role(self): """ Updating the role in the model triggers a pre_save signal that removes the previous role from the user. """ mm_role = Role.objects.create( program=self.program, user=self.user, role='staff', ) self.assert_standard_role_permissions(True) # muting the post_save signal to avoid the reassignment of the roles and related permissions # in this way only the pre_save will run and the effect will be only to remove the old role with mute_signals(post_save): mm_role.role = 'instructor' mm_role.save() self.assert_standard_role_permissions(False)
def setUpTestData(cls): with mute_signals(post_save): profile = ProfileFactory.create() cls.user = profile.user cls.program = ProgramFactory.create() Role.objects.create( user=cls.user, program=cls.program, role=Staff.ROLE_ID ) with mute_signals(post_save): profile = ProfileFactory.create(email_optin=True, filled_out=True) profile2 = ProfileFactory.create(email_optin=False, filled_out=True) cls.learner = profile.user cls.learner2 = profile2.user # self.user with role staff on program for user in [cls.learner, cls.learner2, cls.user]: ProgramEnrollmentFactory( user=user, program=cls.program, )
def test_update_percolate_memberships(self, source_type, is_member, query_matches, mock_on_commit): """ Tests that existing memberships are updated where appropriate """ with mute_signals(post_save): query = PercolateQueryFactory.create(source_type=source_type) profile = ProfileFactory.create(filled_out=True) program_enrollment = ProgramEnrollmentFactory.create(user=profile.user) membership = PercolateQueryMembershipFactory.create( user=profile.user, query=query, is_member=is_member, needs_update=False ) with patch( 'search.api._search_percolate_queries', return_value=[query.id] if query_matches else [] ) as search_percolate_queries_mock: update_percolate_memberships(profile.user, source_type) search_percolate_queries_mock.assert_called_once_with(program_enrollment) membership.refresh_from_db() assert membership.needs_update is (is_member is not query_matches)
def test_index_program_enrolled_users_missing_profiles(self, mock_on_commit): """ Test that index_program_enrolled_users doesn't index users missing profiles """ with mute_signals(post_save): program_enrollments = [ProgramEnrollmentFactory.build() for _ in range(10)] with patch( 'search.indexing_api._index_chunk', autospec=True, return_value=0 ) as index_chunk, patch( 'search.indexing_api.serialize_program_enrolled_user', autospec=True, side_effect=lambda x: None # simulate a missing profile ) as serialize_mock, patch( 'search.indexing_api.serialize_public_enrolled_user', autospec=True, side_effect=lambda x: x ) as serialize_public_mock: index_program_enrolled_users(program_enrollments) assert index_chunk.call_count == 0 assert serialize_public_mock.call_count == 0 assert serialize_mock.call_count == len(program_enrollments)
def setUpTestData(cls): super().setUpTestData() cls.program_enrollment_unsent = ProgramEnrollmentFactory.create() cls.program_enrollment_sent = ProgramEnrollmentFactory.create() cls.automatic_email = AutomaticEmailFactory.create(enabled=True) cls.percolate_query = cls.automatic_email.query cls.other_query = PercolateQueryFactory.create(source_type=PercolateQuery.DISCUSSION_CHANNEL_TYPE) cls.percolate_queries = [cls.percolate_query, cls.other_query] cls.automatic_email_disabled = AutomaticEmailFactory.create(enabled=False) cls.percolate_query_disabled = cls.automatic_email_disabled.query SentAutomaticEmail.objects.create( automatic_email=cls.automatic_email, user=cls.program_enrollment_sent.user, status=SentAutomaticEmail.SENT, ) # User was sent email connected to a different AutomaticEmail SentAutomaticEmail.objects.create( user=cls.program_enrollment_unsent.user, automatic_email=AutomaticEmailFactory.create(enabled=True), status=SentAutomaticEmail.SENT, ) with mute_signals(post_save): cls.staff_user = UserFactory.create()
def post_save_handler(sender, instance, **kwargs): """ Called when any of models saved or updated. """ logger.info( 'Singal sync on post_save from django to google fusiontables' ' for model: %s and row id: %s', sender._meta.db_table, instance.id ) try: kft = _get_kft_instance(sender=sender, raw=kwargs.get('raw'))[1] if not instance._ft_id: kft.insert_rows(sender=sender, row_id=instance.id) else: kft.update_rows(sender=sender, row_id=instance._ft_id) except SkipException as exc: logger.debug("Skip synchronization: %s", exc.args) # else: # raise CannotCreateInstanceException( # "Internal error: Cannot create kfusiontables instance." # )
def ready(self): from django.db.models.signals import post_save, post_delete from devour.django.receivers import produce_post_save, produce_post_delete from django.conf import settings import os signal_types = { 'post_save': produce_post_save, 'post_delete': produce_post_delete } if getattr(settings, 'KAFKA_CONFIG', None): try: excl = settings.KAFKA_CONFIG['producers'].get('exclude') except: #TODO: Error handling excl = [] pass for s,f in signal_types.items(): if next((True for ex in excl if s == ex), False): del signal_types[s] for sig in signal_types.keys(): getattr(signals, sig).connect(signal_types[sig])
def handle(self, *args, **options): with mute_signals(post_save, post_delete): logger.info('Destroying all people') Person.objects.all().delete() logger.info('Creating 20 people') people = PersonFactory.create_batch(20) logger.info('Pushing new people to connected clients') SyncConsumer.broadcast('list', people) send_notification( 'info', ('<strong>Database Reset!</strong>' ' Your data is updated automatically!' ' Database intentionally wiped every few minutes.') )
def __init__(self, create=True, update=True, delete=True, custom=None): from actionslog.receivers import action_log_create, action_log_update, action_log_delete self._registry = {} self._signals = {} if create: self._signals[post_save] = action_log_create if update: self._signals[pre_save] = action_log_update if delete: self._signals[post_delete] = action_log_delete if custom is not None: self._signals.update(custom)
def signals(patching): signals = Mock(name='signals') patching('django.db.models.signals.post_save', signals.post_save) patching('django.db.models.signals.post_delete', signals.post_delete) patching('django.db.models.signals.m2m_changed', signals.m2m_changed) return signals
def test_deserialize_user_data(self): """Test that user data is correctly deserialized""" new_course_run = CourseRunFactory.create(edx_course_key='course-v1:MITx+Analog+Learning+100+Aug_2015') new_program = new_course_run.course.program with mute_signals(post_save): user = deserialize_user_data(self.USER_DATA, [new_program]) assert user.username == '{}mario.medina'.format(FAKE_USER_USERNAME_PREFIX) assert user.profile.first_name == 'Mario' assert user.profile.date_of_birth == '1961-04-29' assert CachedEnrollment.objects.filter(user=user, course_run=new_course_run).count() == 1
def create_enrolled_user_batch(num_to_create, program, **kwargs): """Create a batch of users enrolled in a program""" with mute_signals(post_save): new_users = create_user_batch(num_to_create, **kwargs) program_enrollments = [ ProgramEnrollment.objects.create( program=program, user=user ) for user in new_users ] index_program_enrolled_users(program_enrollments) return program_enrollments
def test_index_context_logged_in_no_social_auth(self): """ Assert context values when logged in without a social_auth account """ with mute_signals(post_save): profile = ProfileFactory.create() self.client.force_login(profile.user) ga_tracking_id = FuzzyText().fuzz() with self.settings( GA_TRACKING_ID=ga_tracking_id, ), patch('ui.templatetags.render_bundle._get_bundle') as get_bundle: response = self.client.get('/') bundles = [bundle[0][1] for bundle in get_bundle.call_args_list] assert set(bundles) == { 'common', 'public', 'sentry_client', 'style', 'style_public', } assert response.context['authenticated'] is True assert response.context['username'] is None assert response.context['title'] == HomePage.objects.first().title assert response.context['is_public'] is True assert response.context['has_zendesk_widget'] is False assert response.context['is_staff'] is False assert response.context['programs'] == [] self.assertContains(response, 'Share this page') js_settings = json.loads(response.context['js_settings_json']) assert js_settings['gaTrackingID'] == ga_tracking_id
def test_index_context_logged_in_staff(self, role): """ Assert context values when logged in as staff for a program """ program = ProgramFactory.create(live=True) with mute_signals(post_save): profile = ProfileFactory.create() Role.objects.create( role=role, program=program, user=profile.user, ) self.client.force_login(profile.user) ga_tracking_id = FuzzyText().fuzz() with self.settings( GA_TRACKING_ID=ga_tracking_id, ): response = self.client.get('/') assert response.context['authenticated'] is True assert response.context['username'] is None assert response.context['title'] == HomePage.objects.first().title assert response.context['is_public'] is True assert response.context['has_zendesk_widget'] is False assert response.context['is_staff'] is True assert response.context['programs'] == [ (program, None), ] self.assertContains(response, 'Share this page') js_settings = json.loads(response.context['js_settings_json']) assert js_settings['gaTrackingID'] == ga_tracking_id
def test_500_error_context_logged_in(self): """ Assert context values for 500 error page when logged in """ with mute_signals(post_save): profile = self.create_and_login_user() self.client.force_login(profile.user) with override_settings(EMAIL_SUPPORT='support'), patch( 'ui.templatetags.render_bundle._get_bundle' ) as get_bundle: response = self.client.get('/500/') assert response.context['authenticated'] is True assert response.context['name'] == profile.preferred_name assert response.context['support_email'] == 'support' assert response.context['is_public'] is True assert response.context['has_zendesk_widget'] is True self.assertContains(response, 'Share this page', status_code=status.HTTP_500_INTERNAL_SERVER_ERROR) bundles = [bundle[0][1] for bundle in get_bundle.call_args_list] assert set(bundles) == { 'common', 'public', 'sentry_client', 'style', 'style_public', 'zendesk_widget', }
def create_user_and_login(self, agreed_to_terms_of_service, filled_out): """ Create a user and social auth, and login that user """ with mute_signals(post_save): profile = ProfileFactory.create( agreed_to_terms_of_service=agreed_to_terms_of_service, filled_out=filled_out, ) profile.user.social_auth.create( provider=EdxOrgOAuth2.name, uid="{}_edx".format(profile.user.username) ) self.client.force_login(profile.user)
def test_get_token_for_user_no_discussion_user(): """ Assert that get_token_for_user returns None for a user with no DiscussionUser """ with mute_signals(post_save): user = UserFactory.create() assert DiscussionUser.objects.count() == 0 assert get_token_for_user(user) is None
def test_get_token_for_user_no_username(): """ Assert that get_token_for_user returns None for a user with no username """ with mute_signals(post_save): user = UserFactory.create() DiscussionUser.objects.create(user=user, username=None) assert get_token_for_user(user) is None
def test_sync_user_profile_disabled(settings, mocker): """Test that sync_user_profile doesn't call the api if disabled""" settings.FEATURES['OPEN_DISCUSSIONS_USER_SYNC'] = False mock_task = mocker.patch('discussions.tasks.sync_discussion_user') with mute_signals(post_save): profile = ProfileFactory.create() profile.save() assert mock_task.delay.called is False
def test_create_or_update_discussion_user_no_username(mocker): """Test that create_or_update_discussion_user creates if we don't have a username""" create_mock = mocker.patch('discussions.api.create_discussion_user') update_mock = mocker.patch('discussions.api.update_discussion_user') with mute_signals(post_save): profile = ProfileFactory.create() assert DiscussionUser.objects.count() == 0 api.create_or_update_discussion_user(profile.user_id) assert create_mock.call_count == 1 assert update_mock.call_count == 0 assert DiscussionUser.objects.count() == 1
def test_create_or_update_discussion_user_has_username(mocker): """Test that create_or_update_discussion_user updates if we have a username""" create_mock = mocker.patch('discussions.api.create_discussion_user') update_mock = mocker.patch('discussions.api.update_discussion_user') with mute_signals(post_save): profile = ProfileFactory.create() DiscussionUser.objects.create(user=profile.user, username='username') api.create_or_update_discussion_user(profile.user_id) assert create_mock.call_count == 0 assert update_mock.call_count == 1 assert DiscussionUser.objects.count() == 1
def test_create_discussion_user_error(mock_staff_client): """Verify create_discussion_user handles non 2xx status codes""" mock_staff_client.users.create.return_value.raise_for_status.side_effect = HTTPError with mute_signals(post_save): profile = ProfileFactory.create() discussion_user = DiscussionUser.objects.create(user=profile.user) with pytest.raises(DiscussionUserSyncException) as exc: api.create_discussion_user(discussion_user) assert str(exc.value) == "Error creating discussion user for {}".format(profile.user.username)
def test_update_discussion_user_error(mock_staff_client): """Verify update_discussion_user handles non-2xx status codes""" mock_staff_client.users.update.return_value.raise_for_status.side_effect = HTTPError with mute_signals(post_save): profile = ProfileFactory.create() discussion_user = DiscussionUser.objects.create(user=profile.user, username='username') with pytest.raises(DiscussionUserSyncException) as exc: api.update_discussion_user(discussion_user) assert str(exc.value) == "Error updating discussion user for {}".format(profile.user.username)
def create(cls, *args, **kwargs): """ Overrides the default .create() method to turn off save signals """ with mute_signals(post_save): return super().create(*args, **kwargs)
def test_verified_enroll_factory_fa_create(self): """ Tests that CachedEnrollmentFactory creates additional data for a FA-enabled course run """ assert Line.objects.count() == 0 with mute_signals(post_save): user = UserFactory.create() ProfileFactory.create(user=user) fa_program = FullProgramFactory.create(financial_aid_availability=True) CachedEnrollmentFactory.create(user=user, course_run__course__program=fa_program, verified=True) lines = Line.objects.all() assert len(lines) == 1 assert lines[0].order.status == Order.FULFILLED