diff --git a/CHANGELOG.md b/CHANGELOG.md index 7048f30..2e85e65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +* bexio.py: new library +* url.py: added new optional flag `response_on_error` for `fetch()`/`fetch_json()` to return the response body instead of an error message on failure. Primarily for use with APIs that provide machine-readable error responses such as the Bexio API. + ### Changed * args.py: the developer-only `--test` parameter is now hidden from plugin `--help` output. It is still accepted on the command line, so the unit-test suite keeps working. diff --git a/README.md b/README.md index ad248ee..b36a54b 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,7 @@ These libraries are built with a clear set of priorities: | Module | Description | Key Functions | |--------|-------------|---------------| +| **bexio.py** | [Bexio](https://www.bexio.com/) all-in-one business software REST API. | `fetch_json()` | | **grassfish.py** | [Grassfish](https://www.grassfish.com/) digital signage REST API. | `fetch_json()` | | **huawei.py** | Huawei storage system status parsing (controller models, health, LED status). | `get_controller_model()` | | **icinga.py** | Icinga2 REST API client for querying services, setting acknowledgements, and managing downtimes. | `get_service()`, `set_ack()`, `set_downtime()`, `remove_downtime()` | diff --git a/bexio.py b/bexio.py new file mode 100644 index 0000000..a40592e --- /dev/null +++ b/bexio.py @@ -0,0 +1,987 @@ +#! /usr/bin/env python3 +# -*- coding: utf-8; py-indent-offset: 4 -*- +# +# Author: Linuxfabrik GmbH, Zurich, Switzerland +# Contact: info (at) linuxfabrik (dot) ch +# https://www.linuxfabrik.ch/ +# License: The Unlicense, see LICENSE file. + +# https://github.com/Linuxfabrik/monitoring-plugins/blob/main/CONTRIBUTING.rst + +"""Interacts with the Bexio API, providing a simplified interface for use with Python.""" + +__author__ = 'Linuxfabrik GmbH, Zurich/Switzerland' +__version__ = '2026060501' + +import urllib.parse + +from . import url + +# https://docs.bexio.com/ +BEXIO_API_BASE_URL = 'https://api.bexio.com' +BEXIO_API_CONTACT_TYPE_COMPANY = 1 +BEXIO_API_CONTACT_TYPE_PERSON = 2 +BEXIO_API_ACCOUNT_URL = '/2.0/accounts' +BEXIO_API_BANK_ACCOUNT_URL = '/3.0/banking/accounts' +# API endpoint still uses an old name in the URL for "business activities" +BEXIO_API_BUSINESS_ACTIVITY_URL = '/2.0/client_service' +BEXIO_API_CONTACT_URL = '/2.0/contact' +BEXIO_API_CONTACT_GROUP_URL = '/2.0/contact_group' +BEXIO_API_CONTACT_RELATION_URL = '/2.0/contact_relation' +# API endpoint still uses an old name in the URL for "contact sectors" +BEXIO_API_CONTACT_SECTOR_URL = '/2.0/contact_branch' +BEXIO_API_COUNTRY_URL = '/2.0/country' +BEXIO_API_CURRENCY_URL = '/3.0/currencies' +BEXIO_API_DEFAULT_API_CALL_TIMEOUT = 20 +BEXIO_API_INVOICE_URL = '/2.0/kb_invoice' +BEXIO_API_ITEM_URL = '/2.0/article' # API endpoint uses a different name in the URL +BEXIO_API_LANGUAGE_URL = '/2.0/language' +BEXIO_API_SALUTATION_URL = '/2.0/salutation' +BEXIO_API_PAYMENT_TYPE_URL = '/2.0/payment_type' +BEXIO_API_PROJECT_STATUS_URL = '/2.0/pr_project_state' +BEXIO_API_PROJECT_TYPE_URL = '/2.0/pr_project_type' +BEXIO_API_PROJECT_URL = '/2.0/pr_project' +BEXIO_API_STOCK_AREA_URL = '/2.0/stock_place' +BEXIO_API_STOCK_LOCATION_URL = '/2.0/stock' +BEXIO_API_TAX_URL = '/3.0/taxes' +BEXIO_API_TIMESHEET_URL = '/2.0/timesheet' +BEXIO_API_TIMESHEET_STATUS_URL = '/2.0/timesheet_status' +BEXIO_API_TITLE_URL = '/2.0/title' +BEXIO_API_UNIT_URL = '/2.0/unit' +BEXIO_API_USER_URL = '/3.0/users' + + +def call_api(api_token, path, data=None, method=None): + """ + Makes an HTTP GET or POST call against the Bexio API and returns the parsed JSON. + + ### Parameters + - **api_token** (`str`) + Bexio API Token. Create at https://developer.bexio.com/pat. + - **path** (`str`) + The URL part to call. + - **data** (`dict`, optional) + Dictionary that will be sent as JSON data. + - **method** (`str`, optional) + HTTP method to use for the request. Defaults to GET if data is None else POST. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `list` | `str`): + - On success, the parsed JSON document from the Bexio API + - On failure, an error message string. + """ + if data is None: + data = {} + + headers = { + 'Accept': 'application/json', + 'Authorization': f'Bearer {api_token}', + } + + return url.fetch_json( + f'{BEXIO_API_BASE_URL}{path}', + data=data, + header=headers, + timeout=BEXIO_API_DEFAULT_API_CALL_TIMEOUT, + encoding='serialized-json', + method=method, + response_on_error=True, + ) + + +def create_contact(api_token, data=None): + """ + Creates a contact using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **data** (`dict`, optional): + Contact data to be created, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the created contact. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Contacts/operation/v2CreateContact + """ + return call_api(api_token, BEXIO_API_CONTACT_URL, data) + + +def create_contact_relation(api_token, data=None): + """ + Creates a contact relation using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **data** (`dict`): + Contact relation data to be created, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the created contact relation. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: + https://docs.bexio.com/#tag/Contact-Relations/operation/v2CreateContactRelation + """ + return call_api(api_token, BEXIO_API_CONTACT_RELATION_URL, data) + + +def create_invoice(api_token, data=None): + """ + Creates an invoice using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **data** (`dict`, optional): + Invoice data to be created, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the created invoice. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Invoices/operation/v2CreateInvoice + """ + return call_api(api_token, BEXIO_API_INVOICE_URL, data) + + +def create_item(api_token, data=None): + """ + Creates an item using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **data** (`dict`, optional): + Item data to be created, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the created item. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Items/operation/v2CreateItem + """ + return call_api(api_token, BEXIO_API_ITEM_URL, data) + + +def create_project(api_token, data=None): + """ + Creates a project using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **data** (`dict`, optional): + Project data to be created, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the created project. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Projects/operation/v2CreateProject + """ + return call_api(api_token, BEXIO_API_PROJECT_URL, data) + + +def create_timesheet(api_token, data=None): + """ + Creates a timesheet using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **data** (`dict`, optional): + Timesheet data to be created, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the created timesheet. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Timesheets/operation/v2CreateTimesheet + """ + return call_api(api_token, BEXIO_API_TIMESHEET_URL, data) + + +def delete_contact_relation(api_token, contact_relation_id): + """ + Deletes a contact relation using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **contact_relation_id** (`int`): + ID of the contact relation to delete. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the deletion status. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: + https://docs.bexio.com/#tag/Contact-Relations/operation/v2DeleteContactRelation + """ + return call_api( + api_token, + BEXIO_API_CONTACT_RELATION_URL + '/' + str(contact_relation_id), + method='DELETE', + ) + + +def edit_contact(api_token, contact_id, data=None): + """ + Edits a contact using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **contact_id** (`int`): + ID of the contact to edit. + - **data** (`dict`): + Contact data to be edited, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the edited contact. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Contacts/operation/v2EditContact + """ + return call_api(api_token, BEXIO_API_CONTACT_URL + '/' + str(contact_id), data) + + +def edit_contact_relation(api_token, contact_relation_id, data=None): + """ + Edits a contact relation using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **contact_relation_id** (`int`): + ID of the contact relation to edit. + - **data** (`dict`): + Contact relation data to be edited, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the edited contact relation. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: + https://docs.bexio.com/#tag/Contact-Relations/operation/v2EditContactRelation + """ + return call_api( + api_token, + BEXIO_API_CONTACT_RELATION_URL + '/' + str(contact_relation_id), + data, + ) + + +def edit_invoice(api_token, invoice_id, data=None): + """ + Edits an invoice using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **invoice_id** (`int`): + ID of the invoice to edit. + - **data** (`dict`, optional): + Invoice data to be edited, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the edited invoice. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Invoices/operation/v2EditInvoice + """ + return call_api(api_token, BEXIO_API_INVOICE_URL + '/' + str(invoice_id), data) + + +def edit_item(api_token, item_id, data=None): + """ + Edits an item using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **item_id** (`int`): + ID of the item to edit. + - **data** (`dict`, optional): + Item data to be edited, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the edited item. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Items/operation/v2EditItem + """ + return call_api(api_token, BEXIO_API_ITEM_URL + '/' + str(item_id), data) + + +def edit_project(api_token, project_id, data=None): + """ + Edits a project using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **project_id** (`int`): + ID of the project to edit. + - **data** (`dict`, optional): + Project data to be edited, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the edited project. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Projects/operation/v2EditProject + """ + return call_api(api_token, BEXIO_API_PROJECT_URL + '/' + str(project_id), data) + + +def edit_timesheet(api_token, timesheet_id, data=None): + """ + Edits a timesheet using the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **timesheet_id** (`int`): + ID of the timesheet to edit. + - **data** (`dict`, optional): + Project data to be edited, as per the Bexio API documentation. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`dict` | `str`): + - On success, a dictionary of the edited timesheet. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Timesheets/operation/v2EditTimesheet + """ + return call_api(api_token, BEXIO_API_TIMESHEET_URL + '/' + str(timesheet_id), data) + + +def fetch_accounts(api_token): + """ + Fetches all accounts from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all accounts as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Accounts/operation/v2ListAccounts + """ + return get_all(api_token, BEXIO_API_ACCOUNT_URL) + + +def fetch_bank_accounts(api_token): + """ + Fetches all bank accounts from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all bank accounts as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to API doc: https://docs.bexio.com/#tag/Bank-Accounts/operation/ListBankAccounts + """ + return get_all(api_token, BEXIO_API_BANK_ACCOUNT_URL) + + +def fetch_business_activities(api_token): + """ + Fetches all business activities from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all business activities as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to API doc: + https://docs.bexio.com/#tag/Business-Activities/operation/v2ListBusinessActivities + """ + return get_all(api_token, BEXIO_API_BUSINESS_ACTIVITY_URL) + + +def fetch_contact_groups(api_token): + """ + Fetches all contact groups from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all contact groups as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Contact-Groups/operation/v2ListContactGroups + """ + return get_all(api_token, BEXIO_API_CONTACT_GROUP_URL) + + +def fetch_contact_relations(api_token): + """ + Fetches all contact relations from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all contact relations as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: + https://docs.bexio.com/#tag/Contact-Relations/operation/v2ListContactRelations + """ + return get_all(api_token, BEXIO_API_CONTACT_RELATION_URL) + + +def fetch_contact_sectors(api_token): + """ + Fetches all contact sectors (branches) from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all contact sectors (branches) as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: + https://docs.bexio.com/#tag/Contact-Sectors/operation/v2ListContactSectors + """ + return get_all(api_token, BEXIO_API_CONTACT_SECTOR_URL) + + +def fetch_contacts(api_token, archived=False): + """ + Fetches all (optionally including archived) contacts from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + - **archived** (`bool`, optional): + If `True`, also request archived contacts from the API. Defaults to `False`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all contacts as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Contacts/operation/v2ListContacts + """ + return get_all(api_token, BEXIO_API_CONTACT_URL, {'show_archived': archived}) + + +def fetch_countries(api_token): + """ + Fetches all countries from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all countries as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Countries/operation/v2ListCountries + """ + return get_all(api_token, BEXIO_API_COUNTRY_URL) + + +def fetch_currencies(api_token): + """ + Fetches all currencies from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all currencies as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Currencies/operation/ListCurrencies + """ + return get_all(api_token, BEXIO_API_CURRENCY_URL) + + +def fetch_invoices(api_token): + """ + Fetches all invoices from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all invoices as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Invoices/operation/v2ListInvoices + """ + return get_all(api_token, BEXIO_API_INVOICE_URL) + + +def fetch_items(api_token): + """ + Fetches all items from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all items as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Items/operation/v2ListItems + """ + return get_all(api_token, BEXIO_API_ITEM_URL) + + +def fetch_languages(api_token): + """ + Fetches all languages from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all languages as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Languages/operation/v2ListLanguages + """ + return get_all(api_token, BEXIO_API_LANGUAGE_URL) + + +def fetch_payment_types(api_token): + """ + Fetches all payment types from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all payment types as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Payment-Types/operation/v2ListPaymentTypes + """ + return get_all(api_token, BEXIO_API_PAYMENT_TYPE_URL) + + +def fetch_project_statuses(api_token): + """ + Fetches all project statuses from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all project statuses as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Projects/operation/v2ListProjectStatus + """ + return get_all(api_token, BEXIO_API_PROJECT_STATUS_URL) + + +def fetch_project_types(api_token): + """ + Fetches all project types from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all project types as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Projects/operation/v2ListProjectType + """ + return get_all(api_token, BEXIO_API_PROJECT_TYPE_URL) + + +def fetch_projects(api_token): + """ + Fetches all projects from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all projects as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Projects/operation/v2ListProjects + """ + return get_all(api_token, BEXIO_API_PROJECT_URL) + + +def fetch_salutations(api_token): + """ + Fetches all salutations from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all salutations as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Salutations/operation/v2ListSalutations + """ + return get_all(api_token, BEXIO_API_SALUTATION_URL) + + +def fetch_stock_areas(api_token): + """ + Fetches all stock areas from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all stock areas as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Stock-Areas/operation/v2ListStockAreas + """ + return get_all(api_token, BEXIO_API_STOCK_AREA_URL) + + +def fetch_stock_locations(api_token): + """ + Fetches all stock locations from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all stock locations as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: + https://docs.bexio.com/#tag/Stock-locations/operation/v2ListStockLocations + """ + return get_all(api_token, BEXIO_API_STOCK_LOCATION_URL) + + +def fetch_taxes(api_token): + """ + Fetches all taxes from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all taxes as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Taxes/operation/ListTaxes + """ + return get_all(api_token, BEXIO_API_TAX_URL) + + +def fetch_timesheet_statuses(api_token): + """ + Fetches all timesheet statuses from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all timesheet statuses as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Timesheets/operation/v2ListTimesheets + """ + return get_all(api_token, BEXIO_API_TIMESHEET_STATUS_URL) + + +def fetch_timesheets(api_token): + """ + Fetches all timesheets from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all timesheets as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Timesheets/operation/v2ListTimesheets + """ + return get_all(api_token, BEXIO_API_TIMESHEET_URL) + + +def fetch_titles(api_token): + """ + Fetches all titles from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all titles as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Titles/operation/v2ListTitles + """ + return get_all(api_token, BEXIO_API_TITLE_URL) + + +def fetch_units(api_token): + """ + Fetches all units from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all units as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/Units/operation/v2ListUnits + """ + return get_all(api_token, BEXIO_API_UNIT_URL) + + +def fetch_users(api_token): + """ + Fetches all users from the Bexio API. + + ### Parameters + - **api_token** (`str`): + See `get_all()`. + + ### Returns + - **tuple**: + - **success** (`bool`): `True` if the request was successful, `False` otherwise. + - **result** (`list` | `str`): + - On success, a list of all users as dictionaries. + - On failure, an error message string. + + ### Notes + - Refer to the API doc: https://docs.bexio.com/#tag/User-Management/operation/v3ListUsers + """ + return get_all(api_token, BEXIO_API_USER_URL) + + +def get_all(api_token, path, params=None): + """ + A wrapper function around `call_api()` that handles the pagination of the Bexio API and + returns all items. + + ### Parameters + - **api_token** (`str`): + See `call_api()`. + - **path** (`str`): + See `call_api()`. + - **params** (`dict` | optional): + Additional URL parameters to be added to the request. + + ### Returns + - **tuple**: + See `call_api()`. + """ + if params is None: + params = {} + + offset = 0 + result = [] + # highest limit of all endpoints. let's use that, if the max is less that's fine as well + max_limit = 2000 + while True: + params['offset'] = offset + params['limit'] = max_limit + current_path = f'{path}?{urllib.parse.urlencode(params)}' + + success, current_result = call_api(api_token, current_path) + if not success: + return success, current_result + result.extend(current_result) + + # we get an empty list if the offset is too high + if len(current_result) == 0: + break + + offset += len(current_result) + + return (True, result) diff --git a/tests/bexio/lib b/tests/bexio/lib new file mode 120000 index 0000000..c25bddb --- /dev/null +++ b/tests/bexio/lib @@ -0,0 +1 @@ +../.. \ No newline at end of file diff --git a/tests/bexio/unit-test/run b/tests/bexio/unit-test/run new file mode 100755 index 0000000..9f96144 --- /dev/null +++ b/tests/bexio/unit-test/run @@ -0,0 +1,378 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8; py-indent-offset: 4 -*- +# +# Author: Linuxfabrik GmbH, Zurich, Switzerland +# Contact: info (at) linuxfabrik (dot) ch +# https://www.linuxfabrik.ch/ +# License: The Unlicense, see LICENSE file. + +# https://github.com/Linuxfabrik/lib/blob/main/CONTRIBUTING.md + +"""Unit tests for lib.bexio. + +The module is a thin client for the Bexio REST API. Every network call goes +through `lib.url.fetch_json`, so the suite is made fully hermetic by replacing +`lib.bexio.url.fetch_json` with a MagicMock. No socket, no DNS, no TLS +handshake ever happens. + +Bexio is a paid hosted SaaS with no public container image, so there is no +container-test class; the unit tests are the full coverage. + +The tests probe three things: + + * `call_api()` is the one place where URL, auth header, encoding, timeout, + and the `response_on_error` flag are constructed; nearly every wrapper + funnels through it, so this gets the bulk of the assertions. + * `get_all()` implements the pagination loop (offset/limit, terminate on + empty page, short-circuit on failure) and is covered end-to-end. + * The thin CRUD wrappers (~30 of them) all reduce to "call `get_all` or + `call_api` with the right URL constant". One representative per category + is asserted, plus the wrappers with non-trivial argument handling + (`fetch_contacts(archived=...)`, `delete_contact_relation` with its + DELETE method, `edit_*` with the id-to-string conversion) and a single + `test_url_constants_match_documented_endpoints` to pin the URL constants. +""" + +import sys +import unittest +import urllib.parse +from unittest import mock + +sys.path.insert(0, '..') + +import lib.bexio as bexio + +API_TOKEN = 'linuxfabrik' + + +def make_fetch_json(return_value=(True, [])): + """Return a MagicMock standing in for url.fetch_json with a canned result.""" + return mock.MagicMock(name='fetch_json', return_value=return_value) + + +class BexioTestBase(unittest.TestCase): + """Patches url.fetch_json for every test. + + self.fetch_json is the mock; the helpers expose the URL and kwargs of the + request the module crafted so the wire contract can be asserted. + """ + + def setUp(self): + self.fetch_json = make_fetch_json() + self._p_fetch = mock.patch.object(bexio.url, 'fetch_json', self.fetch_json) + self._p_fetch.start() + self.addCleanup(self._p_fetch.stop) + + def last_call(self): + """Return (args, kwargs) of the most recent fetch_json call.""" + self.assertTrue(self.fetch_json.called, 'fetch_json was never called') + return self.fetch_json.call_args + + def sent_url(self): + """First positional arg passed to fetch_json (the full URL).""" + return self.last_call().args[0] + + def sent_kwargs(self): + return self.last_call().kwargs + + def sent_header(self): + return self.sent_kwargs()['header'] + + +# --------------------------------------------------------------------------- +# call_api() +# --------------------------------------------------------------------------- + + +class TestCallApi(BexioTestBase): + def test_returns_fetch_json_result_unchanged(self): + sentinel = (True, {'id': 1, 'name': 'Linuxfabrik'}) + self.fetch_json.return_value = sentinel + result = bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertIs(result, sentinel) + + def test_error_tuple_propagates(self): + self.fetch_json.return_value = (False, 'HTTP 401: invalid token') + success, body = bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertFalse(success) + self.assertEqual(body, 'HTTP 401: invalid token') + + def test_full_url_is_base_plus_path(self): + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertEqual(self.sent_url(), 'https://api.bexio.com/2.0/contact') + + def test_authorization_header_is_bearer_token(self): + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertEqual(self.sent_header()['Authorization'], f'Bearer {API_TOKEN}') + + def test_accept_header_is_application_json(self): + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertEqual(self.sent_header()['Accept'], 'application/json') + + def test_no_extra_headers_set(self): + # Regression guard: only Accept and Authorization, nothing else. + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertEqual(set(self.sent_header()), {'Accept', 'Authorization'}) + + def test_encoding_is_serialized_json(self): + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertEqual(self.sent_kwargs()['encoding'], 'serialized-json') + + def test_timeout_is_default_constant(self): + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertEqual( + self.sent_kwargs()['timeout'], bexio.BEXIO_API_DEFAULT_API_CALL_TIMEOUT + ) + + def test_default_api_call_timeout_constant_value(self): + self.assertEqual(bexio.BEXIO_API_DEFAULT_API_CALL_TIMEOUT, 20) + + def test_response_on_error_true_is_forwarded(self): + # Bexio's API returns error details in the response body, so call_api + # must opt into url.fetch_json's response_on_error mode. + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertTrue(self.sent_kwargs()['response_on_error']) + + def test_data_none_becomes_empty_dict(self): + bexio.call_api(API_TOKEN, '/2.0/contact', data=None) + self.assertEqual(self.sent_kwargs()['data'], {}) + + def test_data_dict_passed_through(self): + payload = {'name_1': 'Linuxfabrik', 'contact_type_id': 1} + bexio.call_api(API_TOKEN, '/2.0/contact', data=payload) + self.assertEqual(self.sent_kwargs()['data'], payload) + + def test_method_default_none_forwarded(self): + # No explicit method: fetch_json infers GET/POST from the body presence, + # so call_api must not silently pin a method. + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertIsNone(self.sent_kwargs()['method']) + + def test_method_delete_forwarded(self): + # The only wrapper that overrides the method is delete_contact_relation; + # pin that the override survives the call_api seam. + bexio.call_api(API_TOKEN, '/2.0/contact_relation/1', method='DELETE') + self.assertEqual(self.sent_kwargs()['method'], 'DELETE') + + def test_called_exactly_once(self): + bexio.call_api(API_TOKEN, '/2.0/contact') + self.assertEqual(self.fetch_json.call_count, 1) + + +# --------------------------------------------------------------------------- +# get_all() - pagination loop +# --------------------------------------------------------------------------- + + +def _query(url_with_params): + """Return the parsed query-string of a URL as a dict.""" + return dict(urllib.parse.parse_qsl(urllib.parse.urlparse(url_with_params).query)) + + +class TestGetAll(BexioTestBase): + def test_empty_first_page_returns_empty_list(self): + self.fetch_json.return_value = (True, []) + success, result = bexio.get_all(API_TOKEN, '/2.0/contact') + self.assertTrue(success) + self.assertEqual(result, []) + self.assertEqual(self.fetch_json.call_count, 1) + + def test_single_page_then_terminator_returns_items(self): + self.fetch_json.side_effect = [ + (True, [{'id': 1}, {'id': 2}]), + (True, []), + ] + success, result = bexio.get_all(API_TOKEN, '/2.0/contact') + self.assertTrue(success) + self.assertEqual(result, [{'id': 1}, {'id': 2}]) + self.assertEqual(self.fetch_json.call_count, 2) + + def test_multi_page_pagination_concatenates_in_order(self): + self.fetch_json.side_effect = [ + (True, [{'id': 1}, {'id': 2}]), + (True, [{'id': 3}]), + (True, [{'id': 4}, {'id': 5}, {'id': 6}]), + (True, []), + ] + success, result = bexio.get_all(API_TOKEN, '/2.0/contact') + self.assertTrue(success) + self.assertEqual(result, [{'id': i} for i in range(1, 7)]) + self.assertEqual(self.fetch_json.call_count, 4) + + def test_offset_advances_by_returned_page_length(self): + # Pages of length 2 then 3: offsets must be 0, 2, 5, then a terminator. + self.fetch_json.side_effect = [ + (True, [{'id': 1}, {'id': 2}]), + (True, [{'id': 3}, {'id': 4}, {'id': 5}]), + (True, []), + ] + bexio.get_all(API_TOKEN, '/2.0/contact') + offsets = [ + int(_query(call.args[0])['offset']) + for call in self.fetch_json.call_args_list + ] + self.assertEqual(offsets, [0, 2, 5]) + + def test_limit_is_2000_on_every_request(self): + self.fetch_json.side_effect = [ + (True, [{'id': 1}]), + (True, [{'id': 2}]), + (True, []), + ] + bexio.get_all(API_TOKEN, '/2.0/contact') + limits = { + _query(call.args[0])['limit'] for call in self.fetch_json.call_args_list + } + self.assertEqual(limits, {'2000'}) + + def test_default_params_none_adds_only_offset_and_limit(self): + self.fetch_json.return_value = (True, []) + bexio.get_all(API_TOKEN, '/2.0/contact') + self.assertEqual(set(_query(self.sent_url())), {'offset', 'limit'}) + + def test_extra_params_are_preserved_on_every_page(self): + self.fetch_json.side_effect = [ + (True, [{'id': 1}]), + (True, []), + ] + bexio.get_all(API_TOKEN, '/2.0/contact', params={'show_archived': True}) + for call in self.fetch_json.call_args_list: + params = _query(call.args[0]) + self.assertEqual(params['show_archived'], 'True') + + def test_call_failure_on_first_page_short_circuits(self): + self.fetch_json.return_value = (False, 'boom') + success, body = bexio.get_all(API_TOKEN, '/2.0/contact') + self.assertFalse(success) + self.assertEqual(body, 'boom') + self.assertEqual(self.fetch_json.call_count, 1) + + def test_call_failure_on_second_page_discards_partial_result(self): + # The current contract is all-or-nothing: a mid-pagination failure + # returns the error verbatim, and the items from the first page are + # dropped, not surfaced. + self.fetch_json.side_effect = [ + (True, [{'id': 1}]), + (False, 'HTTP 500'), + ] + success, body = bexio.get_all(API_TOKEN, '/2.0/contact') + self.assertFalse(success) + self.assertEqual(body, 'HTTP 500') + self.assertEqual(self.fetch_json.call_count, 2) + + def test_authorization_header_present_on_paginated_calls(self): + # get_all funnels through call_api, but pin it once: every page must + # carry the Bearer token, not just the first. + self.fetch_json.side_effect = [ + (True, [{'id': 1}]), + (True, []), + ] + bexio.get_all(API_TOKEN, '/2.0/contact') + for call in self.fetch_json.call_args_list: + self.assertEqual( + call.kwargs['header']['Authorization'], f'Bearer {API_TOKEN}' + ) + + +# --------------------------------------------------------------------------- +# Thin CRUD wrappers - one representative per category, plus the wrappers +# whose argument handling is non-trivial. +# --------------------------------------------------------------------------- + + +class TestThinWrappers(BexioTestBase): + def test_fetch_accounts_targets_account_url(self): + # Representative for all read-only fetch_* wrappers that just delegate + # to get_all(api_token, BEXIO_API_*_URL). + self.fetch_json.return_value = (True, []) + bexio.fetch_accounts(API_TOKEN) + self.assertTrue( + self.sent_url().startswith( + f'{bexio.BEXIO_API_BASE_URL}{bexio.BEXIO_API_ACCOUNT_URL}?' + ) + ) + + def test_fetch_contacts_default_archived_false(self): + self.fetch_json.return_value = (True, []) + bexio.fetch_contacts(API_TOKEN) + self.assertEqual(_query(self.sent_url())['show_archived'], 'False') + + def test_fetch_contacts_archived_true(self): + self.fetch_json.return_value = (True, []) + bexio.fetch_contacts(API_TOKEN, archived=True) + self.assertEqual(_query(self.sent_url())['show_archived'], 'True') + + def test_create_contact_posts_to_contact_url_with_data(self): + # Representative for all create_* wrappers. + payload = {'name_1': 'Linuxfabrik', 'contact_type_id': 1} + self.fetch_json.return_value = (True, {'id': 42}) + bexio.create_contact(API_TOKEN, data=payload) + self.assertEqual( + self.sent_url(), + f'{bexio.BEXIO_API_BASE_URL}{bexio.BEXIO_API_CONTACT_URL}', + ) + self.assertEqual(self.sent_kwargs()['data'], payload) + + def test_edit_contact_appends_id_to_url(self): + # Representative for all edit_* wrappers. + self.fetch_json.return_value = (True, {'id': 42}) + bexio.edit_contact(API_TOKEN, 42, data={'name_1': 'Updated'}) + self.assertEqual( + self.sent_url(), + f'{bexio.BEXIO_API_BASE_URL}{bexio.BEXIO_API_CONTACT_URL}/42', + ) + + def test_edit_contact_int_id_stringified(self): + # The wrapper does str(contact_id); pin that an int survives the + # concatenation without a TypeError. + self.fetch_json.return_value = (True, {}) + bexio.edit_contact(API_TOKEN, 42, data={}) + self.assertTrue(self.sent_url().endswith('/42')) + + def test_delete_contact_relation_uses_delete_method(self): + # The only wrapper that overrides the HTTP method; high-value test. + self.fetch_json.return_value = (True, {}) + bexio.delete_contact_relation(API_TOKEN, 7) + self.assertEqual(self.sent_kwargs()['method'], 'DELETE') + + def test_delete_contact_relation_url_has_id(self): + self.fetch_json.return_value = (True, {}) + bexio.delete_contact_relation(API_TOKEN, 7) + self.assertEqual( + self.sent_url(), + f'{bexio.BEXIO_API_BASE_URL}{bexio.BEXIO_API_CONTACT_RELATION_URL}/7', + ) + + def test_url_constants_match_documented_endpoints(self): + # Pin the URL constants. Catches accidental edits and makes the + # version-pinning of the Bexio API surface explicit in one place. + self.assertEqual(bexio.BEXIO_API_BASE_URL, 'https://api.bexio.com') + self.assertEqual(bexio.BEXIO_API_ACCOUNT_URL, '/2.0/accounts') + self.assertEqual(bexio.BEXIO_API_BANK_ACCOUNT_URL, '/3.0/banking/accounts') + self.assertEqual(bexio.BEXIO_API_BUSINESS_ACTIVITY_URL, '/2.0/client_service') + self.assertEqual(bexio.BEXIO_API_CONTACT_URL, '/2.0/contact') + self.assertEqual(bexio.BEXIO_API_CONTACT_GROUP_URL, '/2.0/contact_group') + self.assertEqual(bexio.BEXIO_API_CONTACT_RELATION_URL, '/2.0/contact_relation') + self.assertEqual(bexio.BEXIO_API_CONTACT_SECTOR_URL, '/2.0/contact_branch') + self.assertEqual(bexio.BEXIO_API_COUNTRY_URL, '/2.0/country') + self.assertEqual(bexio.BEXIO_API_CURRENCY_URL, '/3.0/currencies') + self.assertEqual(bexio.BEXIO_API_INVOICE_URL, '/2.0/kb_invoice') + self.assertEqual(bexio.BEXIO_API_ITEM_URL, '/2.0/article') + self.assertEqual(bexio.BEXIO_API_LANGUAGE_URL, '/2.0/language') + self.assertEqual(bexio.BEXIO_API_SALUTATION_URL, '/2.0/salutation') + self.assertEqual(bexio.BEXIO_API_PAYMENT_TYPE_URL, '/2.0/payment_type') + self.assertEqual(bexio.BEXIO_API_PROJECT_STATUS_URL, '/2.0/pr_project_state') + self.assertEqual(bexio.BEXIO_API_PROJECT_TYPE_URL, '/2.0/pr_project_type') + self.assertEqual(bexio.BEXIO_API_PROJECT_URL, '/2.0/pr_project') + self.assertEqual(bexio.BEXIO_API_STOCK_AREA_URL, '/2.0/stock_place') + self.assertEqual(bexio.BEXIO_API_STOCK_LOCATION_URL, '/2.0/stock') + self.assertEqual(bexio.BEXIO_API_TAX_URL, '/3.0/taxes') + self.assertEqual(bexio.BEXIO_API_TIMESHEET_URL, '/2.0/timesheet') + self.assertEqual(bexio.BEXIO_API_TIMESHEET_STATUS_URL, '/2.0/timesheet_status') + self.assertEqual(bexio.BEXIO_API_TITLE_URL, '/2.0/title') + self.assertEqual(bexio.BEXIO_API_UNIT_URL, '/2.0/unit') + self.assertEqual(bexio.BEXIO_API_USER_URL, '/3.0/users') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/url/unit-test/run b/tests/url/unit-test/run index 91eb600..92473b6 100755 --- a/tests/url/unit-test/run +++ b/tests/url/unit-test/run @@ -697,6 +697,94 @@ class TestFetchErrors(unittest.TestCase): self.assertIn('while fetching', msg) +# ----------------------------------------------------------------------------- +# fetch(): response_on_error +# ----------------------------------------------------------------------------- + + +@skip_without_httpx +class TestFetchResponseOnError(unittest.TestCase): + """The flag exists for APIs (e.g. Bexio) that return machine-readable error + details in the HTTP error body. Default behaviour is unchanged; opting in + swaps the formatted error message for the response body and (in extended + mode) the full response dict. + """ + + def _http_status_error(self, code): + req = httpx.Request('GET', 'https://example.com') + resp = httpx.Response(code, request=req) + return httpx.HTTPStatusError('err', request=req, response=resp) + + def test_default_false_keeps_formatted_message(self): + # Regression guard for the unchanged default path. + resp = FakeStreamResponse( + body=b'{"error":"bad"}', + status_code=400, + raise_exc=self._http_status_error(400), + ) + fake, _ = make_fake_client(resp) + with patch_client(fake): + ok, msg = url.fetch('https://example.com') + self.assertFalse(ok) + self.assertIn('HTTP error', msg) + self.assertIn('400', msg) + self.assertIn('Bad Request', msg) + + def test_true_returns_decoded_body(self): + resp = FakeStreamResponse( + body=b'{"error":"validation failed"}', + status_code=400, + raise_exc=self._http_status_error(400), + ) + fake, _ = make_fake_client(resp) + with patch_client(fake): + ok, body = url.fetch('https://example.com', response_on_error=True) + self.assertFalse(ok) + self.assertEqual(body, '{"error":"validation failed"}') + + def test_true_extended_returns_full_dict(self): + # Pins that status_code, response_header and timings.total are populated + # on the error path. + resp = FakeStreamResponse( + body=b'{"error":"x"}', + status_code=422, + headers={'Content-Type': 'application/json', 'X-Req-Id': 'abc'}, + elapsed_seconds=0.75, + raise_exc=self._http_status_error(422), + ) + fake, _ = make_fake_client(resp) + with no_timing_transport(), patch_client(fake): + ok, res = url.fetch( + 'https://example.com', + extended=True, + response_on_error=True, + ) + self.assertFalse(ok) + self.assertEqual(res['response'], '{"error":"x"}') + self.assertEqual(res['status_code'], 422) + self.assertEqual(res['response_header']['x-req-id'], 'abc') + self.assertEqual(res['timings']['total'], 0.75) + + def test_true_only_affects_http_status_error(self): + # A transport-level failure (ConnectError) must still surface as the + # formatted "URL error" message, not as a body or extended dict. + fake, _ = make_fake_client(stream_exc=httpx.ConnectError('refused')) + with patch_client(fake): + ok, msg = url.fetch('https://example.com', response_on_error=True) + self.assertFalse(ok) + self.assertIn('URL error', msg) + self.assertIn('refused', msg) + + def test_true_success_response_unchanged(self): + # A 2xx response with response_on_error=True behaves like the default. + resp = FakeStreamResponse(body=b'ok', status_code=200) + fake, _ = make_fake_client(resp) + with patch_client(fake): + ok, body = url.fetch('https://example.com', response_on_error=True) + self.assertTrue(ok) + self.assertEqual(body, 'ok') + + # ----------------------------------------------------------------------------- # fetch_json() # ----------------------------------------------------------------------------- @@ -795,6 +883,7 @@ class TestFetchJson(unittest.TestCase): tls_min='1.2', tls_max='1.3', method='POST', + response_on_error=True, ) _, kwargs = m.call_args self.assertTrue(kwargs['insecure']) @@ -809,6 +898,22 @@ class TestFetchJson(unittest.TestCase): self.assertEqual(kwargs['tls_min'], '1.2') self.assertEqual(kwargs['tls_max'], '1.3') self.assertEqual(kwargs['method'], 'POST') + self.assertTrue(kwargs['response_on_error']) + + def test_response_on_error_defaults_to_false(self): + with mock.patch.object(url, 'fetch', return_value=(True, '{}')) as m: + url.fetch_json('https://example.com') + self.assertFalse(m.call_args.kwargs['response_on_error']) + + def test_response_on_error_failure_body_returned_unparsed(self): + # When fetch() returns (False, body) because of response_on_error=True, + # fetch_json() forwards the raw body without attempting json.loads(). + # Bexio relies on this to inspect error details verbatim. + body = '{"error":{"code":"invalid_data"}}' + with mock.patch.object(url, 'fetch', return_value=(False, body)): + ok, res = url.fetch_json('https://example.com', response_on_error=True) + self.assertFalse(ok) + self.assertEqual(res, body) # ----------------------------------------------------------------------------- diff --git a/tox.ini b/tox.ini index 2563caf..8fcd6c7 100644 --- a/tox.ini +++ b/tox.ini @@ -16,5 +16,6 @@ no_package = true deps = psutil xmltodict + httpx[http2] commands = python tools/run-unit-tests --no-container {posargs} diff --git a/url.py b/url.py index 29c68ff..f72333d 100644 --- a/url.py +++ b/url.py @@ -11,7 +11,7 @@ """Get for example HTML or JSON from an URL.""" __author__ = 'Linuxfabrik GmbH, Zurich/Switzerland' -__version__ = '2026060701' +__version__ = '2026061901' import base64 import json @@ -281,6 +281,7 @@ def fetch( tls_min=None, tls_max=None, method=None, + response_on_error=False, ): """ Fetch any URL with optional POST, basic/digest authentication and SSL/TLS handling. @@ -306,8 +307,8 @@ def fetch( | |--> client.stream(method, url, ...) | |--> Capture TLS metadata from network stream - | |--> raise_for_status() on 4xx/5xx | |--> Read body + | |--> raise_for_status() on 4xx/5xx | |--> Decode body via response charset (default UTF-8) | @@ -359,6 +360,9 @@ def fetch( system default (typically TLS 1.2 on modern OpenSSL). - **tls_max** (`str`, optional): Maximum TLS version, same accepted values as `tls_min`. + - **response_on_error** (`bool`, optional): + If true, return the response for error conditions (useful when the response body of + an API contains error details) ### Returns - **tuple**: @@ -374,6 +378,7 @@ def fetch( - `alpn`: str like `'h2'` or `'http/1.1'` or None - `peer_cert_der`: DER-encoded server certificate as bytes, or None - On failure, an error message string. + - On failure with `response_on_error=True`, the response body. ### Example >>> result = fetch( @@ -492,12 +497,17 @@ def fetch( elapsed_seconds = 0.0 response_charset = None + success = True try: # No parenthesized context managers here: they are Python 3.10+ syntax and # break `import lib.url` on RHEL 8's default Python 3.6. + # fmt: off with client, client.stream(method, url, headers=headers, content=body) as response: + # fmt: on tls_version, alpn, peer_cert_der = _capture_tls_info(response) - response.raise_for_status() + # Read body and capture metadata before raise_for_status() so the + # response_on_error path can surface error bodies, status codes and + # timings to the caller (when using response_on_error). body_bytes = response.read() status_code = response.status_code # HTTP header field names are case-insensitive (RFC 9110, section 5.1). @@ -509,11 +519,15 @@ def fetch( } elapsed_seconds = response.elapsed.total_seconds() response_charset = response.charset_encoding + response.raise_for_status() except httpx.HTTPStatusError as e: - return False, ( - f'HTTP error "{e.response.status_code} {e.response.reason_phrase}"' - f' while fetching {url_safe}' - ) + if not response_on_error: + return False, ( + f'HTTP error "{e.response.status_code} {e.response.reason_phrase}"' + f' while fetching {url_safe}' + ) + else: + success = False except httpx.HTTPError as e: return False, f'URL error "{e}" for {url_safe}' except TypeError as e: @@ -526,12 +540,12 @@ def fetch( body_decoded = body_bytes.decode(charset) if to_text else body_bytes if not extended: - return True, body_decoded + return success, body_decoded timings = {'total': elapsed_seconds} if timing_backend is not None: timings.update(timing_backend.timings) - return True, { + return success, { 'response': body_decoded, 'status_code': status_code, 'response_header': response_headers, @@ -560,6 +574,7 @@ def fetch_json( tls_max=None, method=None, retries=0, + response_on_error=False, ): """ Fetch JSON from a URL with optional POST, authentication and SSL/TLS handling. @@ -605,6 +620,7 @@ def fetch_json( timeout=timeout, tls_max=tls_max, tls_min=tls_min, + response_on_error=response_on_error, ) if success: try: