Python rest_framework.test 模块，APIClient() 实例源码

def setUp(self):
        """Define the test client and other test variables.
        """
        user = User.objects.create(username="simpleuser")

        # Initialize client and force it to use authentication
        self.client = APIClient()
        self.client.force_authenticate(user=user)

        self.rasterbucket_data = {
            'name': 'dem processing result',
            'owner': user.id}
        self.response = self.client.post(
            reverse('api.rasterbuckets'),
            self.rasterbucket_data,
            format="json")

def test_POST_user_detail(self):
        token = Token.objects.get_or_create(user=self.user)[0]
        client = APIClient()
        client.credentials(HTTP_AUTHORIZATION='Token ' + token.key)

        name = self.user_data['first_name']
        user = User.objects.filter(first_name=name)[0]
        temp = {'first_name': 'test', 'email': user.email}

        # TODO: Figure out why we need to pass 'email' to make patch request
        response = client.patch(reverse('user-detail', kwargs={'pk': user.pk}),
                                data=temp)

        self.assertNotEqual(user.first_name, response.data['first_name'])
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(temp['first_name'], response.data['first_name'])

def setUp(self):

        # Disable Signals
        post_save.disconnect(post_save_diary, sender=Diary)

        # Run celery task synchronous
        app.conf.update(CELERY_ALWAYS_EAGER=True)

        self.test_username = TEST_USERNAME
        self.test_password = TEST_PASSWORD
        self.test_email = TEST_EMAIL

        # Create a user
        self.user = get_user_model().objects.create_user(
            username=self.test_username,
            password=self.test_password,
            email=self.test_email,
        )

        # Login
        self.client = APIClient()
        self.client.login(
            username=self.test_username,
            password=self.test_password,
        )

def test_create_database():
    client = APIClient()
    data = {'name': 'first_database'}
    response = client.post(
        reverse('pgdb:create_database_api'),
        data,
        format="json")
    assert response.status_code == status.HTTP_201_CREATED
    res = Database.objects.all()
    assert len(res) == 2
    db = Database.objects.get(name='first_database')
    assert db.neo4j_http_port == 7404
    assert db.influxdb_http_port == 8404
    assert db.influxdb_udp_port == 8406
    response = client.delete(
        reverse('pgdb:database_api', args=[data['name']]),
        format="json")
    assert response.status_code == status.HTTP_202_ACCEPTED

def test_login_logout(me, my_password):
    api = APIClient()
    assert not_authenticated(api)

    # Login
    resp = api.post('/api/auth/login/', data={
        'email': me.email,
        'password': my_password,
    })
    assert resp.status_code == 204
    assert is_authenticated(api)

    # Logout
    resp = api.post('/api/auth/logout/')
    assert resp.status_code == 204
    assert not_authenticated(api)

    # It's ok to log out again
    resp = api.post('/api/auth/logout/')
    assert resp.status_code == 204
    assert not_authenticated(api)

def test_login_invalid(me, my_password):
    error_resp = {
        'nonFieldErrors': ['Unable to login with provided credentials.'],
    }

    api = APIClient()
    assert not_authenticated(api)

    resp = api.post('/api/auth/login/', data={
        'email': me.email,
        'password': my_password + 's',
    })
    assert resp.status_code == 400
    assert resp.json() == error_resp
    assert not_authenticated(api)

    resp = api.post('/api/auth/login/', data={
        'email': me.email + 's',
        'password': my_password,
    })
    assert resp.status_code == 400
    assert resp.json() == error_resp
    assert not_authenticated(api)

def setUp(self):
        super(AnnotationTestCase, self).setUp()

        self.client = APIClient()
        self.index_create_url = reverse("annotations-list")
        self.annotation = {
            "annotator_schema_version": "v1.0",
            "text": "A note I wrote",
            "quote": "the text that was annotated",
            "uri": "http://example.com",
            "ranges": [
                {
                    "start": "/p[69]/span/span",
                    "end": "/p[70]/span/span",
                    "startOffset": 0,
                    "endOffset": 120
                }
            ]
        }

