customerio-python/tests/test_api.py at 0037d0cf9cde35d2a1d4fc3939647bd913f44bd7 · customerio/customerio-python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
import base64
from datetime import datetime
from functools import partial
import json
import sys
import unittest

from customerio import APIClient, SendEmailRequest, SendPushRequest, SendSMSRequest, SendInboxMessageRequest, Regions, CustomerIOException
from customerio.__version__ import __version__ as ClientVersion
from tests.server import HTTPSTestCase

import requests
from requests.auth import _basic_auth_str

# test uses a self signed certificate so disable the warning messages
requests.packages.urllib3.disable_warnings()


class TestAPIClient(HTTPSTestCase):
    '''Starts server which the client connects to in the following tests'''

    def setUp(self):
        self.client = APIClient(
            key='app_api_key',
            url="https://{addr}:{port}".format(
                addr=self.server.server_address[0], port=self.server.server_port))

        # do not verify the ssl certificate as it is self signed
        # should only be done for tests
        self.client.http.verify = False

    def _check_request(self, resp, rq, *args, **kwargs):
        request = resp.request
        self.assertEqual(request.method, rq['method'])
        self.assertEqual(json.loads(request.body.decode('utf-8')), rq['body'])
        self.assertEqual(request.headers['Authorization'], rq['authorization'])
        self.assertEqual(request.headers['Content-Type'], rq['content_type'])
        self.assertEqual(
            int(request.headers['Content-Length']), len(json.dumps(rq['body'])))
        self.assertTrue(request.url.endswith(rq['url_suffix']),
                        'url: {} expected suffix: {}'.format(request.url, rq['url_suffix']))

    def test_client_setup(self):
        client = APIClient(key='app_api_key')
        self.assertEqual(client.url, 'https://{host}'.format(host=Regions.US.api_host))

        client = APIClient(key='app_api_key', region=Regions.US)
        self.assertEqual(client.url, 'https://{host}'.format(host=Regions.US.api_host))

        client = APIClient(key='app_api_key', region=Regions.EU)
        self.assertEqual(client.url, 'https://{host}'.format(host=Regions.EU.api_host))

        self.assertEqual(self.client.http.headers['User-Agent'], 'Customer.io Python Client/{}'.format(ClientVersion))

        # Raises an exception when an invalid region is passed in
        with self.assertRaises(CustomerIOException):
            APIClient(key='app_api_key', region='au')

    def test_send_email(self):
        data = "1,2,3"
        expected = base64.b64encode(bytes(data,"utf-8")).decode()

        self.client.http.hooks = dict(response=partial(self._check_request, rq={
            'method': 'POST',
            'authorization': "Bearer app_api_key",
            'content_type': 'application/json',
            'url_suffix': '/v1/send/email',
            'body': {"identifiers": {"id":"customer_1"}, "transactional_message_id": 100, "subject": "transactional message", "attachments":{"sample.csv": expected}},
        }))

        email = SendEmailRequest(
            identifiers={"id":"customer_1"},
            transactional_message_id=100,
            subject="transactional message"
        )
        email.attach('sample.csv', data)

        self.client.send_email(email)

    def test_send_push(self):
        self.client.http.hooks = dict(response=partial(self._check_request, rq={
            'method': 'POST',
            'authorization': "Bearer app_api_key",
            'content_type': 'application/json',
            'url_suffix': '/v1/send/push',
            'body': {"identifiers": {"id":"customer_1"}, "transactional_message_id": 100, "title": "transactional push message", "message": "push message content"}
        }))

        push = SendPushRequest(
            identifiers={"id":"customer_1"},
            transactional_message_id=100,
            title="transactional push message",
            message="push message content"
        )

        self.client.send_push(push)

    def test_send_sms(self):
        self.client.http.hooks = dict(response=partial(self._check_request, rq={
            'method': 'POST',
            'authorization': "Bearer app_api_key",
            'content_type': 'application/json',
            'url_suffix': '/v1/send/sms',
            'body': {"identifiers": {"id":"customer_1"}, "transactional_message_id": 100}
        }))

        sms = SendSMSRequest(
            identifiers={"id":"customer_1"},
            transactional_message_id=100,
        )

        self.client.send_sms(sms)

    def test_send_inbox_message(self):
        self.client.http.hooks = dict(response=partial(self._check_request, rq={
            'method': 'POST',
            'authorization': "Bearer app_api_key",
            'content_type': 'application/json',
            'url_suffix': '/v1/send/inbox_message',
            'body': {"identifiers": {"id":"customer_1"}, "transactional_message_id": 100}
        }))

        inbox_message = SendInboxMessageRequest(
            identifiers={"id":"customer_1"},
            transactional_message_id=100,
        )

        self.client.send_inbox_message(inbox_message)

if __name__ == '__main__':
    unittest.main()