From 6e8639ea03ba11ac18126a02af529c7cf33f26a8 Mon Sep 17 00:00:00 2001 From: sydneycollins-cio <149523950+sydneycollins-cio@users.noreply.github.com> Date: Mon, 5 Oct 2026 17:13:44 -0400 Subject: [PATCH 1/5] feat: support the new ak_/wk_ API keys Co-Authored-By: Claude Opus 5.5 --- customerio/analytics/request.py | 13 ++++++-- customerio/analytics/test/request.py | 50 ++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 2 deletions(-) diff --git a/customerio/analytics/request.py b/customerio/analytics/request.py index 8573fe85..206d77b2 100644 --- a/customerio/analytics/request.py +++ b/customerio/analytics/request.py @@ -3,6 +3,7 @@ from gzip import GzipFile import logging import json +import re from dateutil.tz import tzutc from requests.auth import HTTPBasicAuth from requests import sessions @@ -12,20 +13,28 @@ _session = sessions.Session() +_API_KEY_PREFIX = re.compile(r'^(ak|wk)_(us|eu)_') + def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs): """Post the `kwargs` to the API""" log = logging.getLogger('customerio') body = kwargs body["sentAt"] = datetime.utcnow().replace(tzinfo=tzutc()).isoformat() - url = remove_trailing_slash(host or 'https://cdp.customer.io') + '/v1/batch' - auth = HTTPBasicAuth(write_key, '') + api_key = _API_KEY_PREFIX.match(write_key) + default_host = 'https://cdp.customer.io' + if api_key and api_key.group(2) == 'eu': + default_host = 'https://cdp-eu.customer.io' + url = remove_trailing_slash(host or default_host) + '/v1/batch' + auth = None if api_key else HTTPBasicAuth(write_key, '') data = json.dumps(body, cls=DatetimeSerializer) log.debug('making request: %s', data) headers = { 'Content-Type': 'application/json', 'User-Agent': 'analytics-python/' + VERSION } + if api_key: + headers['Authorization'] = 'Bearer ' + write_key if gzip: headers['Content-Encoding'] = 'gzip' buf = BytesIO() diff --git a/customerio/analytics/test/request.py b/customerio/analytics/test/request.py index ecc25091..2658090f 100644 --- a/customerio/analytics/test/request.py +++ b/customerio/analytics/test/request.py @@ -84,3 +84,53 @@ def test_no_proxies_by_default(self): mock_post.assert_called_once() _, call_kwargs = mock_post.call_args self.assertNotIn('proxies', call_kwargs) + + def _post_and_get_call(self, write_key, **kwargs): + mock_response = mock.Mock() + mock_response.status_code = 200 + with mock.patch('customerio.analytics.request._session.post', + return_value=mock_response) as mock_post: + post(write_key, batch=[{ + 'userId': 'userId', + 'event': 'python event', + 'type': 'track', + }], **kwargs) + mock_post.assert_called_once() + return mock_post.call_args + + def test_secret_api_key_uses_bearer(self): + key = 'ak_us_' + 'a' * 32 + '_abcdef' + _, call_kwargs = self._post_and_get_call(key) + self.assertEqual(call_kwargs['headers']['Authorization'], + 'Bearer ' + key) + self.assertIsNone(call_kwargs['auth']) + + def test_public_api_key_uses_bearer(self): + key = 'wk_us_' + 'a' * 32 + '_abcdef' + _, call_kwargs = self._post_and_get_call(key) + self.assertEqual(call_kwargs['headers']['Authorization'], + 'Bearer ' + key) + self.assertIsNone(call_kwargs['auth']) + + def test_legacy_write_key_uses_basic(self): + call_args, call_kwargs = self._post_and_get_call('testsecret') + self.assertEqual(call_args[0], 'https://cdp.customer.io/v1/batch') + self.assertNotIn('Authorization', call_kwargs['headers']) + self.assertEqual(call_kwargs['auth'].username, 'testsecret') + self.assertEqual(call_kwargs['auth'].password, '') + + def test_eu_api_key_uses_eu_host(self): + key = 'ak_eu_' + 'a' * 32 + '_abcdef' + call_args, _ = self._post_and_get_call(key) + self.assertEqual(call_args[0], 'https://cdp-eu.customer.io/v1/batch') + + def test_us_api_key_uses_us_host(self): + key = 'ak_us_' + 'a' * 32 + '_abcdef' + call_args, _ = self._post_and_get_call(key) + self.assertEqual(call_args[0], 'https://cdp.customer.io/v1/batch') + + def test_explicit_host_wins_over_api_key(self): + key = 'ak_eu_' + 'a' * 32 + '_abcdef' + call_args, _ = self._post_and_get_call( + key, host='https://example.com') + self.assertEqual(call_args[0], 'https://example.com/v1/batch') From 0804c72596d1eacddb64afe22e570c8b8e959aca Mon Sep 17 00:00:00 2001 From: sydneycollins-cio <149523950+sydneycollins-cio@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:11:36 -0400 Subject: [PATCH 2/5] feat: rename API key prefixes to pk_/sk_ Read the region from any key shaped {xx}_{us|eu}_ and treat pk_ as the public key. Co-Authored-By: Claude Opus 5.5 --- customerio/analytics/request.py | 4 ++-- customerio/analytics/test/request.py | 10 +++++----- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/customerio/analytics/request.py b/customerio/analytics/request.py index 206d77b2..bddd484f 100644 --- a/customerio/analytics/request.py +++ b/customerio/analytics/request.py @@ -13,7 +13,7 @@ _session = sessions.Session() -_API_KEY_PREFIX = re.compile(r'^(ak|wk)_(us|eu)_') +_API_KEY_PREFIX = re.compile(r'^[a-z]{2}_(us|eu)_') def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs): @@ -23,7 +23,7 @@ def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs): body["sentAt"] = datetime.utcnow().replace(tzinfo=tzutc()).isoformat() api_key = _API_KEY_PREFIX.match(write_key) default_host = 'https://cdp.customer.io' - if api_key and api_key.group(2) == 'eu': + if api_key and api_key.group(1) == 'eu': default_host = 'https://cdp-eu.customer.io' url = remove_trailing_slash(host or default_host) + '/v1/batch' auth = None if api_key else HTTPBasicAuth(write_key, '') diff --git a/customerio/analytics/test/request.py b/customerio/analytics/test/request.py index 2658090f..4e2a7cc2 100644 --- a/customerio/analytics/test/request.py +++ b/customerio/analytics/test/request.py @@ -99,14 +99,14 @@ def _post_and_get_call(self, write_key, **kwargs): return mock_post.call_args def test_secret_api_key_uses_bearer(self): - key = 'ak_us_' + 'a' * 32 + '_abcdef' + key = 'sk_us_' + 'a' * 32 + '_abcdef' _, call_kwargs = self._post_and_get_call(key) self.assertEqual(call_kwargs['headers']['Authorization'], 'Bearer ' + key) self.assertIsNone(call_kwargs['auth']) def test_public_api_key_uses_bearer(self): - key = 'wk_us_' + 'a' * 32 + '_abcdef' + key = 'pk_us_' + 'a' * 32 + '_abcdef' _, call_kwargs = self._post_and_get_call(key) self.assertEqual(call_kwargs['headers']['Authorization'], 'Bearer ' + key) @@ -120,17 +120,17 @@ def test_legacy_write_key_uses_basic(self): self.assertEqual(call_kwargs['auth'].password, '') def test_eu_api_key_uses_eu_host(self): - key = 'ak_eu_' + 'a' * 32 + '_abcdef' + key = 'sk_eu_' + 'a' * 32 + '_abcdef' call_args, _ = self._post_and_get_call(key) self.assertEqual(call_args[0], 'https://cdp-eu.customer.io/v1/batch') def test_us_api_key_uses_us_host(self): - key = 'ak_us_' + 'a' * 32 + '_abcdef' + key = 'sk_us_' + 'a' * 32 + '_abcdef' call_args, _ = self._post_and_get_call(key) self.assertEqual(call_args[0], 'https://cdp.customer.io/v1/batch') def test_explicit_host_wins_over_api_key(self): - key = 'ak_eu_' + 'a' * 32 + '_abcdef' + key = 'sk_eu_' + 'a' * 32 + '_abcdef' call_args, _ = self._post_and_get_call( key, host='https://example.com') self.assertEqual(call_args[0], 'https://example.com/v1/batch') From a62bc8d61a8870136bdc6610212320fe82e99b65 Mon Sep 17 00:00:00 2001 From: sydneycollins-cio <149523950+sydneycollins-cio@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:29:54 -0400 Subject: [PATCH 3/5] Send the Bearer key through an explicit auth handler so .netrc cannot override it Co-Authored-By: Claude Sonnet 5.5 --- customerio/analytics/request.py | 18 +++++++++++++---- customerio/analytics/test/request.py | 29 +++++++++++++++++++++------- 2 files changed, 36 insertions(+), 11 deletions(-) diff --git a/customerio/analytics/request.py b/customerio/analytics/request.py index bddd484f..dc6a735b 100644 --- a/customerio/analytics/request.py +++ b/customerio/analytics/request.py @@ -5,7 +5,7 @@ import json import re from dateutil.tz import tzutc -from requests.auth import HTTPBasicAuth +from requests.auth import AuthBase, HTTPBasicAuth from requests import sessions from customerio.analytics.version import VERSION @@ -16,6 +16,18 @@ _API_KEY_PREFIX = re.compile(r'^[a-z]{2}_(us|eu)_') +class _BearerAuth(AuthBase): + """Sends the key as a Bearer token. Passed as `auth` so Requests never + swaps in Basic auth from a matching .netrc entry.""" + + def __init__(self, key): + self.key = key + + def __call__(self, request): + request.headers['Authorization'] = 'Bearer ' + self.key + return request + + def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs): """Post the `kwargs` to the API""" log = logging.getLogger('customerio') @@ -26,15 +38,13 @@ def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs): if api_key and api_key.group(1) == 'eu': default_host = 'https://cdp-eu.customer.io' url = remove_trailing_slash(host or default_host) + '/v1/batch' - auth = None if api_key else HTTPBasicAuth(write_key, '') + auth = _BearerAuth(write_key) if api_key else HTTPBasicAuth(write_key, '') data = json.dumps(body, cls=DatetimeSerializer) log.debug('making request: %s', data) headers = { 'Content-Type': 'application/json', 'User-Agent': 'analytics-python/' + VERSION } - if api_key: - headers['Authorization'] = 'Bearer ' + write_key if gzip: headers['Content-Encoding'] = 'gzip' buf = BytesIO() diff --git a/customerio/analytics/test/request.py b/customerio/analytics/test/request.py index 4e2a7cc2..97b7d9e8 100644 --- a/customerio/analytics/test/request.py +++ b/customerio/analytics/test/request.py @@ -2,9 +2,11 @@ import unittest import unittest.mock as mock import json +import os +import tempfile import requests -from customerio.analytics.request import post, DatetimeSerializer +from customerio.analytics.request import post, DatetimeSerializer, _BearerAuth class TestRequests(unittest.TestCase): @@ -101,16 +103,29 @@ def _post_and_get_call(self, write_key, **kwargs): def test_secret_api_key_uses_bearer(self): key = 'sk_us_' + 'a' * 32 + '_abcdef' _, call_kwargs = self._post_and_get_call(key) - self.assertEqual(call_kwargs['headers']['Authorization'], - 'Bearer ' + key) - self.assertIsNone(call_kwargs['auth']) + self.assertIsInstance(call_kwargs['auth'], _BearerAuth) + self.assertEqual(call_kwargs['auth'].key, key) def test_public_api_key_uses_bearer(self): key = 'pk_us_' + 'a' * 32 + '_abcdef' _, call_kwargs = self._post_and_get_call(key) - self.assertEqual(call_kwargs['headers']['Authorization'], - 'Bearer ' + key) - self.assertIsNone(call_kwargs['auth']) + self.assertIsInstance(call_kwargs['auth'], _BearerAuth) + self.assertEqual(call_kwargs['auth'].key, key) + + def test_netrc_entry_does_not_override_bearer(self): + key = 'sk_us_' + 'a' * 32 + '_abcdef' + with tempfile.TemporaryDirectory() as tmp: + netrc = os.path.join(tmp, 'netrc') + with open(netrc, 'w') as f: + f.write('machine cdp.customer.io login user password pass\n') + mock_response = mock.Mock() + mock_response.status_code = 200 + with mock.patch.dict(os.environ, {'NETRC': netrc}), \ + mock.patch('requests.sessions.Session.send', + return_value=mock_response) as mock_send: + post(key, batch=[{'userId': 'userId', 'type': 'track'}]) + prepared = mock_send.call_args[0][0] + self.assertEqual(prepared.headers['Authorization'], 'Bearer ' + key) def test_legacy_write_key_uses_basic(self): call_args, call_kwargs = self._post_and_get_call('testsecret') From 60cdae9a0b50193c416a7e38e21924538c578cac Mon Sep 17 00:00:00 2001 From: sydneycollins-cio <149523950+sydneycollins-cio@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:27:54 -0400 Subject: [PATCH 4/5] Strip the API key before use and warn when a region key goes to the other region's host Co-Authored-By: Claude Opus 5.5 --- customerio/analytics/request.py | 19 ++++++++++++++---- customerio/analytics/test/request.py | 30 ++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 4 deletions(-) diff --git a/customerio/analytics/request.py b/customerio/analytics/request.py index dc6a735b..8e7c0bb2 100644 --- a/customerio/analytics/request.py +++ b/customerio/analytics/request.py @@ -14,6 +14,10 @@ _session = sessions.Session() _API_KEY_PREFIX = re.compile(r'^[a-z]{2}_(us|eu)_') +_REGION_HOSTS = { + 'us': 'https://cdp.customer.io', + 'eu': 'https://cdp-eu.customer.io', +} class _BearerAuth(AuthBase): @@ -33,11 +37,18 @@ def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs): log = logging.getLogger('customerio') body = kwargs body["sentAt"] = datetime.utcnow().replace(tzinfo=tzutc()).isoformat() + # Keys read from mounted secret files often end in a newline, which + # http.client rejects with an error that includes the whole key. + write_key = write_key.strip() api_key = _API_KEY_PREFIX.match(write_key) - default_host = 'https://cdp.customer.io' - if api_key and api_key.group(1) == 'eu': - default_host = 'https://cdp-eu.customer.io' - url = remove_trailing_slash(host or default_host) + '/v1/batch' + region = api_key.group(1) if api_key else 'us' + base_url = remove_trailing_slash(host or _REGION_HOSTS[region]) + if api_key: + for other_region, other_host in _REGION_HOSTS.items(): + if other_region != region and base_url == other_host: + log.warning('%s API key is being sent to the %s host %s', + region.upper(), other_region.upper(), base_url) + url = base_url + '/v1/batch' auth = _BearerAuth(write_key) if api_key else HTTPBasicAuth(write_key, '') data = json.dumps(body, cls=DatetimeSerializer) log.debug('making request: %s', data) diff --git a/customerio/analytics/test/request.py b/customerio/analytics/test/request.py index 97b7d9e8..34ed8666 100644 --- a/customerio/analytics/test/request.py +++ b/customerio/analytics/test/request.py @@ -2,6 +2,7 @@ import unittest import unittest.mock as mock import json +import logging import os import tempfile import requests @@ -149,3 +150,32 @@ def test_explicit_host_wins_over_api_key(self): call_args, _ = self._post_and_get_call( key, host='https://example.com') self.assertEqual(call_args[0], 'https://example.com/v1/batch') + + def test_api_key_with_trailing_newline_is_stripped(self): + key = 'sk_eu_' + 'a' * 32 + '_abcdef' + mock_response = mock.Mock() + mock_response.status_code = 200 + with mock.patch('requests.sessions.Session.send', + return_value=mock_response) as mock_send: + post(key + '\n', batch=[{'userId': 'userId', 'type': 'track'}]) + prepared = mock_send.call_args[0][0] + self.assertEqual(prepared.url, 'https://cdp-eu.customer.io/v1/batch') + self.assertEqual(prepared.headers['Authorization'], 'Bearer ' + key) + + def test_api_key_sent_to_other_region_host_warns(self): + key = 'sk_eu_' + 'a' * 32 + '_abcdef' + with self.assertLogs('customerio', level='WARNING') as logs: + call_args, _ = self._post_and_get_call( + key, host='https://cdp.customer.io/') + self.assertEqual(call_args[0], 'https://cdp.customer.io/v1/batch') + self.assertEqual(logs.output, [ + 'WARNING:customerio:EU API key is being sent to the US host ' + 'https://cdp.customer.io', + ]) + + def test_api_key_sent_to_own_region_host_does_not_warn(self): + key = 'sk_eu_' + 'a' * 32 + '_abcdef' + log = logging.getLogger('customerio') + with mock.patch.object(log, 'warning') as mock_warning: + self._post_and_get_call(key, host='https://cdp-eu.customer.io') + mock_warning.assert_not_called() From 2a3dd09036d3c1021d5d999693785cc9c2330488 Mon Sep 17 00:00:00 2001 From: sydneycollins-cio <149523950+sydneycollins-cio@users.noreply.github.com> Date: Tue, 20 Oct 2026 11:46:53 -0400 Subject: [PATCH 5/5] Match only pk_/sk_ key prefixes and warn about a wrong-region host once per client Co-Authored-By: Claude Opus 5.5 --- customerio/analytics/client.py | 5 +++- customerio/analytics/request.py | 31 ++++++++++++++------ customerio/analytics/test/client.py | 42 ++++++++++++++++++++++++++++ customerio/analytics/test/request.py | 23 ++++----------- 4 files changed, 74 insertions(+), 27 deletions(-) diff --git a/customerio/analytics/client.py b/customerio/analytics/client.py index c5387fb3..bf570c5e 100644 --- a/customerio/analytics/client.py +++ b/customerio/analytics/client.py @@ -9,7 +9,8 @@ from dateutil.tz import tzutc from customerio.analytics.consumer import Consumer, MAX_MSG_SIZE -from customerio.analytics.request import post, DatetimeSerializer +from customerio.analytics.request import ( + post, DatetimeSerializer, warn_if_other_region_host) from customerio.analytics.utils import guess_timezone, clean from customerio.analytics.version import VERSION @@ -66,6 +67,8 @@ def __init__(self, if debug: self.log.setLevel(logging.DEBUG) + warn_if_other_region_host(write_key, host) + if sync_mode: self.consumers = None else: diff --git a/customerio/analytics/request.py b/customerio/analytics/request.py index 8e7c0bb2..434fd85d 100644 --- a/customerio/analytics/request.py +++ b/customerio/analytics/request.py @@ -13,7 +13,7 @@ _session = sessions.Session() -_API_KEY_PREFIX = re.compile(r'^[a-z]{2}_(us|eu)_') +_API_KEY_PREFIX = re.compile(r'^(pk|sk)_(us|eu)_') _REGION_HOSTS = { 'us': 'https://cdp.customer.io', 'eu': 'https://cdp-eu.customer.io', @@ -32,6 +32,26 @@ def __call__(self, request): return request +def _base_url(api_key, host): + region = api_key.group(2) if api_key else 'us' + return remove_trailing_slash(host or _REGION_HOSTS[region]) + + +def warn_if_other_region_host(write_key, host): + """Logs a warning when a region API key is set to go to the other + region's default host. Called once per client, not per request.""" + api_key = _API_KEY_PREFIX.match(write_key.strip()) + if not api_key: + return + region = api_key.group(2) + base_url = _base_url(api_key, host) + for other_region, other_host in _REGION_HOSTS.items(): + if other_region != region and base_url == other_host: + logging.getLogger('customerio').warning( + '%s API key is being sent to the %s host %s', + region.upper(), other_region.upper(), base_url) + + def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs): """Post the `kwargs` to the API""" log = logging.getLogger('customerio') @@ -41,14 +61,7 @@ def post(write_key, host=None, gzip=False, timeout=15, proxies=None, **kwargs): # http.client rejects with an error that includes the whole key. write_key = write_key.strip() api_key = _API_KEY_PREFIX.match(write_key) - region = api_key.group(1) if api_key else 'us' - base_url = remove_trailing_slash(host or _REGION_HOSTS[region]) - if api_key: - for other_region, other_host in _REGION_HOSTS.items(): - if other_region != region and base_url == other_host: - log.warning('%s API key is being sent to the %s host %s', - region.upper(), other_region.upper(), base_url) - url = base_url + '/v1/batch' + url = _base_url(api_key, host) + '/v1/batch' auth = _BearerAuth(write_key) if api_key else HTTPBasicAuth(write_key, '') data = json.dumps(body, cls=DatetimeSerializer) log.debug('making request: %s', data) diff --git a/customerio/analytics/test/client.py b/customerio/analytics/test/client.py index 4c10be30..b27c5be2 100644 --- a/customerio/analytics/test/client.py +++ b/customerio/analytics/test/client.py @@ -1,6 +1,7 @@ from datetime import date, datetime import unittest import unittest.mock as mock +import logging import time from customerio.analytics.version import VERSION @@ -368,3 +369,44 @@ def test_proxies(self): mock_post.assert_called_once() _, call_kwargs = mock_post.call_args self.assertEqual(call_kwargs['proxies'], proxies) + + def _track_with_mocked_post(self, client, times=1): + mock_response = mock.Mock() + mock_response.status_code = 200 + with mock.patch('customerio.analytics.request._session.post', + return_value=mock_response) as mock_post: + for _ in range(times): + client.track('userId', 'python event') + return mock_post.call_args + + def test_api_key_sent_to_other_region_host_warns(self): + key = 'sk_eu_' + 'a' * 32 + '_abcdef' + with self.assertLogs('customerio', level='WARNING') as logs: + client = Client(key, host='https://cdp.customer.io/', + sync_mode=True) + call_args = self._track_with_mocked_post(client) + self.assertEqual(call_args[0][0], 'https://cdp.customer.io/v1/batch') + self.assertEqual(logs.output, [ + 'WARNING:customerio:EU API key is being sent to the US host ' + 'https://cdp.customer.io', + ]) + + def test_other_region_host_warns_once_per_client(self): + key = 'sk_eu_' + 'a' * 32 + '_abcdef' + with self.assertLogs('customerio', level='WARNING') as logs: + client = Client(key, host='https://cdp.customer.io', + sync_mode=True) + self._track_with_mocked_post(client, times=3) + self.assertEqual(logs.output, [ + 'WARNING:customerio:EU API key is being sent to the US host ' + 'https://cdp.customer.io', + ]) + + def test_api_key_sent_to_own_region_host_does_not_warn(self): + key = 'sk_eu_' + 'a' * 32 + '_abcdef' + log = logging.getLogger('customerio') + with mock.patch.object(log, 'warning') as mock_warning: + client = Client(key, host='https://cdp-eu.customer.io', + sync_mode=True) + self._track_with_mocked_post(client) + mock_warning.assert_not_called() diff --git a/customerio/analytics/test/request.py b/customerio/analytics/test/request.py index 34ed8666..077e678b 100644 --- a/customerio/analytics/test/request.py +++ b/customerio/analytics/test/request.py @@ -2,7 +2,6 @@ import unittest import unittest.mock as mock import json -import logging import os import tempfile import requests @@ -162,20 +161,10 @@ def test_api_key_with_trailing_newline_is_stripped(self): self.assertEqual(prepared.url, 'https://cdp-eu.customer.io/v1/batch') self.assertEqual(prepared.headers['Authorization'], 'Bearer ' + key) - def test_api_key_sent_to_other_region_host_warns(self): - key = 'sk_eu_' + 'a' * 32 + '_abcdef' - with self.assertLogs('customerio', level='WARNING') as logs: - call_args, _ = self._post_and_get_call( - key, host='https://cdp.customer.io/') + def test_unknown_prefix_uses_basic_and_us_host(self): + key = 'xx_us_' + 'a' * 32 + '_abcdef' + call_args, call_kwargs = self._post_and_get_call(key) self.assertEqual(call_args[0], 'https://cdp.customer.io/v1/batch') - self.assertEqual(logs.output, [ - 'WARNING:customerio:EU API key is being sent to the US host ' - 'https://cdp.customer.io', - ]) - - def test_api_key_sent_to_own_region_host_does_not_warn(self): - key = 'sk_eu_' + 'a' * 32 + '_abcdef' - log = logging.getLogger('customerio') - with mock.patch.object(log, 'warning') as mock_warning: - self._post_and_get_call(key, host='https://cdp-eu.customer.io') - mock_warning.assert_not_called() + self.assertNotIsInstance(call_kwargs['auth'], _BearerAuth) + self.assertEqual(call_kwargs['auth'].username, key) + self.assertEqual(call_kwargs['auth'].password, '')