def test_can_join_if_invited(self):
    """ Test it's possible to join organization if invited """
    self.test_can_invite_user()
    self.assertTrue(self.user2 not in self.organization.members.all())

    mail.outbox = []
    self.assertTrue(len(mail.outbox) == 0)

    client = APIClient()
    client.force_authenticate(self.user2)
    response = client.post(reverse("organization-join", ["test-organization"]), {}, format="json")
    self.assertTrue(response.status_code == 200)
    self.assertTrue(response.data["detail"] == "Joined organization.")
    self.assertTrue(self.user2 in self.organization.members.all())


    subjects = [x.subject for x in mail.outbox]
    if is_email_enabled("default", "userJoined-toUser"): # pragma: no cover
      self.assertTrue(get_email_subject("default", "userJoined-toUser", "You have joined an organization"))
    if is_email_enabled("default", "userJoined-toOwner"): # pragma: no cover
      self.assertTrue(get_email_subject("default", "userJoined-toOwner", "An user has joined an organization you own"))

def test_creating_project_creates_objects_on_correct_channel(self):
    """ Assert creating a project creates the object on the correct channel """
    client = APIClient()
    client.login(email=self.email, password=self.password, channel="test-channel")

    # Create
    data = copy.copy(base_project)
    response = client.post(reverse("project-list"), data, format="json", HTTP_X_OVP_CHANNEL="test-channel")
    self.assertTrue(response.status_code == 201)
    self.assertTrue(Project.objects.last().channel.slug == "test-channel")
    self.assertTrue(Work.objects.last().channel.slug == "test-channel")

    # Modify
    updated_project = {"disponibility": {"type": "job", "job": {"dates": [{"name": "update", "start_date": "2013-01-29T12:34:56.123Z", "end_date": "2013-01-29T13:34:56.123Z"}, {"name": "test1", "start_date": "2013-02-01T12:34:56.123Z", "end_date": "2013-02-01T13:34:56.123Z"}]}}}
    response = client.patch(reverse("project-detail", ["test-project"]), updated_project, format="json", HTTP_X_OVP_CHANNEL="test-channel")
    self.assertTrue(Job.objects.last().channel.slug == "test-channel")
    self.assertTrue(JobDate.objects.last().channel.slug == "test-channel")

def test_cant_create_modify_on_another_channel(self):
    """ Assert it's not possible to create or modify object on another channel """
    client = APIClient()
    client.login(email=self.email, password=self.password, channel="test-channel")

    # Create
    data = copy.copy(base_project)
    response = client.post(reverse("project-list"), data, format="json", HTTP_X_OVP_CHANNEL="default")
    self.assertTrue(response.status_code == 400)

    response = client.post(reverse("project-list"), data, format="json", HTTP_X_OVP_CHANNEL="test-channel")
    self.assertTrue(response.status_code == 201)

    # Modify
    updated_project = {"disponibility": {"type": "job", "job": {"dates": [{"name": "update", "start_date": "2013-01-29T12:34:56.123Z", "end_date": "2013-01-29T13:34:56.123Z"}, {"name": "test1", "start_date": "2013-02-01T12:34:56.123Z", "end_date": "2013-02-01T13:34:56.123Z"}]}}}
    response = client.patch(reverse("project-detail", ["test-project"]), updated_project, format="json", HTTP_X_OVP_CHANNEL="default")
    self.assertTrue(response.status_code == 400)

def test_available_country_cities(self):
    client = APIClient()

    response = client.get(reverse("available-country-cities", ["Brazil"]), format="json")
    self.assertEqual(response.status_code, 200)
    self.assertEqual(len(response.data["projects"]), 1)
    self.assertEqual(len(response.data["organizations"]), 1)
    self.assertEqual(len(response.data["common"]), 1)
    self.assertIn("Campinas", response.data["projects"])
    self.assertIn("Santo André", response.data["organizations"])
    self.assertIn("São Paulo", response.data["common"])

    response = client.get(reverse("available-country-cities", ["United States"]), format="json")
    self.assertEqual(response.status_code, 200)
    self.assertEqual(len(response.data["projects"]), 0)
    self.assertEqual(len(response.data["organizations"]), 0)
    self.assertEqual(len(response.data["common"]), 1)
    self.assertIn("New York", response.data["common"])

