diff --git a/Makefile b/Makefile index 93b13ea5..8ea2abef 100644 --- a/Makefile +++ b/Makefile @@ -2,8 +2,8 @@ install: pip install --edit .[test] test: - pylint --rcfile=.pylintrc --reports=y --exit-zero analytics - flake8 --max-complexity=10 --statistics analytics || true - python -m unittest customerio/analytics/test/*.py analytics/test/*.py + pylint --rcfile=.pylintrc --reports=y --exit-zero customerio/analytics + flake8 --max-complexity=10 --statistics customerio/analytics || true + python -m unittest customerio/analytics/test/*.py -v .PHONY: install test diff --git a/analytics/consumer.py b/analytics/consumer.py deleted file mode 100644 index a49f616a..00000000 --- a/analytics/consumer.py +++ /dev/null @@ -1,133 +0,0 @@ -import logging -from threading import Thread -import json -import monotonic -import backoff - - -from customerio.analytics.request import post, APIError, DatetimeSerializer - -from queue import Empty - -MAX_MSG_SIZE = 32 << 10 - -# Our servers only accept batches less than 500KB. Here limit is set slightly -# lower to leave space for extra data that will be added later, eg. "sentAt". -BATCH_SIZE_LIMIT = 475000 - - -class Consumer(Thread): - """Consumes the messages from the client's queue.""" - log = logging.getLogger('customerio') - - def __init__(self, queue, write_key, upload_size=100, host=None, - on_error=None, upload_interval=0.5, gzip=False, retries=10, - timeout=15, proxies=None): - """Create a consumer thread.""" - Thread.__init__(self) - # Make consumer a daemon thread so that it doesn't block program exit - self.daemon = True - self.upload_size = upload_size - self.upload_interval = upload_interval - self.write_key = write_key - self.host = host - self.on_error = on_error - self.queue = queue - self.gzip = gzip - # It's important to set running in the constructor: if we are asked to - # pause immediately after construction, we might set running to True in - # run() *after* we set it to False in pause... and keep running - # forever. - self.running = True - self.retries = retries - self.timeout = timeout - self.proxies = proxies - - def run(self): - """Runs the consumer.""" - self.log.debug('consumer is running...') - while self.running: - self.upload() - - self.log.debug('consumer exited.') - - def pause(self): - """Pause the consumer.""" - self.running = False - - def upload(self): - """Upload the next batch of items, return whether successful.""" - success = False - batch = self.next() - if len(batch) == 0: - return False - - try: - self.request(batch) - success = True - except Exception as e: - self.log.error('error uploading: %s', e) - success = False - if self.on_error: - self.on_error(e, batch) - finally: - # mark items as acknowledged from queue - for _ in batch: - self.queue.task_done() - return success - - def next(self): - """Return the next batch of items to upload.""" - queue = self.queue - items = [] - - start_time = monotonic.monotonic() - total_size = 0 - - while len(items) < self.upload_size: - elapsed = monotonic.monotonic() - start_time - if elapsed >= self.upload_interval: - break - try: - item = queue.get( - block=True, timeout=self.upload_interval - elapsed) - item_size = len(json.dumps( - item, cls=DatetimeSerializer).encode()) - if item_size > MAX_MSG_SIZE: - self.log.error( - 'Item exceeds 32kb limit, dropping. (%s)', str(item)) - continue - items.append(item) - total_size += item_size - if total_size >= BATCH_SIZE_LIMIT: - self.log.debug( - 'hit batch size limit (size: %d)', total_size) - break - except Empty: - break - - return items - - def request(self, batch): - """Attempt to upload the batch and retry before raising an error """ - - def fatal_exception(exc): - if isinstance(exc, APIError): - # retry on server errors and client errors - # with 429 status code (rate limited), - # don't retry on other client errors - return (400 <= exc.status < 500) and exc.status != 429 - else: - # retry on all other errors (eg. network) - return False - - @backoff.on_exception( - backoff.expo, - Exception, - max_tries=self.retries + 1, - giveup=fatal_exception) - def send_request(): - post(self.write_key, self.host, gzip=self.gzip, - timeout=self.timeout, batch=batch, proxies=self.proxies) - - send_request() diff --git a/analytics/test/client.py b/analytics/test/client.py deleted file mode 100644 index 80ceaa3d..00000000 --- a/analytics/test/client.py +++ /dev/null @@ -1,342 +0,0 @@ -from datetime import date, datetime -import unittest -import time -import mock - -from customerio.analytics.version import VERSION -from customerio.analytics.client import Client - - -class TestClient(unittest.TestCase): - - def fail(self): - """Mark the failure handler""" - self.failed = True - - def setUp(self): - self.failed = False - self.client = Client('testsecret', on_error=self.fail) - - def test_requires_write_key(self): - self.assertRaises(AssertionError, Client) - - def test_empty_flush(self): - self.client.flush() - - def test_basic_track(self): - client = self.client - success, msg = client.track('userId', 'python test event') - client.flush() - self.assertTrue(success) - self.assertFalse(self.failed) - - self.assertEqual(msg['event'], 'python test event') - self.assertTrue(isinstance(msg['timestamp'], str)) - self.assertTrue(isinstance(msg['messageId'], str)) - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['properties'], {}) - self.assertEqual(msg['type'], 'track') - - def test_stringifies_user_id(self): - # A large number that loses precision in node: - # node -e "console.log(157963456373623802 + 1)" > 157963456373623800 - client = self.client - success, msg = client.track( - user_id=157963456373623802, event='python test event') - client.flush() - self.assertTrue(success) - self.assertFalse(self.failed) - - self.assertEqual(msg['userId'], '157963456373623802') - self.assertEqual(msg['anonymousId'], None) - - def test_stringifies_anonymous_id(self): - # A large number that loses precision in node: - # node -e "console.log(157963456373623803 + 1)" > 157963456373623800 - client = self.client - success, msg = client.track( - anonymous_id=157963456373623803, event='python test event') - client.flush() - self.assertTrue(success) - self.assertFalse(self.failed) - - self.assertEqual(msg['userId'], None) - self.assertEqual(msg['anonymousId'], '157963456373623803') - - def test_advanced_track(self): - client = self.client - success, msg = client.track( - 'userId', 'python test event', {'property': 'value'}, - {'ip': '192.168.0.1'}, datetime(2014, 9, 3), 'anonymousId', - {'Amplitude': True}, 'messageId') - - self.assertTrue(success) - - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') - self.assertEqual(msg['properties'], {'property': 'value'}) - self.assertEqual(msg['integrations'], {'Amplitude': True}) - self.assertEqual(msg['context']['ip'], '192.168.0.1') - self.assertEqual(msg['event'], 'python test event') - self.assertEqual(msg['anonymousId'], 'anonymousId') - self.assertEqual(msg['context']['library'], { - 'name': 'analytics-python', - 'version': VERSION - }) - self.assertEqual(msg['messageId'], 'messageId') - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'track') - - def test_basic_identify(self): - client = self.client - success, msg = client.identify('userId', {'trait': 'value'}) - client.flush() - self.assertTrue(success) - self.assertFalse(self.failed) - - self.assertEqual(msg['traits'], {'trait': 'value'}) - self.assertTrue(isinstance(msg['timestamp'], str)) - self.assertTrue(isinstance(msg['messageId'], str)) - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'identify') - - def test_advanced_identify(self): - client = self.client - success, msg = client.identify( - 'userId', {'trait': 'value'}, {'ip': '192.168.0.1'}, - datetime(2014, 9, 3), 'anonymousId', {'Amplitude': True}, - 'messageId') - - self.assertTrue(success) - - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') - self.assertEqual(msg['integrations'], {'Amplitude': True}) - self.assertEqual(msg['context']['ip'], '192.168.0.1') - self.assertEqual(msg['traits'], {'trait': 'value'}) - self.assertEqual(msg['anonymousId'], 'anonymousId') - self.assertEqual(msg['context']['library'], { - 'name': 'analytics-python', - 'version': VERSION - }) - self.assertTrue(isinstance(msg['timestamp'], str)) - self.assertEqual(msg['messageId'], 'messageId') - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'identify') - - def test_basic_group(self): - client = self.client - success, msg = client.group('userId', 'groupId') - client.flush() - self.assertTrue(success) - self.assertFalse(self.failed) - - self.assertEqual(msg['groupId'], 'groupId') - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'group') - - def test_advanced_group(self): - client = self.client - success, msg = client.group( - 'userId', 'groupId', {'trait': 'value'}, {'ip': '192.168.0.1'}, - datetime(2014, 9, 3), 'anonymousId', {'Amplitude': True}, - 'messageId') - - self.assertTrue(success) - - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') - self.assertEqual(msg['integrations'], {'Amplitude': True}) - self.assertEqual(msg['context']['ip'], '192.168.0.1') - self.assertEqual(msg['traits'], {'trait': 'value'}) - self.assertEqual(msg['anonymousId'], 'anonymousId') - self.assertEqual(msg['context']['library'], { - 'name': 'analytics-python', - 'version': VERSION - }) - self.assertTrue(isinstance(msg['timestamp'], str)) - self.assertEqual(msg['messageId'], 'messageId') - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'group') - - def test_basic_alias(self): - client = self.client - success, msg = client.alias('previousId', 'userId') - client.flush() - self.assertTrue(success) - self.assertFalse(self.failed) - self.assertEqual(msg['previousId'], 'previousId') - self.assertEqual(msg['userId'], 'userId') - - def test_basic_page(self): - client = self.client - success, msg = client.page('userId', name='name') - self.assertFalse(self.failed) - client.flush() - self.assertTrue(success) - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'page') - self.assertEqual(msg['name'], 'name') - - def test_advanced_page(self): - client = self.client - success, msg = client.page( - 'userId', 'category', 'name', {'property': 'value'}, - {'ip': '192.168.0.1'}, datetime(2014, 9, 3), 'anonymousId', - {'Amplitude': True}, 'messageId') - - self.assertTrue(success) - - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') - self.assertEqual(msg['integrations'], {'Amplitude': True}) - self.assertEqual(msg['context']['ip'], '192.168.0.1') - self.assertEqual(msg['properties'], {'property': 'value'}) - self.assertEqual(msg['anonymousId'], 'anonymousId') - self.assertEqual(msg['context']['library'], { - 'name': 'analytics-python', - 'version': VERSION - }) - self.assertEqual(msg['category'], 'category') - self.assertTrue(isinstance(msg['timestamp'], str)) - self.assertEqual(msg['messageId'], 'messageId') - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'page') - self.assertEqual(msg['name'], 'name') - - def test_basic_screen(self): - client = self.client - success, msg = client.screen('userId', name='name') - client.flush() - self.assertTrue(success) - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'screen') - self.assertEqual(msg['name'], 'name') - - def test_advanced_screen(self): - client = self.client - success, msg = client.screen( - 'userId', 'category', 'name', {'property': 'value'}, - {'ip': '192.168.0.1'}, datetime(2014, 9, 3), 'anonymousId', - {'Amplitude': True}, 'messageId') - - self.assertTrue(success) - - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') - self.assertEqual(msg['integrations'], {'Amplitude': True}) - self.assertEqual(msg['context']['ip'], '192.168.0.1') - self.assertEqual(msg['properties'], {'property': 'value'}) - self.assertEqual(msg['anonymousId'], 'anonymousId') - self.assertEqual(msg['context']['library'], { - 'name': 'analytics-python', - 'version': VERSION - }) - self.assertTrue(isinstance(msg['timestamp'], str)) - self.assertEqual(msg['messageId'], 'messageId') - self.assertEqual(msg['category'], 'category') - self.assertEqual(msg['userId'], 'userId') - self.assertEqual(msg['type'], 'screen') - self.assertEqual(msg['name'], 'name') - - def test_flush(self): - client = self.client - # set up the consumer with more requests than a single batch will allow - for _ in range(1000): - _, _ = client.identify('userId', {'trait': 'value'}) - # We can't reliably assert that the queue is non-empty here; that's - # a race condition. We do our best to load it up though. - client.flush() - # Make sure that the client queue is empty after flushing - self.assertTrue(client.queue.empty()) - - def test_shutdown(self): - client = self.client - # set up the consumer with more requests than a single batch will allow - for _ in range(1000): - _, _ = client.identify('userId', {'trait': 'value'}) - client.shutdown() - # we expect two things after shutdown: - # 1. client queue is empty - # 2. consumer thread has stopped - self.assertTrue(client.queue.empty()) - for consumer in client.consumers: - self.assertFalse(consumer.is_alive()) - - def test_synchronous(self): - client = Client('testsecret', sync_mode=True) - - success, _ = client.identify('userId') - self.assertFalse(client.consumers) - self.assertTrue(client.queue.empty()) - self.assertTrue(success) - - def test_overflow(self): - client = Client('testsecret', max_queue_size=1) - # Ensure consumer thread is no longer uploading - client.join() - - for _ in range(10): - client.identify('userId') - - success, _ = client.identify('userId') - # Make sure we are informed that the queue is at capacity - self.assertFalse(success) - - def test_success_on_invalid_write_key(self): - client = Client('bad_key', on_error=self.fail) - client.track('userId', 'event') - client.flush() - self.assertFalse(self.failed) - - def test_numeric_user_id(self): - self.client.track(1234, 'python event') - self.client.flush() - self.assertFalse(self.failed) - - def test_identify_with_date_object(self): - client = self.client - success, msg = client.identify( - 'userId', - { - 'birthdate': date(1981, 2, 2), - }, - ) - client.flush() - self.assertTrue(success) - self.assertFalse(self.failed) - - self.assertEqual(msg['traits'], {'birthdate': date(1981, 2, 2)}) - - def test_gzip(self): - client = Client('testsecret', on_error=self.fail, gzip=True) - for _ in range(10): - client.identify('userId', {'trait': 'value'}) - client.flush() - self.assertFalse(self.failed) - - def test_user_defined_upload_size(self): - client = Client('testsecret', on_error=self.fail, - upload_size=10, upload_interval=3) - - def mock_post_fn(**kwargs): - self.assertEqual(len(kwargs['batch']), 10) - - # the post function should be called 2 times, with a batch size of 10 - # each time. - with mock.patch('analytics.consumer.post', side_effect=mock_post_fn) \ - as mock_post: - for _ in range(20): - client.identify('userId', {'trait': 'value'}) - time.sleep(1) - self.assertEqual(mock_post.call_count, 2) - - def test_user_defined_timeout(self): - client = Client('testsecret', timeout=10) - for consumer in client.consumers: - self.assertEqual(consumer.timeout, 10) - - def test_default_timeout_15(self): - client = Client('testsecret') - for consumer in client.consumers: - self.assertEqual(consumer.timeout, 15) - - def test_proxies(self): - client = Client('testsecret', proxies='203.243.63.16:80') - success, msg = client.identify('userId', {'trait': 'value'}) - self.assertTrue(success) diff --git a/customerio/analytics/client.py b/customerio/analytics/client.py index e063820e..4185d776 100644 --- a/customerio/analytics/client.py +++ b/customerio/analytics/client.py @@ -162,7 +162,6 @@ def group(self, user_id=None, group_id=None, traits=None, context=None, traits = traits or {} context = context or {} integrations = integrations or {} - require('user_id or anonymous_id', user_id or anonymous_id, ID_TYPES) require('group_id', group_id, ID_TYPES) require('traits', traits, dict) diff --git a/customerio/analytics/test/__init__.py b/customerio/analytics/test/__init__.py index 5b1c7fd9..d936c6d3 100644 --- a/customerio/analytics/test/__init__.py +++ b/customerio/analytics/test/__init__.py @@ -2,7 +2,6 @@ import pkgutil import logging import sys -import analytics from customerio.analytics.client import Client from customerio import analytics @@ -10,7 +9,7 @@ def all_names(): for _, modname, _ in pkgutil.iter_modules(__path__): - yield 'analytics.test.' + modname + yield 'customerio.analytics.test.' + modname def all(): diff --git a/customerio/analytics/test/client.py b/customerio/analytics/test/client.py index 000776d7..512e136e 100644 --- a/customerio/analytics/test/client.py +++ b/customerio/analytics/test/client.py @@ -72,7 +72,7 @@ def test_advanced_track(self): self.assertTrue(success) - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') + self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00.000+00:00') self.assertEqual(msg['properties'], {'property': 'value'}) self.assertEqual(msg['integrations'], {'Amplitude': True}) self.assertEqual(msg['context']['ip'], '192.168.0.1') @@ -108,7 +108,7 @@ def test_advanced_identify(self): self.assertTrue(success) - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') + self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00.000+00:00') self.assertEqual(msg['integrations'], {'Amplitude': True}) self.assertEqual(msg['context']['ip'], '192.168.0.1') self.assertEqual(msg['traits'], {'trait': 'value'}) @@ -142,7 +142,7 @@ def test_advanced_group(self): self.assertTrue(success) - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') + self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00.000+00:00') self.assertEqual(msg['integrations'], {'Amplitude': True}) self.assertEqual(msg['context']['ip'], '192.168.0.1') self.assertEqual(msg['traits'], {'trait': 'value'}) @@ -184,7 +184,7 @@ def test_advanced_page(self): self.assertTrue(success) - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') + self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00.000+00:00') self.assertEqual(msg['integrations'], {'Amplitude': True}) self.assertEqual(msg['context']['ip'], '192.168.0.1') self.assertEqual(msg['properties'], {'property': 'value'}) @@ -218,7 +218,7 @@ def test_advanced_screen(self): self.assertTrue(success) - self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00+00:00') + self.assertEqual(msg['timestamp'], '2014-09-03T00:00:00.000+00:00') self.assertEqual(msg['integrations'], {'Amplitude': True}) self.assertEqual(msg['context']['ip'], '192.168.0.1') self.assertEqual(msg['properties'], {'property': 'value'}) @@ -325,7 +325,7 @@ def mock_post_fn(*args, **kwargs): # the post function should be called 2 times, with a batch size of 10 # each time. - with mock.patch('analytics.consumer.post', side_effect=mock_post_fn) \ + with mock.patch('customerio.analytics.consumer.post', side_effect=mock_post_fn) \ as mock_post: for _ in range(20): client.identify('userId', {'trait': 'value'}) diff --git a/customerio/analytics/test/consumer.py b/customerio/analytics/test/consumer.py index cf62acc4..5ef02fc5 100644 --- a/customerio/analytics/test/consumer.py +++ b/customerio/analytics/test/consumer.py @@ -8,7 +8,7 @@ except ImportError: from Queue import Queue -from analytics import Consumer, MAX_MSG_SIZE, request +from customerio.analytics.consumer import Consumer, MAX_MSG_SIZE from customerio.analytics.request import APIError @@ -59,7 +59,7 @@ def test_upload_interval(self): upload_interval = 0.3 consumer = Consumer(q, 'testsecret', upload_size=10, upload_interval=upload_interval) - with mock.patch('analytics.consumer.post') as mock_post: + with mock.patch('customerio.analytics.consumer.post') as mock_post: consumer.start() for i in range(0, 3): track = { @@ -79,7 +79,7 @@ def test_multiple_uploads_per_interval(self): upload_size = 10 consumer = Consumer(q, 'testsecret', upload_size=upload_size, upload_interval=upload_interval) - with mock.patch('analytics.consumer.post') as mock_post: + with mock.patch('customerio.analytics.consumer.post') as mock_post: consumer.start() for i in range(0, upload_size * 2): track = { @@ -110,7 +110,7 @@ def mock_post(*args, **kwargs): raise expected_exception mock_post.call_count = 0 - with mock.patch('analytics.consumer.post', + with mock.patch('customerio.analytics.consumer.post', mock.Mock(side_effect=mock_post)): track = { 'type': 'track', @@ -190,7 +190,7 @@ def mock_post_fn(_, data, **kwargs): % len(data.encode())) return res - with mock.patch('analytics.request._session.post', + with mock.patch('customerio.analytics.request._session.post', side_effect=mock_post_fn) as mock_post: consumer.start() for _ in range(0, n_msgs + 2): diff --git a/customerio/analytics/test/module.py b/customerio/analytics/test/module.py deleted file mode 100644 index 0a3849e4..00000000 --- a/customerio/analytics/test/module.py +++ /dev/null @@ -1,46 +0,0 @@ -import unittest - -from customerio.analytics import analytics - - -class TestModule(unittest.TestCase): - - def setUp(self): - self.failed = False - analytics.write_key = 'testsecret' - analytics.on_error = self.failed - - def test_no_write_key(self): - analytics.write_key = None - self.assertRaises(Exception, analytics.track) - - def test_no_host(self): - analytics.host = None - self.assertRaises(Exception, analytics.track) - - def test_track(self): - analytics.track('userId', 'python module event') - analytics.flush() - - def test_identify(self): - analytics.identify('userId', {'email': 'user@email.com'}) - analytics.flush() - - def test_group(self): - analytics.group('userId', 'groupId') - analytics.flush() - - def test_alias(self): - analytics.alias('previousId', 'userId') - analytics.flush() - - def test_page(self): - analytics.page('userId') - analytics.flush() - - def test_screen(self): - analytics.screen('userId') - analytics.flush() - - def test_flush(self): - analytics.flush() diff --git a/customerio/analytics/test/utils.py b/customerio/analytics/test/utils.py index cb9575a2..73907ce5 100644 --- a/customerio/analytics/test/utils.py +++ b/customerio/analytics/test/utils.py @@ -4,7 +4,7 @@ from dateutil.tz import tzutc -from analytics import utils +from customerio.analytics import utils class TestUtils(unittest.TestCase): diff --git a/setup.py b/setup.py index 1310350a..143a9ab5 100644 --- a/setup.py +++ b/setup.py @@ -37,7 +37,7 @@ maintainer='Customer.io', maintainer_email='cdp@customer.io', test_suite='analytics.test.all', - packages=['customerio.analytics', 'analytics.test'], + packages=['customerio.analytics'], python_requires='>=3.6.0', license='MIT License', install_requires=install_requires,