From 0b53bc2025ec5ff2b5cb9608c86c169784df53a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jozef=20Bi=CC=81ros=CC=8C?= Date: Thu, 17 Sep 2026 16:44:28 +0200 Subject: [PATCH 1/5] chore: ignore .venv and drop obsolete compose version key - .gitignore: local virtualenvs are non-portable (absolute interpreter paths, platform-specific native extensions), so they must never be committed. - docker-compose.yml: Compose V2 ignores the top-level `version` key and warns on every invocation. Both `docker compose` and `docker-compose` resolve to V2 locally, and CI only uses `docker compose`. Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 1 + docker-compose.yml | 1 - 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 0460fd8..ed476a8 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ __pycache__/ /.coverage /kbcstorage.egg-info /dist +.venv/ diff --git a/docker-compose.yml b/docker-compose.yml index 91d50c4..8c7eacd 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,3 @@ -version: '3' services: ci: &ci build: . From 6dfcab4c66a937c8b5137709ef40e1e572c7db49 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jozef=20Bi=CC=81ros=CC=8C?= Date: Thu, 17 Sep 2026 17:02:32 +0200 Subject: [PATCH 2/5] docs: document local development setup Adds a Development section covering virtualenv creation, dependency installation and editor interpreter selection, written to apply to any editor rather than a specific one. Supporting changes the new instructions require: - pyproject.toml: declare flake8 in a `dev` extra. It was previously installed ad hoc by the Dockerfile and by nothing else, so local and CI versions were unconstrained and free to diverge. Bounded to 7.x because flake8 majors add checks, which fails CI on untouched code. - Dockerfile: install `.[dev]` instead of flake8 and the package separately. - .flake8: exclude .venv. flake8's default exclude list does not cover it, so a project-root virtualenv makes it lint site-packages. Uses extend-exclude to keep the defaults. - .dockerignore: exclude .venv, which `COPY . /code/` would otherwise copy into the image. - README: the Tests command named the image, `sapi-python-client`, rather than a service, and failed with "no such service". Corrected to `ci`. Co-Authored-By: Claude Opus 5 (1M context) --- .dockerignore | 1 + .flake8 | 1 + Dockerfile | 3 +-- README.md | 64 +++++++++++++++++++++++++++++++++++++++++++++++++- pyproject.toml | 5 ++++ 5 files changed, 71 insertions(+), 3 deletions(-) diff --git a/.dockerignore b/.dockerignore index 338a1d6..646aef3 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,3 +4,4 @@ docker-compose.yml Dockerfile README.md +.venv/ diff --git a/.flake8 b/.flake8 index 6deafc2..ef264a0 100644 --- a/.flake8 +++ b/.flake8 @@ -1,2 +1,3 @@ [flake8] max-line-length = 120 +extend-exclude = .venv diff --git a/Dockerfile b/Dockerfile index 4d61e65..7ee8c5b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,6 +3,5 @@ FROM python:3.11 WORKDIR /code COPY . /code/ RUN python -m pip install --upgrade pip -RUN pip install --no-cache-dir flake8 -RUN pip install --no-cache-dir . +RUN pip install --no-cache-dir ".[dev]" ENTRYPOINT ["python"] diff --git a/README.md b/README.md index 969c176..eaa89b1 100644 --- a/README.md +++ b/README.md @@ -64,11 +64,73 @@ tables.detail('in.c-demo.some-table') ``` +## Development + +### Local environment + +The package requires Python 3.7 or newer; CI builds and tests on 3.11. + +Create a virtual environment and install the package into it in editable mode: + +```bash +python3 -m venv .venv +.venv/bin/pip install -e ".[dev]" +``` + +`pyproject.toml` declares the test dependencies alongside the runtime ones, and the `dev` extra adds +the linter, so this single command covers everything. The `-e` flag keeps `kbcstorage` resolving to +the working tree, so edits take effect without reinstalling. + +Prefixing commands with `.venv/bin/` always uses the environment. Activating it instead makes the +plain `python` and `pip` commands resolve there for the rest of the shell session: + +```bash +source .venv/bin/activate +``` + +On Windows the paths are `.venv\Scripts\python` and `.venv\Scripts\activate`. + +### Editor setup + +Editors and IDEs resolve imports through a configured interpreter. Until yours is pointed at the +virtual environment, every third-party import is reported as unresolved. The interpreter path is: + +``` +.venv/bin/python +``` + +| Editor | Where to set it | +| --- | --- | +| VS Code | `Python: Select Interpreter` | +| PyCharm | Settings → Project → Python Interpreter → Add → Existing environment | +| Zed | `toolchain: select` | +| Vim, Neovim, Emacs | the interpreter or `venv` option of your LSP client | + +Most editors scan for a directory named `.venv` or `venv` in the project root, so keeping the default +name avoids manual configuration in the common case. A restart of the language server is often +needed before existing errors clear. + +### Checks + +Lint. CI runs this before the tests, so a failure stops the build before anything is executed: + +```bash +.venv/bin/python -m flake8 +``` + +Unit tests, which mock all API responses and need no credentials: + +```bash +.venv/bin/python -m unittest discover -s tests/mocks -t . +``` + +The functional tests run against a real project and are covered below. + ## Tests Create `.env` file according to the `.env.template` file and run the tests with: ```bash -$ docker compose run --rm -e KBC_TEST_TOKEN -e KBC_TEST_API_URL sapi-python-client -m unittest discover +$ docker compose run --rm -e KBC_TEST_TOKEN -e KBC_TEST_API_URL ci -m unittest discover ``` ## Contribution Guide diff --git a/pyproject.toml b/pyproject.toml index 0d5b328..91b016e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,5 +32,10 @@ dependencies = [ ] dynamic = ["version"] +[project.optional-dependencies] +dev = [ + "flake8>=7,<8", +] + [tool.setuptools-git-versioning] enabled = true From 73f784e783b9f941823ccfbdcd8a5e7c3760390f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jozef=20Bi=CC=81ros=CC=8C?= Date: Thu, 17 Sep 2026 17:06:47 +0200 Subject: [PATCH 3/5] feat: authenticate with Authorization: Bearer for programmatic tokens Storage API tokens are project-bound and go in X-StorageApi-Token. Programmatic tokens (kbc_at_* session, kbc_pat_* personal access) are not, so they are sent as Authorization: Bearer alongside the project id in X-KBC-ProjectId. `kbcstorage.auth` holds the two strategies and `coerce`, which turns the `token` argument into a strategy object. A bare string still becomes a StorageApiToken, so every existing caller is unaffected: Client(url, 'token') # unchanged Client(url, BearerToken('kbc_at_...', 1234)) # bearer Endpoint now stores the strategy on `self.auth` and builds its header block from `auth.headers()`. Endpoints constructed internally by other endpoints are passed `self.auth` rather than the raw token, so the scheme propagates through the multi-step paths in tables, workspaces and buckets. `self.token` still exposes the raw string. Buckets.list_tables no longer hardcodes X-StorageApi-Token, which would have sent both schemes at once under a bearer token. The four request verbs merge caller headers into a new dict instead of mutating the caller's, which the previous `headers.update()` did. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 15 +++++++ kbcstorage/auth.py | 46 ++++++++++++++++++++++ kbcstorage/base.py | 26 +++++++------ kbcstorage/buckets.py | 6 +-- kbcstorage/client.py | 36 +++++++++++------ kbcstorage/tables.py | 14 +++---- kbcstorage/workspaces.py | 6 +-- tests/mocks/test_auth.py | 48 +++++++++++++++++++++++ tests/mocks/test_base.py | 71 ++++++++++++++++++++++++++++++++++ tests/mocks/test_buckets.py | 21 ++++++++++ tests/mocks/test_client.py | 41 +++++++++++++++++++- tests/mocks/test_workspaces.py | 29 ++++++++++++++ 12 files changed, 320 insertions(+), 39 deletions(-) create mode 100644 kbcstorage/auth.py create mode 100644 tests/mocks/test_auth.py create mode 100644 tests/mocks/test_base.py diff --git a/README.md b/README.md index eaa89b1..8cd122a 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,21 @@ client.tables.detail('in.c-demo.some-table') ``` +## Authentication + +By default the token is a Storage API token, sent as `X-StorageApi-Token`. Programmatic tokens +(`kbc_at_*` session tokens, `kbc_pat_*` personal access tokens) are not bound to a project, so they +are sent as `Authorization: Bearer` together with the project id in `X-KBC-ProjectId`: + +```python +from kbcstorage.client import Client, BearerToken + +client = Client('https://connection.keboola.com', BearerToken('kbc_at_...', project_id=1234)) +``` + +Both forms work for directly constructed endpoints too, and the token is used as given — the client +never refreshes or decodes it, so an expired token surfaces as a `requests.HTTPError` with a 401. + ## Endpoint Classes Usage ```python from kbcstorage.tables import Tables diff --git a/kbcstorage/auth.py b/kbcstorage/auth.py new file mode 100644 index 0000000..c63f24a --- /dev/null +++ b/kbcstorage/auth.py @@ -0,0 +1,46 @@ +class StorageApiToken: + """ + Legacy Storage API token, sent as ``X-StorageApi-Token``. + """ + def __init__(self, token): + if not token: + raise ValueError("Token is required.") + self.token = token + + def headers(self): + return {'X-StorageApi-Token': self.token} + + +class BearerToken: + """ + Programmatic session or personal access token (``kbc_at_*``, ``kbc_pat_*``). + """ + def __init__(self, token, project_id): + if not token: + raise ValueError("Token is required.") + if not project_id: + raise ValueError("Project ID is required for a bearer token.") + self.token = token + # requests rejects non-string header values, and callers naturally pass + # the numeric id straight from a token verify payload + self.project_id = str(project_id) + + def headers(self): + return {'Authorization': 'Bearer {}'.format(self.token), + 'X-KBC-ProjectId': self.project_id} + + +def coerce(token): + """ + Normalise the ``token`` argument of an endpoint into a strategy object. + + Args: + token: A strategy object, or a bare token string for backward + compatibility. + + Returns: + The strategy object to authenticate with. + """ + if hasattr(token, 'headers'): + return token + return StorageApiToken(token) diff --git a/kbcstorage/base.py b/kbcstorage/base.py index baccb02..04631ab 100644 --- a/kbcstorage/base.py +++ b/kbcstorage/base.py @@ -11,6 +11,7 @@ """ import os +from kbcstorage.auth import coerce from kbcstorage.retry_requests import MAX_RETRIES_DEFAULT, RetryRequests import requests @@ -23,6 +24,9 @@ class Endpoint: Attributes: base_url (str): The base URL for this endpoint. token (str): A key for the Storage API. + auth (kbcstorage.auth.StorageApiToken|kbcstorage.auth.BearerToken): The + authentication strategy to pass on to endpoints created from this + one. """ def __init__(self, root_url, path_component, token, max_requests_retries=MAX_RETRIES_DEFAULT): """ @@ -33,8 +37,9 @@ def __init__(self, root_url, path_component, token, max_requests_retries=MAX_RET "https://connection.keboola.com/" path_component (str): The section of the path specific to the endpoint. eg. "buckets" - token (str): A key for the Storage API. Can be found in the storage - console. + token (str|kbcstorage.auth.StorageApiToken|kbcstorage.auth.BearerToken): + A key for the Storage API, found in the storage console, or an + authentication strategy from `kbcstorage.auth`. """ if not root_url: raise ValueError("Root URL is required.") @@ -43,8 +48,9 @@ def __init__(self, root_url, path_component, token, max_requests_retries=MAX_RET self.root_url = root_url self.base_url = '{}/v2/storage/{}'.format(root_url.strip('/'), path_component.strip('/')) - self.token = token - self._auth_header = {'X-StorageApi-Token': self.token, + self.auth = coerce(token) + self.token = self.auth.token + self._auth_header = {**self.auth.headers(), 'X-KBC-RunId': os.environ.get('KBC_RUNID'), 'Accept-Encoding': 'gzip', 'User-Agent': 'Keboola Storage API Python Client'} @@ -68,8 +74,7 @@ def _get_raw(self, url, params=None, **kwargs): Raises: requests.HTTPError: If the API request fails. """ - headers = kwargs.pop('headers', {}) - headers.update(self._auth_header) + headers = {**kwargs.pop('headers', {}), **self._auth_header} r = self.requests.get(url, params=params, headers=headers, **kwargs) try: @@ -114,8 +119,7 @@ def _post(self, *args, **kwargs): Raises: requests.HTTPError: If the API request fails. """ - headers = kwargs.pop('headers', {}) - headers.update(self._auth_header) + headers = {**kwargs.pop('headers', {}), **self._auth_header} r = self.requests.post(headers=headers, *args, **kwargs) try: r.raise_for_status() @@ -140,8 +144,7 @@ def _put(self, *args, **kwargs): Raises: requests.HTTPError: If the API request fails. """ - headers = kwargs.pop('headers', {}) - headers.update(self._auth_header) + headers = {**kwargs.pop('headers', {}), **self._auth_header} r = self.requests.put(headers=headers, *args, **kwargs) try: r.raise_for_status() @@ -166,8 +169,7 @@ def _delete(self, *args, **kwargs): Raises: requests.HTTPError: If the API request fails. """ - headers = kwargs.pop('headers', {}) - headers.update(self._auth_header) + headers = {**kwargs.pop('headers', {}), **self._auth_header} r = self.requests.delete(headers=headers, *args, **kwargs) try: r.raise_for_status() diff --git a/kbcstorage/buckets.py b/kbcstorage/buckets.py index 6146229..4ff0690 100644 --- a/kbcstorage/buckets.py +++ b/kbcstorage/buckets.py @@ -50,13 +50,11 @@ def list_tables(self, bucket_id, include=None): Raises: requests.HTTPError: If the API request fails. """ - headers = {'X-StorageApi-Token': self.token} - url = '{}/{}/tables'.format(self.base_url, bucket_id) params = {} if include is not None and isinstance(include, list): params['include'] = ','.join(include) - return self._get(url, headers=headers, params=params) + return self._get(url, params=params) def detail(self, bucket_id): """ @@ -121,7 +119,7 @@ def delete(self, bucket_id, force=False, asynchronous=True): params = {'force': force, 'async': asynchronous} if (asynchronous): job = self._delete(url, params=params) - jobs = Jobs(self.root_url, self.token) + jobs = Jobs(self.root_url, self.auth) job = jobs.block_until_completed(job['id']) if job['status'] == 'error': raise RuntimeError(job['error']['message']) diff --git a/kbcstorage/client.py b/kbcstorage/client.py index 0b0a7a3..7477839 100644 --- a/kbcstorage/client.py +++ b/kbcstorage/client.py @@ -1,6 +1,7 @@ """" Entry point for the Storage API client. """ +from kbcstorage.auth import BearerToken, StorageApiToken, coerce # noqa: F401 from kbcstorage.branches import Branches from kbcstorage.buckets import Buckets from kbcstorage.components import Components @@ -24,33 +25,44 @@ def __init__(self, api_domain, token, branch_id='default', file_storage_support= Args: api_domain (str): The domain on which the API sits. eg. "https://connection.keboola.com". - token (str): A storage API key. + token (str|StorageApiToken|BearerToken): A storage API key, or an + authentication strategy from `kbcstorage.auth`. branch_id (str): The ID of branch to use, use 'default' to work without branch (in main). file_storage_support (bool): If False, it saves memory by not importing libraries for all storage backends. """ self.root_url = api_domain.rstrip("/") - self._token = token + self._auth = coerce(token) + self._token = self._auth.token self._branch_id = branch_id - self.buckets = Buckets(self.root_url, self.token) + self.buckets = Buckets(self.root_url, self.auth) if file_storage_support: from kbcstorage.files import Files - self.files = Files(self.root_url, self.token) + self.files = Files(self.root_url, self.auth) - self.jobs = Jobs(self.root_url, self.token) - self.tables = Tables(self.root_url, self.token) - self.workspaces = Workspaces(self.root_url, self.token) - self.components = Components(self.root_url, self.token, self.branch_id) - self.configurations = Configurations(self.root_url, self.token, self.branch_id) - self.tokens = Tokens(self.root_url, self.token) - self.branches = Branches(self.root_url, self.token) - self.triggers = Triggers(self.root_url, self.token) + self.jobs = Jobs(self.root_url, self.auth) + self.tables = Tables(self.root_url, self.auth) + self.workspaces = Workspaces(self.root_url, self.auth) + self.components = Components(self.root_url, self.auth, self.branch_id) + self.configurations = Configurations(self.root_url, self.auth, self.branch_id) + self.tokens = Tokens(self.root_url, self.auth) + self.branches = Branches(self.root_url, self.auth) + self.triggers = Triggers(self.root_url, self.auth) @property def token(self): return self._token + @property + def auth(self): + return self._auth + + @property + def project_id(self): + """The project the requests are scoped to, or None for a storage API token.""" + return getattr(self._auth, 'project_id', None) + @property def branch_id(self): return self._branch_id diff --git a/kbcstorage/tables.py b/kbcstorage/tables.py index 5285fb1..c65da2f 100644 --- a/kbcstorage/tables.py +++ b/kbcstorage/tables.py @@ -91,14 +91,14 @@ def create(self, bucket_id, name, file_path, delimiter=',', enclosure='"', Raises: requests.HTTPError: If the API request fails. """ - files = Files(self.root_url, self.token) + files = Files(self.root_url, self.auth) file_id = files.upload_file(file_path=file_path, tags=['file-import'], do_notify=False, is_public=False) job = self.create_raw(bucket_id=bucket_id, name=name, data_file_id=file_id, delimiter=delimiter, enclosure=enclosure, escaped_by=escaped_by, primary_key=primary_key) - jobs = Jobs(self.root_url, self.token) + jobs = Jobs(self.root_url, self.auth) job = jobs.block_until_completed(job['id']) if job['status'] == 'error': raise RuntimeError(job['error']['message']) @@ -229,7 +229,7 @@ def load(self, table_id, file_path, is_incremental=False, delimiter=',', Raises: requests.HTTPError: If the API request fails. """ - files = Files(self.root_url, self.token) + files = Files(self.root_url, self.auth) file_id = files.upload_file(file_path=file_path, tags=['file-import'], do_notify=False, is_public=False) job = self.load_raw(table_id=table_id, data_file_id=file_id, @@ -237,7 +237,7 @@ def load(self, table_id, file_path, is_incremental=False, delimiter=',', escaped_by=escaped_by, is_incremental=is_incremental, columns=columns, without_headers=without_headers) - jobs = Jobs(self.root_url, self.token) + jobs = Jobs(self.root_url, self.auth) job = jobs.block_until_completed(job['id']) if job['status'] == 'error': raise RuntimeError(job['error']['message']) @@ -399,11 +399,11 @@ def export_to_file(self, table_id, path_name, limit=None, where_column=where_column, where_values=where_values, where_operator=where_operator, is_gzip=is_gzip) - jobs = Jobs(self.root_url, self.token) + jobs = Jobs(self.root_url, self.auth) job = jobs.block_until_completed(job['id']) if job['status'] == 'error': raise RuntimeError(job['error']['message']) - files = Files(self.root_url, self.token) + files = Files(self.root_url, self.auth) temp_path = tempfile.TemporaryDirectory() local_file = files.download(file_id=job['results']['file']['id'], local_path=temp_path.name) @@ -467,7 +467,7 @@ def export(self, table_id, limit=None, file_format='rfc', where_column=where_column, where_values=where_values, where_operator=where_operator, is_gzip=is_gzip) - jobs = Jobs(self.root_url, self.token) + jobs = Jobs(self.root_url, self.auth) job = jobs.block_until_completed(job['id']) if job['status'] == 'error': raise RuntimeError(job['error']['message']) diff --git a/kbcstorage/workspaces.py b/kbcstorage/workspaces.py index 83d4643..ea9b7e8 100644 --- a/kbcstorage/workspaces.py +++ b/kbcstorage/workspaces.py @@ -181,7 +181,7 @@ def _get_default_backend(self): effectively immutable for the client's lifetime. """ if self._default_backend is _DEFAULT_BACKEND_UNRESOLVED: - token_info = Tokens(self.root_url, self.token).verify() + token_info = Tokens(self.root_url, self.auth).verify() self._default_backend = (token_info.get('owner') or {}).get('defaultBackend') return self._default_backend @@ -285,14 +285,14 @@ def load_files(self, workspace_id, file_mapping): workspace = self.detail(workspace_id) if (workspace['type'] != 'file' and workspace['connection']['backend'] != 'abs'): raise Exception('Loading files to workspace is only available for ABS workspaces') - files = Files(self.root_url, self.token) + files = Files(self.root_url, self.auth) if ('operator' in file_mapping and file_mapping['operator'] == 'and'): query = ' AND '.join(map(lambda tag: 'tags:"' + tag + '"', file_mapping['tags'])) file_list = files.list(q=query) else: file_list = files.list(tags=file_mapping['tags']) - jobs = Jobs(self.root_url, self.token) + jobs = Jobs(self.root_url, self.auth) jobs_list = [] for file in file_list: inputs = { diff --git a/tests/mocks/test_auth.py b/tests/mocks/test_auth.py new file mode 100644 index 0000000..6dc5b14 --- /dev/null +++ b/tests/mocks/test_auth.py @@ -0,0 +1,48 @@ +import unittest + +from kbcstorage.auth import BearerToken, StorageApiToken, coerce + + +class TestStorageApiToken(unittest.TestCase): + def test_headers(self): + self.assertEqual({'X-StorageApi-Token': 'dummy_token'}, + StorageApiToken('dummy_token').headers()) + + def test_empty_token(self): + with self.assertRaisesRegex(ValueError, 'Token is required.'): + StorageApiToken('') + + +class TestBearerToken(unittest.TestCase): + def test_headers(self): + self.assertEqual({'Authorization': 'Bearer kbc_at_dummy', + 'X-KBC-ProjectId': '1234'}, + BearerToken('kbc_at_dummy', '1234').headers()) + + def test_numeric_project_id_is_stringified(self): + token = BearerToken('kbc_at_dummy', 1234) + self.assertEqual('1234', token.project_id) + self.assertEqual('1234', token.headers()['X-KBC-ProjectId']) + + def test_empty_token(self): + with self.assertRaisesRegex(ValueError, 'Token is required.'): + BearerToken('', 1234) + + def test_missing_project_id(self): + with self.assertRaisesRegex(ValueError, 'Project ID is required'): + BearerToken('kbc_at_dummy', None) + + +class TestCoerce(unittest.TestCase): + def test_bare_string_becomes_storage_api_token(self): + auth = coerce('dummy_token') + self.assertIsInstance(auth, StorageApiToken) + self.assertEqual('dummy_token', auth.token) + + def test_strategy_passes_through(self): + auth = BearerToken('kbc_at_dummy', 1234) + self.assertIs(auth, coerce(auth)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/mocks/test_base.py b/tests/mocks/test_base.py new file mode 100644 index 0000000..2ca02bf --- /dev/null +++ b/tests/mocks/test_base.py @@ -0,0 +1,71 @@ +import unittest + +import responses + +from kbcstorage.auth import BearerToken +from kbcstorage.base import Endpoint + +BASE_URL = 'https://connection.keboola.com/' +URL = 'https://connection.keboola.com/v2/storage/dummy' + + +def _mock_get(): + responses.add(responses.Response(method='GET', url=URL, json={})) + + +class TestEndpointAuthHeaders(unittest.TestCase): + @responses.activate + def test_storage_api_token_headers(self): + _mock_get() + Endpoint(BASE_URL, 'dummy', 'dummy_token')._get(URL) + + headers = responses.calls[0].request.headers + self.assertEqual('dummy_token', headers['X-StorageApi-Token']) + self.assertNotIn('Authorization', headers) + self.assertNotIn('X-KBC-ProjectId', headers) + + @responses.activate + def test_bearer_token_headers(self): + _mock_get() + Endpoint(BASE_URL, 'dummy', BearerToken('kbc_at_dummy', 1234))._get(URL) + + headers = responses.calls[0].request.headers + self.assertEqual('Bearer kbc_at_dummy', headers['Authorization']) + self.assertEqual('1234', headers['X-KBC-ProjectId']) + self.assertNotIn('X-StorageApi-Token', headers) + + @responses.activate + def test_common_headers_sent_with_both_strategies(self): + for token in ('dummy_token', BearerToken('kbc_at_dummy', 1234)): + with self.subTest(token=token): + responses.reset() + _mock_get() + Endpoint(BASE_URL, 'dummy', token)._get(URL) + + headers = responses.calls[0].request.headers + self.assertEqual('gzip', headers['Accept-Encoding']) + self.assertEqual('Keboola Storage API Python Client', headers['User-Agent']) + + def test_token_attribute_is_the_raw_token(self): + self.assertEqual('dummy_token', Endpoint(BASE_URL, 'dummy', 'dummy_token').token) + self.assertEqual('kbc_at_dummy', + Endpoint(BASE_URL, 'dummy', BearerToken('kbc_at_dummy', 1234)).token) + + def test_missing_token(self): + with self.assertRaisesRegex(ValueError, 'Token is required.'): + Endpoint(BASE_URL, 'dummy', '') + + @responses.activate + def test_caller_headers_are_kept_and_not_mutated(self): + _mock_get() + caller_headers = {'x-foo': 'bar'} + Endpoint(BASE_URL, 'dummy', 'dummy_token')._get(URL, headers=caller_headers) + + headers = responses.calls[0].request.headers + self.assertEqual('bar', headers['x-foo']) + self.assertEqual('dummy_token', headers['X-StorageApi-Token']) + self.assertEqual({'x-foo': 'bar'}, caller_headers) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/mocks/test_buckets.py b/tests/mocks/test_buckets.py index fba81ed..08f4afc 100644 --- a/tests/mocks/test_buckets.py +++ b/tests/mocks/test_buckets.py @@ -1,5 +1,6 @@ import unittest import responses +from kbcstorage.auth import BearerToken from kbcstorage.buckets import Buckets from .bucket_responses import list_response, detail_response, create_response @@ -10,6 +11,26 @@ def setUp(self): base_url = 'https://connection.keboola.com/' self.buckets = Buckets(base_url, token) + @responses.activate + def test_list_tables_sends_one_auth_scheme_only(self): + """ + list_tables passes its own headers to _get, so it is the one method that + could ship both schemes at once. + """ + url = 'https://connection.keboola.com/v2/storage/buckets/in.c-ga/tables' + for token, expected, unexpected in ( + ('dummy_token', 'X-StorageApi-Token', 'Authorization'), + (BearerToken('kbc_at_dummy', 1234), 'Authorization', 'X-StorageApi-Token'), + ): + with self.subTest(token=token): + responses.reset() + responses.add(responses.Response(method='GET', url=url, json=[])) + Buckets('https://connection.keboola.com/', token).list_tables('in.c-ga') + + headers = responses.calls[0].request.headers + self.assertIn(expected, headers) + self.assertNotIn(unexpected, headers) + @responses.activate def test_list(self): """ diff --git a/tests/mocks/test_client.py b/tests/mocks/test_client.py index f7cfddc..f68e838 100644 --- a/tests/mocks/test_client.py +++ b/tests/mocks/test_client.py @@ -1,6 +1,9 @@ import unittest -from kbcstorage.client import Client +from kbcstorage.client import BearerToken, Client + +ENDPOINTS = ('buckets', 'files', 'jobs', 'tables', 'workspaces', 'components', + 'configurations', 'tokens', 'branches', 'triggers') class TestClient(unittest.TestCase): @@ -14,3 +17,39 @@ def test_token_not_updatable(self): def test_url_trimmed(self): client = Client('https://example.com/', 'password') self.assertEqual(client.root_url, 'https://example.com') + + def test_positional_args_still_accepted(self): + client = Client('https://example.com', 'password', 'dev-123', False) + self.assertEqual('dev-123', client.branch_id) + self.assertFalse(hasattr(client, 'files')) + + def test_storage_api_token_reaches_every_endpoint(self): + client = Client('https://example.com', 'password') + self.assertIsNone(client.project_id) + + for name in ENDPOINTS: + with self.subTest(endpoint=name): + headers = getattr(client, name)._auth_header + self.assertEqual('password', headers['X-StorageApi-Token']) + self.assertNotIn('Authorization', headers) + + def test_bearer_token_reaches_every_endpoint(self): + client = Client('https://example.com', BearerToken('kbc_at_dummy', 1234)) + self.assertEqual('kbc_at_dummy', client.token) + self.assertEqual('1234', client.project_id) + + for name in ENDPOINTS: + with self.subTest(endpoint=name): + headers = getattr(client, name)._auth_header + self.assertEqual('Bearer kbc_at_dummy', headers['Authorization']) + self.assertEqual('1234', headers['X-KBC-ProjectId']) + self.assertNotIn('X-StorageApi-Token', headers) + + def test_bearer_token_reaches_nested_metadata_endpoints(self): + client = Client('https://example.com', BearerToken('kbc_at_dummy', 1234)) + + for endpoint in (client.tables.metadata, client.configurations.metadata): + with self.subTest(endpoint=endpoint): + headers = endpoint._auth_header + self.assertEqual('Bearer kbc_at_dummy', headers['Authorization']) + self.assertNotIn('X-StorageApi-Token', headers) diff --git a/tests/mocks/test_workspaces.py b/tests/mocks/test_workspaces.py index f598562..0542c82 100644 --- a/tests/mocks/test_workspaces.py +++ b/tests/mocks/test_workspaces.py @@ -8,6 +8,7 @@ import responses from requests import HTTPError +from kbcstorage.auth import BearerToken from kbcstorage.workspaces import Workspaces from .token_responses import verify_token_response @@ -93,6 +94,34 @@ def test_detail_inexsitent_workspace(self): self.ws.detail(workspace_id) assert error_context.exception.args[0] == msg + @responses.activate + def test_create_propagates_bearer_token_to_the_tokens_endpoint(self): + """ + The Tokens endpoint that create() builds internally to resolve the + default backend authenticates the same way the Workspaces endpoint does. + """ + responses.add( + responses.Response( + method='GET', + url='https://connection.keboola.com/v2/storage/tokens/verify', + json=verify_token_response + ) + ) + responses.add( + responses.Response( + method='POST', + url='https://connection.keboola.com/v2/storage/workspaces', + json=keypair_create_response + ) + ) + ws = Workspaces('https://connection.keboola.com/', BearerToken('kbc_at_dummy', 1234)) + ws.create() + + for call in responses.calls: + self.assertEqual('Bearer kbc_at_dummy', call.request.headers['Authorization']) + self.assertEqual('1234', call.request.headers['X-KBC-ProjectId']) + self.assertNotIn('X-StorageApi-Token', call.request.headers) + @responses.activate def test_create(self): """ From 36673e257eb0c7200b54588dc23895de41209ae1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jozef=20Bi=CC=81ros=CC=8C?= Date: Thu, 17 Sep 2026 17:49:09 +0200 Subject: [PATCH 4/5] feat: type the auth strategies and guard the token property Types ----- `coerce` was unannotated, so a type checker inferred `Unknown | StorageApiToken` for its result. It is now `Union[str, Auth] -> Auth` with `Auth = Union[StorageApiToken, BearerToken]`, and the strategies annotate their own constructors and `headers()`. `isinstance` replaces the `hasattr` duck-typing check so the union narrows; nothing consumes `kbcstorage.auth` yet, so no contract is lost. basedpyright now reports `StorageApiToken | BearerToken` for all three input forms. Programmatic tokens are rejected as Storage API tokens ------------------------------------------------------ A `kbc_at_*`/`kbc_pat_*` token cannot authenticate as `X-StorageApi-Token`; the backend answers 401. `StorageApiToken` now raises instead, so the mistake surfaces at construction rather than at the request. The check sits in the constructor so both `coerce` and an explicit `StorageApiToken(...)` are covered, and the message never echoes the token. The `token` property -------------------- Under a bearer token `.token` holds the programmatic token, not a Storage API token, and anything that forwards it into a Storage API header breaks. `Client.token` and `Endpoint.token` document that, and `Endpoint.token` became a read-only property: it was a plain attribute whose assignment silently did nothing, since the header block is built in `__init__`. `coerce` also warns once when a bearer token is used, and warns that a bare string is deprecated. Warning attribution ------------------- Endpoints build other endpoints, so a fixed `stacklevel` named a line inside kbcstorage. One `Client(BearerToken(...))` emitted 13 warnings, 12 of them pointing at library internals, and a DeprecationWarning attributed to a library module is dropped by the default filter, which hid the bare-string deprecation for every Endpoint subclass. `_caller_stacklevel` walks out of the package, which collapses those 13 to one and makes the deprecation visible. Tables and Configurations passed the raw token to their metadata endpoint instead of `self.auth`, coercing twice and building a redundant strategy. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 36 +++++++++++++------ kbcstorage/auth.py | 67 ++++++++++++++++++++++++++++++++---- kbcstorage/base.py | 12 ++++++- kbcstorage/client.py | 7 ++++ kbcstorage/configurations.py | 2 +- kbcstorage/tables.py | 2 +- tests/mocks/test_auth.py | 20 +++++++++-- 7 files changed, 124 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 8cd122a..6d1e853 100644 --- a/README.md +++ b/README.md @@ -18,9 +18,9 @@ pip install git+https://github.com/keboola/sapi-python-client.git ## Client Class Usage ```python -from kbcstorage.client import Client +from kbcstorage.client import Client, StorageApiToken -client = Client('https://connection.keboola.com', 'your-token') +client = Client('https://connection.keboola.com', StorageApiToken('your-token')) # get table data into local file client.tables.export_to_file(table_id='in.c-demo.some-table', path_name='/data/') @@ -41,25 +41,41 @@ client.tables.detail('in.c-demo.some-table') ## Authentication -By default the token is a Storage API token, sent as `X-StorageApi-Token`. Programmatic tokens -(`kbc_at_*` session tokens, `kbc_pat_*` personal access tokens) are not bound to a project, so they -are sent as `Authorization: Bearer` together with the project id in `X-KBC-ProjectId`: +The `token` argument takes an authentication strategy, either `StorageApiToken` or `BearerToken`: ```python -from kbcstorage.client import Client, BearerToken +from kbcstorage.client import Client, BearerToken, StorageApiToken + +# Storage API token, sent as `X-StorageApi-Token` +client = Client('https://connection.keboola.com', StorageApiToken('your-token')) +# Programmatic token, sent as `Authorization: Bearer` with the project id in `X-KBC-ProjectId` client = Client('https://connection.keboola.com', BearerToken('kbc_at_...', project_id=1234)) ``` -Both forms work for directly constructed endpoints too, and the token is used as given — the client -never refreshes or decodes it, so an expired token surfaces as a `requests.HTTPError` with a 401. +A bare string is also accepted and is equivalent to `StorageApiToken('your-token')`. It is kept for +backward compatibility: + +```python +client = Client('https://connection.keboola.com', 'your-token') +``` + +Storage API tokens are bound to a project. Programmatic tokens (`kbc_at_*` session tokens, +`kbc_pat_*` personal access tokens) are not, which is why `BearerToken` requires the project id. + +All three forms work for directly constructed endpoints too, and the token is used as given — the +client never refreshes or decodes it, so an expired token surfaces as a `requests.HTTPError` with a +401. ## Endpoint Classes Usage ```python +from kbcstorage.auth import StorageApiToken from kbcstorage.tables import Tables from kbcstorage.buckets import Buckets -tables = Tables('https://connection.keboola.com', 'your-token') +token = StorageApiToken('your-token') + +tables = Tables('https://connection.keboola.com', token) # get table data into local file tables.export_to_file(table_id='in.c-demo.some-table', path_name='/data/') @@ -68,7 +84,7 @@ tables.export_to_file(table_id='in.c-demo.some-table', path_name='/data/') tables.create(name='some-table-2', bucket_id='in.c-demo', file_path='/data/some-table') # list buckets -buckets = Buckets('https://connection.keboola.com', 'your-token') +buckets = Buckets('https://connection.keboola.com', token) buckets.list() # list bucket tables diff --git a/kbcstorage/auth.py b/kbcstorage/auth.py index c63f24a..6b0e053 100644 --- a/kbcstorage/auth.py +++ b/kbcstorage/auth.py @@ -1,13 +1,45 @@ +import os +import sys +import warnings +from typing import Dict, Union + +PROGRAMMATIC_PREFIXES = ('kbc_at_', 'kbc_pat_') + +_PKG_DIR = os.path.dirname(os.path.abspath(__file__)) + + +def _caller_stacklevel() -> int: + """ + Frames to skip so a warning lands on the first caller outside this package. + + Endpoints build other endpoints, so a fixed stacklevel points at kbcstorage + itself: the message names the wrong line, the same warning repeats once per + endpoint, and a DeprecationWarning attributed to a library module is dropped + by the default filter. + """ + level, frame = 1, sys._getframe(1) + while frame is not None and os.path.dirname(frame.f_code.co_filename) == _PKG_DIR: + level += 1 + frame = frame.f_back + return level + + class StorageApiToken: """ Legacy Storage API token, sent as ``X-StorageApi-Token``. """ - def __init__(self, token): + def __init__(self, token: str) -> None: if not token: raise ValueError("Token is required.") + # never echo the token itself, the message can end up in logs + if token.startswith(PROGRAMMATIC_PREFIXES): + raise ValueError( + "A token with a 'kbc_at_'/'kbc_pat_' prefix is a programmatic " + "token and cannot be sent as X-StorageApi-Token. Use " + "kbcstorage.auth.BearerToken with the project id instead.") self.token = token - def headers(self): + def headers(self) -> Dict[str, str]: return {'X-StorageApi-Token': self.token} @@ -15,7 +47,7 @@ class BearerToken: """ Programmatic session or personal access token (``kbc_at_*``, ``kbc_pat_*``). """ - def __init__(self, token, project_id): + def __init__(self, token: str, project_id: Union[str, int]) -> None: if not token: raise ValueError("Token is required.") if not project_id: @@ -25,12 +57,15 @@ def __init__(self, token, project_id): # the numeric id straight from a token verify payload self.project_id = str(project_id) - def headers(self): + def headers(self) -> Dict[str, str]: return {'Authorization': 'Bearer {}'.format(self.token), 'X-KBC-ProjectId': self.project_id} -def coerce(token): +Auth = Union[StorageApiToken, BearerToken] + + +def coerce(token: Union[str, Auth]) -> Auth: """ Normalise the ``token`` argument of an endpoint into a strategy object. @@ -41,6 +76,24 @@ def coerce(token): Returns: The strategy object to authenticate with. """ - if hasattr(token, 'headers'): + if isinstance(token, StorageApiToken): + return token + + if isinstance(token, BearerToken): + warnings.warn( + "Authenticating with a bearer token: `.token` returns the " + "programmatic token, not a Storage API token, and will be " + "rejected if sent as X-StorageApi-Token. Pass `.auth` when " + "building further endpoints.", + UserWarning, + stacklevel=_caller_stacklevel(), + ) return token - return StorageApiToken(token) + auth = StorageApiToken(token) + warnings.warn( + "Passing the token as a string is deprecated, " + "pass kbcstorage.auth.StorageApiToken instead.", + DeprecationWarning, + stacklevel=_caller_stacklevel(), + ) + return auth diff --git a/kbcstorage/base.py b/kbcstorage/base.py index 04631ab..9249185 100644 --- a/kbcstorage/base.py +++ b/kbcstorage/base.py @@ -49,13 +49,23 @@ def __init__(self, root_url, path_component, token, max_requests_retries=MAX_RET self.base_url = '{}/v2/storage/{}'.format(root_url.strip('/'), path_component.strip('/')) self.auth = coerce(token) - self.token = self.auth.token self._auth_header = {**self.auth.headers(), 'X-KBC-RunId': os.environ.get('KBC_RUNID'), 'Accept-Encoding': 'gzip', 'User-Agent': 'Keboola Storage API Python Client'} self.requests = RetryRequests(max_requests_retries) + @property + def token(self): + """ + The raw token string, whichever scheme is in use. + + Under a bearer token this is the programmatic token, not a Storage API + token, so sending it as `X-StorageApi-Token` fails. Pass `auth` when + building further endpoints. + """ + return self.auth.token + def _get_raw(self, url, params=None, **kwargs): """ Construct a requests GET call with args and kwargs and process the diff --git a/kbcstorage/client.py b/kbcstorage/client.py index 7477839..e24242d 100644 --- a/kbcstorage/client.py +++ b/kbcstorage/client.py @@ -52,6 +52,13 @@ def __init__(self, api_domain, token, branch_id='default', file_storage_support= @property def token(self): + """ + The raw token string, whichever scheme is in use. + + Under a bearer token this is the programmatic token, not a Storage API + token, so sending it as `X-StorageApi-Token` fails. Pass `auth` when + building further endpoints. + """ return self._token @property diff --git a/kbcstorage/configurations.py b/kbcstorage/configurations.py index 23331f2..77ad382 100644 --- a/kbcstorage/configurations.py +++ b/kbcstorage/configurations.py @@ -23,7 +23,7 @@ def __init__(self, root_url, token, branch_id): branch_id (str): The ID of branch to use, use 'default' to work without branch (in main). """ super().__init__(root_url, f"branch/{branch_id}/components", token) - self.metadata = ConfigurationsMetadata(root_url, token, branch_id) + self.metadata = ConfigurationsMetadata(root_url, self.auth, branch_id) def detail(self, component_id, configuration_id): """ diff --git a/kbcstorage/tables.py b/kbcstorage/tables.py index c65da2f..66a6096 100644 --- a/kbcstorage/tables.py +++ b/kbcstorage/tables.py @@ -27,7 +27,7 @@ def __init__(self, root_url, token): token (:obj:`str`): A storage API key. """ super().__init__(root_url, 'tables', token) - self.metadata = TablesMetadata(root_url, token) + self.metadata = TablesMetadata(root_url, self.auth) def list(self, include=None): """ diff --git a/tests/mocks/test_auth.py b/tests/mocks/test_auth.py index 6dc5b14..2399555 100644 --- a/tests/mocks/test_auth.py +++ b/tests/mocks/test_auth.py @@ -12,6 +12,17 @@ def test_empty_token(self): with self.assertRaisesRegex(ValueError, 'Token is required.'): StorageApiToken('') + def test_programmatic_token_is_rejected(self): + for token in ('kbc_at_dummy', 'kbc_pat_dummy'): + with self.subTest(token=token): + with self.assertRaisesRegex(ValueError, 'programmatic'): + StorageApiToken(token) + + def test_error_does_not_leak_the_token(self): + with self.assertRaises(ValueError) as ctx: + StorageApiToken('kbc_at_secret_value') + self.assertNotIn('secret_value', str(ctx.exception)) + class TestBearerToken(unittest.TestCase): def test_headers(self): @@ -40,8 +51,13 @@ def test_bare_string_becomes_storage_api_token(self): self.assertEqual('dummy_token', auth.token) def test_strategy_passes_through(self): - auth = BearerToken('kbc_at_dummy', 1234) - self.assertIs(auth, coerce(auth)) + for auth in (BearerToken('kbc_at_dummy', 1234), StorageApiToken('dummy_token')): + with self.subTest(auth=type(auth).__name__): + self.assertIs(auth, coerce(auth)) + + def test_programmatic_token_string_is_rejected(self): + with self.assertRaisesRegex(ValueError, 'programmatic'): + coerce('kbc_at_dummy') if __name__ == '__main__': From d4d0b698781649bc19ac4e51e740b990cee20096 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jozef=20Bi=CC=81ros=CC=8C?= Date: Thu, 24 Sep 2026 15:13:37 +0200 Subject: [PATCH 5/5] fix: drop the auth warnings and correct the supported Python version coerce() runs once per endpoint, and endpoints build further endpoints, so a warning there fires 13 times for a single Client(). The bearer caveat is in the .token docstrings; the string form is supported, not deprecated. requires-python and the README claimed 3.7, but workspaces.load_tables has `int | str` annotations that fail at import below 3.10. Co-Authored-By: Claude Opus 5 (1M context) --- README.md | 14 +++++++++--- kbcstorage/auth.py | 48 +++++----------------------------------- pyproject.toml | 4 +++- tests/mocks/test_auth.py | 12 ++++++++++ 4 files changed, 31 insertions(+), 47 deletions(-) diff --git a/README.md b/README.md index 6d1e853..a79505d 100644 --- a/README.md +++ b/README.md @@ -53,8 +53,8 @@ client = Client('https://connection.keboola.com', StorageApiToken('your-token')) client = Client('https://connection.keboola.com', BearerToken('kbc_at_...', project_id=1234)) ``` -A bare string is also accepted and is equivalent to `StorageApiToken('your-token')`. It is kept for -backward compatibility: +A bare string is also accepted and is equivalent to `StorageApiToken('your-token')`. It is fully +supported and not deprecated: ```python client = Client('https://connection.keboola.com', 'your-token') @@ -67,6 +67,13 @@ All three forms work for directly constructed endpoints too, and the token is us client never refreshes or decodes it, so an expired token surfaces as a `requests.HTTPError` with a 401. +### Breaking change in 0.11.0 + +A bare string starting with `kbc_at_` or `kbc_pat_` now raises `ValueError` at construction. Before +0.11.0 it was sent as `X-StorageApi-Token`, which the backend rejects with a 401 — the failure moved +from request time to construction time and the message names the fix. Wrap such tokens in +`BearerToken(token, project_id)`. + ## Endpoint Classes Usage ```python from kbcstorage.auth import StorageApiToken @@ -99,7 +106,8 @@ tables.detail('in.c-demo.some-table') ### Local environment -The package requires Python 3.7 or newer; CI builds and tests on 3.11. +The package requires Python 3.10 or newer (`workspaces.load_tables` uses `int | str` annotations); CI builds +and tests on 3.11. Create a virtual environment and install the package into it in editable mode: diff --git a/kbcstorage/auth.py b/kbcstorage/auth.py index 6b0e053..61e2c94 100644 --- a/kbcstorage/auth.py +++ b/kbcstorage/auth.py @@ -1,28 +1,7 @@ -import os -import sys -import warnings from typing import Dict, Union PROGRAMMATIC_PREFIXES = ('kbc_at_', 'kbc_pat_') -_PKG_DIR = os.path.dirname(os.path.abspath(__file__)) - - -def _caller_stacklevel() -> int: - """ - Frames to skip so a warning lands on the first caller outside this package. - - Endpoints build other endpoints, so a fixed stacklevel points at kbcstorage - itself: the message names the wrong line, the same warning repeats once per - endpoint, and a DeprecationWarning attributed to a library module is dropped - by the default filter. - """ - level, frame = 1, sys._getframe(1) - while frame is not None and os.path.dirname(frame.f_code.co_filename) == _PKG_DIR: - level += 1 - frame = frame.f_back - return level - class StorageApiToken: """ @@ -70,30 +49,13 @@ def coerce(token: Union[str, Auth]) -> Auth: Normalise the ``token`` argument of an endpoint into a strategy object. Args: - token: A strategy object, or a bare token string for backward - compatibility. + token: A strategy object, or a bare token string. Endpoints build other + endpoints and pass their own strategy on, so this runs once per + endpoint and must stay free of side effects. Returns: The strategy object to authenticate with. """ - if isinstance(token, StorageApiToken): - return token - - if isinstance(token, BearerToken): - warnings.warn( - "Authenticating with a bearer token: `.token` returns the " - "programmatic token, not a Storage API token, and will be " - "rejected if sent as X-StorageApi-Token. Pass `.auth` when " - "building further endpoints.", - UserWarning, - stacklevel=_caller_stacklevel(), - ) + if isinstance(token, (StorageApiToken, BearerToken)): return token - auth = StorageApiToken(token) - warnings.warn( - "Passing the token as a string is deprecated, " - "pass kbcstorage.auth.StorageApiToken instead.", - DeprecationWarning, - stacklevel=_caller_stacklevel(), - ) - return auth + return StorageApiToken(token) diff --git a/pyproject.toml b/pyproject.toml index 91b016e..c14cef9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,11 +9,13 @@ authors = [ ] description = "Client for Keboola Storage API" readme = "README.md" -requires-python = ">=3.7" +requires-python = ">=3.10" keywords = ["keboola", "storage"] license = {file = "LICENSE"} classifiers = [ "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", "License :: OSI Approved :: MIT License", "Operating System :: OS Independent", "Development Status :: 5 - Production/Stable", diff --git a/tests/mocks/test_auth.py b/tests/mocks/test_auth.py index 2399555..83623d3 100644 --- a/tests/mocks/test_auth.py +++ b/tests/mocks/test_auth.py @@ -1,4 +1,5 @@ import unittest +import warnings from kbcstorage.auth import BearerToken, StorageApiToken, coerce @@ -59,6 +60,17 @@ def test_programmatic_token_string_is_rejected(self): with self.assertRaisesRegex(ValueError, 'programmatic'): coerce('kbc_at_dummy') + def test_no_warnings(self): + # endpoints build other endpoints, so coerce() runs once per endpoint; + # a warning here fires a dozen times per client and raises under -W error + for token in ('dummy_token', StorageApiToken('dummy_token'), + BearerToken('kbc_at_dummy', 1234)): + with self.subTest(token=type(token).__name__): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + coerce(token) + self.assertEqual([], [str(w.message) for w in caught]) + if __name__ == '__main__': unittest.main()