def test_phone_number_update(self):
        client = APIClient()
        client.force_login(self.test_user)

        original_number = '+6171234567'
        UserPhoneNumberDetails.objects.create(user=self.test_user, phone_number=original_number)
        response = client.get(self.url)
        self.assertEqual(response.data['phone_number'], original_number)

        updated_phone_number = '+16175555555'
        details = {
            'phone_number': updated_phone_number
        }

        response = client.post(self.url, data=details)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data['phone_number'], updated_phone_number)

def setUp(self):
        super(IncomingCallPerformanceTest, self).setUp()
        self.client = APIClient()

        self.ios_app, created = App.objects.get_or_create(platform='apns', app_id='com.voipgrid.vialer')

        two_weeks_ago = datetime.now() - timedelta(days=14)
        Device.objects.create(
            name='test device',
            token='a652aee84bdec6c2859eec89a6e5b1a42c400fba43070f404148f27b502610b6',
            sip_user_id='123456789',
            os_version='8.3',
            client_version='1.0',
            last_seen=two_weeks_ago,
            app=self.ios_app
        )

def setUp(self):
        super(RegisterDeviceTest, self).setUp()
        self.client = APIClient()

        self.ios_app, created = App.objects.get_or_create(platform='apns', app_id='com.voipgrid.vialer')
        self.android_app, created = App.objects.get_or_create(platform='android', app_id='com.voipgrid.vialer')
        self.data = {
            'name': 'test device',
            'token': 'a652aee84bdec6c2859eec89a6e5b1a42c400fba43070f404148f27b502610b6',
            'sip_user_id': '123456789',
            'os_version': '8.3',
            'client_version': '1.0',
            'app': 'com.voipgrid.vialer',
        }

        self.ios_url = '/api/apns-device/'
        self.android_url = '/api/android-device/'

def test_user_can_login(self):
        user = get_user_model().objects.create_user('test', email='test', password='test')

        client = APIClient()

        response = client.post(reverse('auth-login'), {'username': 'test', 'password': 'test'})
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        auth_token = AuthToken.objects.get(user=user)

        data = json.loads(response.content.decode('utf-8'))
        token = data['token']
        self.assertEqual(token, auth_token.key)

def test_user_cant_login_with_invalid_username(self):
        get_user_model().objects.create_user('test', email='test', password='test')

        client = APIClient()

        response = client.post(reverse('auth-login'), {'username': 'invalid', 'password': 'test'})
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

        data = json.loads(response.content.decode('utf-8'))
        self.assertEqual(data, {u'detail': u'Invalid email or password.  Please try logging in again.'})

def test_user_cant_login_with_invalid_password(self):
        get_user_model().objects.create_user('test', email='test', password='test')

        client = APIClient()

        response = client.post(reverse('auth-login'), {'username': 'test', 'password': 'invalid'})
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

        data = json.loads(response.content.decode('utf-8'))
        self.assertEqual(data, {u'detail': u'Invalid email or password.  Please try logging in again.'})

def test_user_can_login_with_device(self):
        user = get_user_model().objects.create_user('test', email='test', password='test')

        client = APIClient()

        response = client.post(reverse('auth-login'), {'username': 'test', 'password': 'test', 'device_id': 'device-1'})
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        auth_token = AuthToken.objects.get(user=user, device_id='device-1')

        data = json.loads(response.content.decode('utf-8'))
        token = data['token']
        self.assertEqual(token, auth_token.key)

        # Same device ID should return same token
        response = client.post(reverse('auth-login'), {'username': 'test', 'password': 'test', 'device_id': 'device-1'})
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        data = json.loads(response.content.decode('utf-8'))
        token = data['token']
        self.assertEqual(token, auth_token.key)

        # Different device ID should return different token
        response = client.post(reverse('auth-login'), {'username': 'test', 'password': 'test', 'device_id': 'device-2'})
        self.assertEqual(response.status_code, status.HTTP_200_OK)

        auth_token = AuthToken.objects.get(user=user, device_id='device-2')

        data = json.loads(response.content.decode('utf-8'))
        token = data['token']
        self.assertEqual(token, auth_token.key)

def test_user_can_signup(self):
        client = APIClient()

        response = client.post(reverse('auth-signup'), {'username': 'test', 'password': 'test'})
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        user = get_user_model().objects.get(username='test')
        self.assertEqual(user.email, user.username)

        auth_token = AuthToken.objects.get(user=user)

        data = json.loads(response.content.decode('utf-8'))
        token = data['token']
        self.assertEqual(token, auth_token.key)

def test_username_email_address_validation(self):
        with self.settings(CHECK_USERNAME_IS_EMAIL=True):
            client = APIClient()

            response = client.post(reverse('auth-signup'), {'username': 'test', 'password': 'test'})
            self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)

            data = json.loads(response.content.decode('utf-8'))
            self.assertEqual(data, {u'detail': u'Please provide a valid email address.'})

def test_user_can_access_authenticated_endpoint(self):
        client = APIClient()

        response = client.get(reverse('token-authenticated-view'))
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

        user = get_user_model().objects.create_user('test', email='test', password='test')
        token = AuthToken.objects.create(user=user)
        client.credentials(HTTP_AUTHORIZATION='DToken ' + token.key)

        response = client.get(reverse('token-authenticated-view'))
        self.assertEqual(response.status_code, status.HTTP_200_OK)

def test_authorization_is_enforced(self):
        """Test that the api has user authorization.
        """
        my_client = APIClient()
        res = my_client.get(
            reverse('api.rasterbuckets'),
            kwargs={'pk': 3}, format="json")
        self.assertEqual(res.status_code, status.HTTP_401_UNAUTHORIZED)

def test_upload_endpoint_that_return_success(self):
        url = reverse('files:file-upload')
        filename = 'bahtiyar.jpg'

        client = APIClient()
        client.login(username='johndoe', password='johndoe')
        client.credentials(HTTP_CONTENT_DISPOSITION='attachment; filename={}'.format(filename))

        file = DjangoFile(open('/data/{}'.format(filename), 'rb'))

        response = client.post(url, {'name': filename, 'attachment': file})

        file.close()

        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

def setUp(self):
        self.user = User.objects.create_user('tester', password='test')
        self.user.save()
        token_response = self.client.post('/auth/login/', {'username': 'tester', 'password': 'test'})

        self.client = APIClient()
        self.token = token_response.data['token']

def setUpClass(cls):
        """Set up class."""

        cls.client = APIClient()

        # Create test user.
        cls.user = factories.TestUsernameSuperAdminUserFactory()

def setUpClass(cls):
        """Set up class."""

        cls.client = APIClient()

        # Create test user.
        cls.user = factories.TestUsernameSuperAdminUserFactory()

def setUp(self):
        self.user   = User.objects.create_user(**fixtures['user'])
        fixtures['question']['author'] = self.user
        self.client = APIClient()

def setUp(self):
        self.usera  = User.objects.create_user(**fixtures['user'])
        self.userb  = User.objects.create_user(**fixtures['voter'])

        fixtures['question']['author'] = self.usera
        fixtures['answer']['author']   = self.userb

        self.question = Question.objects.create(**fixtures['question'])
        fixtures['answer']['question'] = self.question

        self.client = APIClient()

def api_client():
    return APIClient()

def user_api_client(user_factory):
    api_client = APIClient()
    user = user_factory()  # don't use the same user as operator_api_client
    token_authenticate(api_client, user)
    return api_client

def staff_api_client(staff_user):
    api_client = APIClient()
    token_authenticate(api_client, staff_user)
    return api_client