feat(connectors): HubSpot checks follow the design's rows, copy and states

HubSpot follows the design:
- Names and order: the 9 checks are the design's 9 rows ("Sign in to
  HubSpot", "Read tickets", "List linked records", "Read record
  viewers"...). Dependencies follow the design's run order. "List linked
  records" and "Sample notes" run after the sign-in, not after the read
  checks: an empty type would stop them, but they can sample a record of
  another type. They leave out a denied type, which its read check
  reports.
- Every failure gives a short line and a description. A denial names the
  private app scope that HubSpot needs. Passing checks give what they
  found ("Signed in to account {id}", "Found {n} tickets", "{n} users
  found"...).
- Checks that passed without testing anything now say so. An empty type
  is SKIPPED. No record of any selected type FAILS the first read check.
  No record in scope skips the samples. A record without notes, and no
  listed user, are INCOMPLETE.

Fixes from the design's review:
- A rate limit past the retries is RateLimitedValidationError, which
  stops the run.
- Sign-in reports a missing token, an unreachable HubSpot and a timeout.
  On a form run, an unreachable HubSpot or a timeout fails sign-in.
- HubSpotApiError is a SourceApiError with the request and HubSpot's
  error body, so Error Details show them. The gateway raises
  HubSpotConnectionError for a failed or timed-out connection.
- Permission sync validation accepts a portal with no records.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KFMazQbuG8ui3jxBxm47sC
This commit is contained in:
Claude committed 2026-10-17 02:50:21 +00:00
1 parent 5216a53d26
commit 218d5ef709
8 files changed
+1141 -158

No files matched your search

@@ -18,6 +18,7 @@ from onyx.connectors.capability_checks.runner import run_capability_checks
from onyx.connectors.confluence.connector import ConfluenceConnector
from onyx.connectors.exceptions import (
ConnectorValidationError,
NothingToCheckError,
UnexpectedValidationError,
)
from onyx.connectors.factory import identify_connector_class
@@ -40,6 +41,9 @@ from onyx.connectors.sharepoint.capability_checks import (
)
from onyx.connectors.sharepoint.connector import SharepointConnector
from onyx.connectors.zoom.connector import ZoomConnector
from onyx.utils.logger import setup_logger
logger = setup_logger()
def validate_canvas_perm_sync(connector: CanvasConnector) -> None:
@@ -173,7 +177,11 @@ def validate_hubspot_perm_sync(connector: HubSpotConnector) -> None:
source_operations=connector.ops,
)
for check in build_hubspot_doc_permission_sync_checks():
check.run(context)
try:
check.run(context)
except NothingToCheckError as e:
# A portal with no records has no viewers to sync yet.
logger.info("HubSpot check %s found nothing to test: %s", check.check_id, e)
# The single source of truth for which connectors carry a real perm-sync probe:
@@ -1,28 +1,43 @@
"""Capability checks for the HubSpot connector.
Each check probes one permission the connector relies on through the gateway
operations production uses, reading at most one item per call. Checks need no
operations production uses, reading at most one page per call. Checks need no
connector instance: the runner builds the gateway from the credential, so they
run at credential creation too, over every object type until a config narrows
them. A probe with no record or note to call on passes, since there is no id
to exercise the permission with.
them.
The credential is a private app access token, so every denial names the
private app scope that HubSpot needs. A check with no record or note to probe
says so: a full read that finds nothing is SKIPPED, a sample that finds
nothing is INCOMPLETE. When no selected type has a record, the connector would
index nothing, so the first type's read check FAILS.
"""
from datetime import datetime, timezone
from typing import NoReturn
from pydantic import BaseModel
from onyx.connectors.capability_checks.form_state import FormState
from onyx.connectors.capability_checks.models import (
CapabilityCheck,
CapabilityCheckContext,
CheckOutput,
CredentialCapability,
form_config,
)
from onyx.connectors.cross_connector_utils.rate_limit_wrapper import (
RateLimitTriedTooManyTimesError,
)
from onyx.connectors.exceptions import (
ConnectorValidationError,
CredentialInvalidError,
InconclusiveValidationError,
InsufficientPermissionsError,
NothingToCheckError,
RateLimitedValidationError,
UnexpectedValidationError,
UnreachableValidationError,
)
from onyx.connectors.hubspot.config import (
HUBSPOT_OBJECT_SPECS,
@@ -33,9 +48,12 @@ from onyx.connectors.hubspot.connector import ASSOCIATED_TYPES, HS_OBJECT_ID_PRO
from onyx.connectors.hubspot.models import HubSpotPage, HubSpotRecord, HubSpotUser
from onyx.connectors.hubspot.source_operations import (
NOTES_OBJECT_TYPE,
USERS_PAGE_SIZE,
HubSpotApiError,
HubSpotConnectionError,
HubSpotSourceOperations,
)
from onyx.connectors.models import ConnectorMissingCredentialError
from onyx.connectors.source_operations import SourceOperations
from onyx.utils.logger import setup_logger
@@ -49,6 +67,40 @@ _USERS_SCOPE = "settings.users.read"
_INDEXED_TYPES_REMEDIATION = (
"Grant the private app the read scope of every object type the connector indexes."
)
_TOKEN_CHECK_ID = "hubspot_token"
_USERS_CHECK_ID = "hubspot_users"
# What the gateway raises for a call HubSpot refused or never answered.
_GATEWAY_ERRORS = (
HubSpotApiError,
HubSpotConnectionError,
RateLimitTriedTooManyTimesError,
)
def _read_check_id(object_type: HubSpotObjectType) -> str:
return f"hubspot_{object_type.value}_readable"
# check_id to the checks it runs after. The linked-record and notes checks run
# after the sign-in only: they sample any readable type, so an empty or denied
# type does not stop them.
_DEPENDS_ON: dict[str, tuple[str, ...]] = {
**{
_read_check_id(object_type): (_TOKEN_CHECK_ID,)
for object_type in HubSpotObjectType
},
"hubspot_associations": (_TOKEN_CHECK_ID,),
"hubspot_notes": (_TOKEN_CHECK_ID,),
_USERS_CHECK_ID: (_TOKEN_CHECK_ID,),
"hubspot_record_viewers": (_USERS_CHECK_ID,),
}
class _RowText(BaseModel):
"""The row text of one outcome of a probe."""
summary: str
description: str
def _gateway(context: CapabilityCheckContext) -> HubSpotSourceOperations:
@@ -59,115 +111,295 @@ def _gateway(context: CapabilityCheckContext) -> HubSpotSourceOperations:
def _object_types(config: HubSpotConnectorConfig) -> list[HubSpotObjectType]:
"""None means every type, as in the connector. [] means none."""
"""The selected types in registration order. None means every type, as in
the connector. [] means none."""
if config.object_types is None:
return list(HubSpotObjectType)
return config.object_types
return [
object_type
for object_type in HubSpotObjectType
if object_type in config.object_types
]
def _read_scope(object_type: HubSpotObjectType) -> str:
return HUBSPOT_OBJECT_SPECS[object_type].read_scope
def _refused(error: HubSpotApiError, denied: str, scope: str | None) -> NoReturn:
if error.status == 401:
raise CredentialInvalidError(
f"HubSpot rejected the access token when Onyx tried to {denied}."
def _noun(object_type: HubSpotObjectType) -> str:
return HUBSPOT_OBJECT_SPECS[object_type].document_noun
def _counted(count: int | str, noun: str, plural: str) -> str:
"""``count`` with its noun: "1 ticket", "3 tickets", "100+ users"."""
return f"{count} {noun if str(count) == '1' else plural}"
def _need(scope: str) -> str:
"""The line that says what the private app needs."""
return f"In HubSpot, add `{scope}` to the private app's scopes."
_REJECTED_ROW = _RowText(
summary="HubSpot rejected the API token",
description=(
"HubSpot does not accept this token. (HTTP 401)\nIn HubSpot, open the "
"private app and copy its access token."
),
)
_UNREACHABLE_ROW = _RowText(
summary="Cannot reach HubSpot",
description=(
"The request to HubSpot failed.\nCheck that the server can reach "
"api.hubapi.com."
),
)
_UNEXPECTED_ROW = _RowText(
summary="Unexpected response from HubSpot",
description=(
"HubSpot answered in a way that this check does not know.\nRe-run the checks."
),
)
def _raise_for_error(
error: Exception,
*,
denied: str,
scope: str | None = None,
denied_row: _RowText | None = None,
) -> NoReturn:
"""Maps a gateway error onto the validation-exception family. ``denied``
explains a 403 for this probe; ``denied_row`` is what its row shows."""
if isinstance(error, RateLimitTriedTooManyTimesError):
raise RateLimitedValidationError(
"HubSpot rate limited the check past the retries. Run the checks "
"again later.",
summary="Rate limited by HubSpot",
description=(
"HubSpot kept refusing requests after 5 retries. (HTTP 429)\nWait "
"a minute, then re-run."
),
) from error
if error.status == 403:
if isinstance(error, HubSpotConnectionError):
raise UnreachableValidationError(
f"Onyx cannot connect to HubSpot: {error}",
summary=_UNREACHABLE_ROW.summary,
description=_UNREACHABLE_ROW.description,
) from error
if not isinstance(error, HubSpotApiError):
raise error
status: int | None = error.status_code
if status == 401:
raise CredentialInvalidError(
f"HubSpot rejected the access token when Onyx tried to {denied}.",
summary=_REJECTED_ROW.summary,
description=_REJECTED_ROW.description,
) from error
if status == 403:
hint: str = f" Grant it the `{scope}` scope." if scope else ""
raise InsufficientPermissionsError(
f"The private app lacks permission to {denied}.{hint}"
f"The private app lacks permission to {denied}.{hint}",
summary=denied_row.summary if denied_row else None,
description=denied_row.description if denied_row else None,
) from error
raise UnexpectedValidationError(f"Could not {denied}: {error}") from error
if status in (404, 422):
raise ConnectorValidationError(
f"HubSpot refused the request to {denied}: {error}",
summary="HubSpot rejected the request",
description=(
f"HubSpot refused the request as sent. (HTTP {status})\nRe-run "
"the checks. If it repeats, share the error details with support."
),
) from error
raise UnexpectedValidationError(
f"Could not {denied}: {error}",
summary=_UNEXPECTED_ROW.summary,
description=_UNEXPECTED_ROW.description,
) from error
def _first_record(
context: CapabilityCheckContext,
) -> tuple[HubSpotObjectType, HubSpotRecord] | None:
"""The first record of the first configured type that has one. A refused
type raises, since the walk that needs a record lists every type."""
def _list_one(
gateway: HubSpotSourceOperations, object_type: HubSpotObjectType
) -> HubSpotPage[HubSpotRecord]:
return gateway.list_records(
variant=object_type, properties=[HS_OBJECT_ID_PROPERTY], limit=_PROBE_LIMIT
)
def _no_records_error() -> NothingToCheckError:
return NothingToCheckError(
"HubSpot has no record of the selected types, so there is nothing to probe.",
summary="No records in scope",
description="HubSpot has no record of the selected types.",
)
# A sample record and its type.
_Sample = tuple[HubSpotObjectType, HubSpotRecord]
def _first_readable_record(context: CapabilityCheckContext) -> _Sample:
"""The first record of the first selected type that lists one. A denied
type is left out: its read check reports it."""
gateway: HubSpotSourceOperations = _gateway(context)
denied: list[HubSpotObjectType] = []
for object_type in _object_types(form_config(context, HubSpotConnectorConfig)):
try:
page: HubSpotPage[HubSpotRecord] = gateway.list_records(
variant=object_type,
properties=[HS_OBJECT_ID_PROPERTY],
limit=_PROBE_LIMIT,
)
page: HubSpotPage[HubSpotRecord] = _list_one(gateway, object_type)
except HubSpotApiError as error:
_refused(
error,
f"list {object_type.value} for a sample record",
_read_scope(object_type),
if error.status_code != 403:
_raise_for_error(
error, denied=f"list {object_type.value} for a sample record"
)
logger.info("HubSpot denied %s, so the sample leaves it out", object_type)
denied.append(object_type)
continue
except _GATEWAY_ERRORS as error:
_raise_for_error(
error, denied=f"list {object_type.value} for a sample record"
)
if page.items:
return object_type, page.items[0]
logger.info("HubSpot has no records yet, so there is nothing to probe")
return None
if denied:
names: str = ", ".join(object_type.value for object_type in denied)
raise InconclusiveValidationError(
f"No readable type has a record, and HubSpot denied {names}, so Onyx "
"has no record to probe.",
summary="No readable records to test",
description=(
f"The readable types have no records, and HubSpot denied {names}."
"\nSee the Read rows."
),
)
raise _no_records_error()
def _sample_for_optional_check(
context: CapabilityCheckContext,
) -> tuple[HubSpotObjectType, HubSpotRecord] | None:
"""A refused listing is already the read check's failure, so an optional
check reports it as indeterminate instead of a second failed row."""
try:
return _first_record(context)
except ConnectorValidationError as error:
raise UnexpectedValidationError(f"No record to probe: {error}") from error
class _HubSpotCheck(CapabilityCheck[HubSpotConnectorConfig]):
config_class = HubSpotConnectorConfig
class _TokenCheck(CapabilityCheck):
def __init__(self) -> None:
def __init__(
self,
*,
check_id: str,
display_name: str,
capability: CredentialCapability = CredentialCapability.INDEXING,
required: bool = True,
remediation: str | None = None,
fails_when_unreachable: bool = False,
) -> None:
super().__init__(
capability=CredentialCapability.INDEXING,
check_id="hubspot_token",
display_name="Access token is valid",
capability=capability,
check_id=check_id,
display_name=display_name,
depends_on=_DEPENDS_ON.get(check_id, ()),
required=required,
requires_connector_instance=False,
remediation="Create a private app in HubSpot and paste its access token.",
remediation=remediation,
docs_link=_DOCS_LINK,
fails_when_unreachable=fails_when_unreachable,
)
def run(self, context: CapabilityCheckContext) -> None:
class _TokenCheck(_HubSpotCheck):
def __init__(self) -> None:
super().__init__(
check_id=_TOKEN_CHECK_ID,
display_name="Sign in to HubSpot",
remediation="Create a private app in HubSpot and paste its access token.",
fails_when_unreachable=True,
)
def run(self, context: CapabilityCheckContext) -> CheckOutput:
try:
_gateway(context).get_portal_id()
except HubSpotApiError as error:
_refused(error, "read the portal", None)
portal_id: str = _gateway(context).get_portal_id()
except ConnectorMissingCredentialError as error:
raise CredentialInvalidError(
"The credential has no HubSpot access token.",
summary="Account has no API token",
description=(
"The account has no HubSpot token.\nPaste the private app's "
"access token into the API Token field of the account."
),
) from error
except HubSpotConnectionError as error:
if not error.timed_out:
_raise_for_error(error, denied="read the portal")
raise UnreachableValidationError(
f"HubSpot did not answer the sign-in in time: {error}",
summary="Sign-in timed out",
description=(
"HubSpot did not answer the sign-in in time.\nRe-run, or check "
"that HubSpot is reachable from Onyx."
),
) from error
except ValueError as error:
# A body that is not HubSpot's portal info, e.g. a proxy page.
raise UnexpectedValidationError(
f"HubSpot's answer to the sign-in is not portal info: {error}",
summary=_UNEXPECTED_ROW.summary,
description=_UNEXPECTED_ROW.description,
) from error
except _GATEWAY_ERRORS as error:
_raise_for_error(
error,
denied="read the portal",
denied_row=_RowText(
summary="Token cannot read the account",
description=(
"HubSpot refused to name the account. (HTTP 403)\nIn "
"HubSpot, check that the private app is active, then copy "
"its access token again."
),
),
)
return CheckOutput(summary=f"Signed in to account {portal_id}")
class _RecordsReadableCheck(CapabilityCheck[HubSpotConnectorConfig]):
def _all_selected_types_empty(
context: CapabilityCheckContext, object_type: HubSpotObjectType
) -> bool:
"""True when every other selected type lists no record either. A type
that cannot be listed may have records, so it is not empty."""
gateway: HubSpotSourceOperations = _gateway(context)
for other in _object_types(form_config(context, HubSpotConnectorConfig)):
if other == object_type:
continue
try:
if _list_one(gateway, other).items:
return False
except RateLimitTriedTooManyTimesError as error:
_raise_for_error(error, denied=f"list {other.value}")
except _GATEWAY_ERRORS:
logger.debug("Cannot list HubSpot %s, so it may have records", other)
return False
return True
class _RecordsReadableCheck(_HubSpotCheck):
"""Listing, search and batch read of one type share one scope, so one check
probes all three, the batch read only when the listing returns a record."""
config_class = HubSpotConnectorConfig
def __init__(self, object_type: HubSpotObjectType) -> None:
self._object_type: HubSpotObjectType = object_type
super().__init__(
capability=CredentialCapability.INDEXING,
check_id=f"hubspot_{object_type.value}_readable",
display_name=f"{object_type.value.capitalize()} can be read",
requires_connector_instance=False,
check_id=_read_check_id(object_type),
display_name=f"Read {object_type.value}",
remediation=(
f"Grant the private app the `{_read_scope(object_type)}` scope."
),
docs_link=_DOCS_LINK,
)
def applies(self, form_state: FormState[HubSpotConnectorConfig]) -> bool:
return self._object_type in _object_types(form_state.config)
def run(self, context: CapabilityCheckContext) -> None:
def run(self, context: CapabilityCheckContext) -> CheckOutput:
gateway: HubSpotSourceOperations = _gateway(context)
object_type: HubSpotObjectType = self._object_type
scope: str = _read_scope(object_type)
try:
page: HubSpotPage[HubSpotRecord] = gateway.list_records(
variant=object_type,
properties=[HS_OBJECT_ID_PROPERTY],
limit=_PROBE_LIMIT,
)
gateway.search_records(
page: HubSpotPage[HubSpotRecord] = _list_one(gateway, object_type)
search: HubSpotPage[HubSpotRecord] = gateway.search_records(
variant=object_type,
properties=[HS_OBJECT_ID_PROPERTY],
modified_after=_EPOCH,
@@ -180,36 +412,75 @@ class _RecordsReadableCheck(CapabilityCheck[HubSpotConnectorConfig]):
ids=[page.items[0].id],
properties=[HS_OBJECT_ID_PROPERTY],
)
except HubSpotApiError as error:
_refused(error, f"read {object_type.value}", _read_scope(object_type))
except _GATEWAY_ERRORS as error:
_raise_for_error(
error,
denied=f"read {object_type.value}",
scope=scope,
denied_row=_RowText(
summary=f"Token cannot read {object_type.value}",
description=(
f"Indexing cannot list or search {object_type.value}. "
f"(HTTP 403)\n{_need(scope)}"
),
),
)
if not page.items:
self._raise_empty(context)
if search.total is None:
return CheckOutput(summary=f"Read {object_type.value}")
return CheckOutput(
summary=(
f"Found {_counted(search.total, _noun(object_type), object_type.value)}"
)
)
def _raise_empty(self, context: CapabilityCheckContext) -> NoReturn:
object_type: HubSpotObjectType = self._object_type
selected: list[HubSpotObjectType] = _object_types(
form_config(context, HubSpotConnectorConfig)
)
# An empty scope fails the first check that reads it: the connector
# would index nothing.
if selected[0] == object_type and _all_selected_types_empty(
context, object_type
):
raise ConnectorValidationError(
"HubSpot has no record of any selected object type, so the "
"connector would index nothing.",
summary="No records in the selected types",
description=(
"HubSpot has no record of any selected type, so indexing "
"would find nothing.\nAdd records in HubSpot, or check the "
"selected object types."
),
)
raise NothingToCheckError(
f"HubSpot has no {object_type.value}, so Onyx could not read one.",
summary=f"No {object_type.value} in HubSpot",
description=(
f"Listing and search work, and HubSpot has no {object_type.value}."
),
)
class _AssociationsCheck(CapabilityCheck[HubSpotConnectorConfig]):
class _AssociationsCheck(_HubSpotCheck):
"""Documents fold in associated records, so the probe lists one record's
links to another type. Optional, since indexing logs and skips associations
it cannot read."""
config_class = HubSpotConnectorConfig
def __init__(self) -> None:
super().__init__(
capability=CredentialCapability.INDEXING,
check_id="hubspot_associations",
display_name="Associated records can be listed",
requires_connector_instance=False,
display_name="List linked records",
required=False,
remediation=_INDEXED_TYPES_REMEDIATION,
docs_link=_DOCS_LINK,
)
def run(self, context: CapabilityCheckContext) -> None:
sample: tuple[HubSpotObjectType, HubSpotRecord] | None = (
_sample_for_optional_check(context)
)
if sample is None:
return
object_type, record = sample
def run(self, context: CapabilityCheckContext) -> CheckOutput:
object_type, record = _first_readable_record(context)
to_object_type: HubSpotObjectType = ASSOCIATED_TYPES[object_type][0]
to_scope: str = _read_scope(to_object_type)
try:
_gateway(context).list_associations(
object_type=object_type,
@@ -217,37 +488,39 @@ class _AssociationsCheck(CapabilityCheck[HubSpotConnectorConfig]):
to_object_type=to_object_type.value,
limit=_PROBE_LIMIT,
)
except HubSpotApiError as error:
_refused(
except _GATEWAY_ERRORS as error:
_raise_for_error(
error,
f"list the {to_object_type.value} linked to {object_type.value}",
_read_scope(to_object_type),
denied=(
f"list the {to_object_type.value} linked to {object_type.value}"
),
scope=to_scope,
denied_row=_RowText(
summary=f"Cannot read linked {to_object_type.value}",
description=(
"Documents would index without their linked records. "
f"(HTTP 403)\n{_need(to_scope)}"
),
),
)
return CheckOutput(
summary=f"Listed {to_object_type.value} linked to {object_type.value}"
)
class _NotesCheck(CapabilityCheck[HubSpotConnectorConfig]):
class _NotesCheck(_HubSpotCheck):
"""Optional, since indexing logs and skips notes it cannot read."""
config_class = HubSpotConnectorConfig
def __init__(self) -> None:
super().__init__(
capability=CredentialCapability.INDEXING,
check_id="hubspot_notes",
display_name="Notes can be read",
requires_connector_instance=False,
display_name="Sample notes",
required=False,
remediation=_INDEXED_TYPES_REMEDIATION,
docs_link=_DOCS_LINK,
)
def run(self, context: CapabilityCheckContext) -> None:
sample: tuple[HubSpotObjectType, HubSpotRecord] | None = (
_sample_for_optional_check(context)
)
if sample is None:
return
object_type, record = sample
def run(self, context: CapabilityCheckContext) -> CheckOutput:
object_type, record = _first_readable_record(context)
gateway: HubSpotSourceOperations = _gateway(context)
try:
notes: HubSpotPage[str] = gateway.list_associations(
@@ -256,72 +529,140 @@ class _NotesCheck(CapabilityCheck[HubSpotConnectorConfig]):
to_object_type=NOTES_OBJECT_TYPE,
limit=_PROBE_LIMIT,
)
if not notes.items:
logger.info("The sample HubSpot record has no notes to probe")
return
gateway.read_records(
variant=NOTES_OBJECT_TYPE,
ids=notes.items,
properties=[_NOTE_BODY_PROPERTY],
if notes.items:
gateway.read_records(
variant=NOTES_OBJECT_TYPE,
ids=notes.items,
properties=[_NOTE_BODY_PROPERTY],
)
except _GATEWAY_ERRORS as error:
_raise_for_error(
error,
denied="read notes",
denied_row=_RowText(
summary="Token cannot read notes",
description=(
"Record documents would index without their notes. (HTTP "
"403)\nGrant the private app the read scope of each "
"selected object type."
),
),
)
except HubSpotApiError as error:
_refused(error, "read notes", None)
if not notes.items:
raise InconclusiveValidationError(
f"The sample HubSpot {_noun(object_type)} has no notes, so Onyx "
"could not read one.",
summary="No notes found to test",
description="The sample record has no notes.",
)
return CheckOutput(summary=f"Read a note linked to a {_noun(object_type)}")
class _UsersCheck(CapabilityCheck):
class _UsersCheck(_HubSpotCheck):
"""Viewers are user ids, so the sync maps them to emails through the user
listing and, for ids it leaves out, one user at a time."""
def __init__(self) -> None:
super().__init__(
capability=CredentialCapability.DOC_PERMISSION_SYNC,
check_id="hubspot_users",
display_name="Users can be listed",
requires_connector_instance=False,
check_id=_USERS_CHECK_ID,
display_name="List users",
remediation=f"Grant the private app the `{_USERS_SCOPE}` scope.",
docs_link=_DOCS_LINK,
)
def run(self, context: CapabilityCheckContext) -> None:
def run(self, context: CapabilityCheckContext) -> CheckOutput:
gateway: HubSpotSourceOperations = _gateway(context)
try:
page: HubSpotPage[HubSpotUser] = gateway.list_users(limit=_PROBE_LIMIT)
page: HubSpotPage[HubSpotUser] = gateway.list_users(limit=USERS_PAGE_SIZE)
if page.items:
gateway.get_user(user_id=page.items[0].id)
except HubSpotApiError as error:
_refused(error, "list users", _USERS_SCOPE)
except _GATEWAY_ERRORS as error:
_raise_for_error(
error,
denied="list users",
scope=_USERS_SCOPE,
denied_row=_RowText(
summary="Token cannot list users",
description=(
"Permission sync cannot turn viewer IDs into emails. (HTTP "
f"403)\n{_need(_USERS_SCOPE)}"
),
),
)
if not page.items:
raise InconclusiveValidationError(
"HubSpot listed no users, so Onyx could not read one.",
summary="No users found to test",
description="HubSpot listed no users of the account.",
)
count: int | str = (
f"{len(page.items)}+" if page.next_after is not None else len(page.items)
)
return CheckOutput(summary=f"{_counted(count, 'user', 'users')} found")
class _RecordViewersCheck(CapabilityCheck[HubSpotConnectorConfig]):
config_class = HubSpotConnectorConfig
class _RecordViewersCheck(_HubSpotCheck):
def __init__(self) -> None:
super().__init__(
capability=CredentialCapability.DOC_PERMISSION_SYNC,
check_id="hubspot_record_viewers",
display_name="Record viewers can be read",
requires_connector_instance=False,
docs_link=_DOCS_LINK,
display_name="Read record viewers",
)
def run(self, context: CapabilityCheckContext) -> None:
sample: tuple[HubSpotObjectType, HubSpotRecord] | None = _first_record(context)
if sample is None:
return
object_type, record = sample
def run(self, context: CapabilityCheckContext) -> CheckOutput:
object_type, record = self._sample(context)
gateway: HubSpotSourceOperations = _gateway(context)
try:
portal_id: str = gateway.get_portal_id()
except HubSpotApiError as error:
_refused(error, "read the portal", None)
try:
gateway.get_record_viewers(
viewers: dict[str, set[int]] = gateway.get_record_viewers(
portal_id=portal_id,
object_type=object_type,
record_ids=[record.id],
)
except HubSpotApiError as error:
_refused(error, "read a record's viewers", None)
except _GATEWAY_ERRORS as error:
_raise_for_error(
error,
denied="read a record's viewers",
denied_row=_RowText(
summary="Token cannot read record viewers",
description=(
"Permission sync cannot learn who may view each record. "
"(HTTP 403)"
),
),
)
count: int = len(viewers.get(record.id, set()))
return CheckOutput(
summary=(
f"{_counted(count, 'user', 'users')} can view a sample "
f"{_noun(object_type)}"
)
)
def _sample(self, context: CapabilityCheckContext) -> _Sample:
"""Permission sync reads every selected type, so a denied listing
fails here too."""
gateway: HubSpotSourceOperations = _gateway(context)
for object_type in _object_types(form_config(context, HubSpotConnectorConfig)):
scope: str = _read_scope(object_type)
try:
page: HubSpotPage[HubSpotRecord] = _list_one(gateway, object_type)
except _GATEWAY_ERRORS as error:
_raise_for_error(
error,
denied=f"list {object_type.value} for a sample record",
scope=scope,
denied_row=_RowText(
summary=f"Token cannot read {object_type.value}",
description=(
f"Permission sync cannot sync {object_type.value}. "
f"(HTTP 403)\n{_need(scope)}"
),
),
)
if page.items:
return object_type, page.items[0]
raise _no_records_error()
def build_hubspot_indexing_checks() -> list[CapabilityCheck]:
@@ -21,6 +21,8 @@ class HubSpotObjectSpec(BaseModel):
class HubSpotPage(BaseModel, Generic[_ItemT]):
items: list[_ItemT]
next_after: str | None = None
# The match count a search reports. None for other listings.
total: int | None = None
class HubSpotAssociationIds(BaseModel):
@@ -84,6 +86,8 @@ class ApiRecord(BaseModel):
class ApiRecordPage(BaseModel):
results: list[ApiRecord]
paging: ApiPaging | None = None
# Sent by the Search API only.
total: int | None = None
class ApiAssociation(BaseModel):
@@ -8,6 +8,7 @@ from datetime import datetime, timezone
from typing import Any, TypeVar
import requests
import urllib3
from hubspot import HubSpot
from hubspot.crm.associations.v4.exceptions import (
ApiException as AssociationsApiException,
@@ -29,6 +30,7 @@ from pydantic import BaseModel
from onyx.configs.app_configs import REQUEST_TIMEOUT_SECONDS
from onyx.configs.constants import DocumentSource
from onyx.connectors.capabilities import CredentialCapability
from onyx.connectors.exceptions import SourceApiError
from onyx.connectors.hubspot.config import HUBSPOT_OBJECT_SPECS, HubSpotObjectType
from onyx.connectors.hubspot.models import (
ApiActionPermittedUsers,
@@ -60,6 +62,9 @@ PORTAL_INFO_PATH = "/integrations/v1/me"
USERS_PATH = "/settings/users/2026-09"
PERMITTED_USERS_PATH = "/resource-permissions/v3/permitted-users"
VIEW_ACTION = "crm-object:VIEW"
# The SDK calls' paths, for error details.
_OBJECTS_PATH = "/crm/v3/objects"
_ASSOCIATIONS_PATH = "/crm/v4/objects"
CREDENTIAL_ACCESS_TOKEN = "hubspot_access_token"
HUBSPOT_PAGE_SIZE = 100
@@ -88,9 +93,10 @@ _ItemT = TypeVar("_ItemT")
QueryParams = dict[str, str | int | list[str]]
class HubSpotApiError(Exception):
class HubSpotApiError(SourceApiError):
"""A refused or failed HubSpot call. Carries the status and headers the
rate limiter reads, and the operation name for error messages."""
rate limiter reads, the operation name for error messages, and the
request and body for a check's Error Details."""
def __init__(
self,
@@ -98,11 +104,40 @@ class HubSpotApiError(Exception):
status: int | None,
headers: Mapping[str, str] | None,
detail: str,
*,
method: str | None = None,
url: str | None = None,
) -> None:
super().__init__(
f"HubSpot returned {status} for {operation}: {detail[:200]}",
status_code=status,
method=method,
url=url,
body=detail or None,
)
self.operation = operation
self.status = status
self.headers = headers
super().__init__(f"HubSpot returned {status} for {operation}: {detail[:200]}")
@property
def status(self) -> int | None:
# The rate limiter reads `status`, as on the SDK's own exceptions.
return self.status_code
class HubSpotConnectionError(Exception):
"""HubSpot did not answer a call: the connection failed or timed out."""
def __init__(self, operation: str, *, timed_out: bool, detail: str) -> None:
verb: str = "timed out" if timed_out else "failed"
super().__init__(f"The HubSpot {operation} {verb}: {detail}")
self.operation = operation
self.timed_out = timed_out
def _sdk_timed_out(error: urllib3.exceptions.HTTPError) -> bool:
if isinstance(error, urllib3.exceptions.MaxRetryError):
return isinstance(error.reason, urllib3.exceptions.TimeoutError)
return isinstance(error, urllib3.exceptions.TimeoutError)
def iter_pages(
@@ -158,6 +193,7 @@ def _record_page(page: Any) -> HubSpotPage[HubSpotRecord]:
return HubSpotPage(
items=[_record(item) for item in raw.results],
next_after=_next_after(raw.paging),
total=raw.total,
)
@@ -207,10 +243,19 @@ class HubSpotSourceOperations(SourceOperations):
NOTES_OBJECT_TYPE: crm.objects.notes,
}[variant]
def _sdk(self, operation: str, func: Callable[..., _T], **kwargs: Any) -> _T:
def _sdk(
self,
operation: str,
method: str,
path: str,
func: Callable[..., _T],
**kwargs: Any,
) -> _T:
"""Runs one SDK call under the rate limiter. The SDK sends no timeout
unless one is passed per call. An SDK refusal becomes a HubSpotApiError
so callers see one error type for SDK and REST calls."""
and a failed connection a HubSpotConnectionError, so callers see one
error type for SDK and REST calls. ``method`` and
``path`` name the request, which the SDK's error does not carry."""
kwargs.setdefault("_request_timeout", REQUEST_TIMEOUT_SECONDS)
def call() -> _T:
@@ -218,7 +263,16 @@ class HubSpotSourceOperations(SourceOperations):
return func(**kwargs)
except _SDK_API_EXCEPTIONS as e:
raise HubSpotApiError(
operation, e.status, e.headers, str(e.body or e.reason or "")
operation,
e.status,
e.headers,
str(e.body or e.reason or ""),
method=method,
url=f"{HUBSPOT_API_BASE}{path}",
) from e
except urllib3.exceptions.HTTPError as e:
raise HubSpotConnectionError(
operation, timed_out=_sdk_timed_out(e), detail=str(e)
) from e
return self._limiter().call(call)
@@ -227,17 +281,33 @@ class HubSpotSourceOperations(SourceOperations):
self, operation: str, model: type[_M], path: str, params: QueryParams
) -> _M:
"""Runs one REST call the SDK has no client for."""
url: str = f"{HUBSPOT_API_BASE}{path}"
def call() -> _M:
response: requests.Response = requests.get(
f"{HUBSPOT_API_BASE}{path}",
headers={"Authorization": f"Bearer {self._access_token()}"},
params=params,
timeout=REQUEST_TIMEOUT_SECONDS,
)
try:
response: requests.Response = requests.get(
url,
headers={"Authorization": f"Bearer {self._access_token()}"},
params=params,
timeout=REQUEST_TIMEOUT_SECONDS,
)
# Timeout first: a connect timeout is also a ConnectionError.
except requests.Timeout as e:
raise HubSpotConnectionError(
operation, timed_out=True, detail=str(e)
) from e
except requests.ConnectionError as e:
raise HubSpotConnectionError(
operation, timed_out=False, detail=str(e)
) from e
if response.status_code >= 400:
raise HubSpotApiError(
operation, response.status_code, response.headers, response.text
operation,
response.status_code,
response.headers,
response.text,
method="GET",
url=url,
)
return model.model_validate(response.json())
@@ -270,6 +340,8 @@ class HubSpotSourceOperations(SourceOperations):
has_more set when HubSpot paged them."""
page: Any = self._sdk(
f"{variant} listing",
"GET",
f"{_OBJECTS_PATH}/{variant}",
self._crm(variant).basic_api.get_page,
limit=limit,
properties=properties,
@@ -325,6 +397,8 @@ class HubSpotSourceOperations(SourceOperations):
)
page: Any = self._sdk(
f"{variant} search",
"POST",
f"{_OBJECTS_PATH}/{variant}/search",
self._crm(variant).search_api.do_search,
public_object_search_request=request,
)
@@ -346,6 +420,9 @@ class HubSpotSourceOperations(SourceOperations):
"""One page of ids of the records (or notes) linked to one record."""
page: Any = self._sdk(
f"{object_type} to {to_object_type} associations",
"GET",
f"{_ASSOCIATIONS_PATH}/{object_type}/{object_id}/associations/"
f"{to_object_type}",
self._api().crm.associations.v4.basic_api.get_page,
object_type=object_type.value,
object_id=object_id,
@@ -381,6 +458,8 @@ class HubSpotSourceOperations(SourceOperations):
)
response: Any = self._sdk(
f"{variant} batch read",
"POST",
f"{_OBJECTS_PATH}/{variant}/batch/read",
self._crm(variant).batch_api.read,
batch_read_input_simple_public_object_id=request,
)
@@ -25,7 +25,9 @@ from onyx.connectors.capability_checks.registry import get_capability_checks
from onyx.connectors.capability_checks.runner import run_capability_checks
from onyx.connectors.exceptions import (
CredentialInvalidError,
InconclusiveValidationError,
InsufficientPermissionsError,
NothingToCheckError,
UnexpectedValidationError,
)
from onyx.connectors.hubspot.capability_checks import (
@@ -36,6 +38,7 @@ from onyx.connectors.hubspot.config import HubSpotObjectType
from onyx.connectors.hubspot.connector import HubSpotConnector
from onyx.connectors.hubspot.models import HubSpotPage, HubSpotRecord, HubSpotUser
from onyx.connectors.hubspot.source_operations import (
USERS_PAGE_SIZE,
HubSpotApiError,
HubSpotSourceOperations,
)
@@ -166,13 +169,35 @@ def test_a_read_check_applies_only_to_configured_types(
assert applying == applies
def test_an_optional_check_reports_a_refused_listing_as_incomplete() -> None:
"""The read check already fails for the refused type, so the associations
check adds no second failed row."""
def _deny_tickets(
variant: HubSpotObjectType, **_: object
) -> HubSpotPage[HubSpotRecord]:
if variant == HubSpotObjectType.TICKETS:
raise _refusal("tickets listing", 403)
return HubSpotPage(items=[_record("1")])
def test_an_optional_check_samples_past_a_denied_type() -> None:
"""The read check already fails for the denied type, so the associations
check samples the next type instead of adding a second failed row."""
gateway: MagicMock = _gateway()
gateway.list_records.side_effect = _deny_tickets
_run("hubspot_associations", _context(gateway))
gateway.list_associations.assert_called_once_with(
object_type=HubSpotObjectType.COMPANIES,
object_id="1",
to_object_type="contacts",
limit=1,
)
def test_an_optional_check_with_every_type_denied_is_inconclusive() -> None:
gateway: MagicMock = _gateway()
gateway.list_records.side_effect = _refusal("tickets listing", 403)
with pytest.raises(UnexpectedValidationError, match="No record to probe"):
with pytest.raises(InconclusiveValidationError):
_run("hubspot_associations", _context(gateway))
@@ -212,18 +237,20 @@ def test_an_empty_portal_has_nothing_to_probe(check_id: str) -> None:
gateway: MagicMock = _gateway()
gateway.list_records.return_value = HubSpotPage(items=[])
_run(check_id, _context(gateway))
with pytest.raises(NothingToCheckError):
_run(check_id, _context(gateway))
assert gateway.list_records.call_count == len(HubSpotObjectType)
gateway.list_associations.assert_not_called()
gateway.get_record_viewers.assert_not_called()
def test_a_record_without_notes_reads_none() -> None:
def test_a_record_without_notes_is_inconclusive() -> None:
gateway: MagicMock = _gateway()
gateway.list_associations.return_value = HubSpotPage(items=[])
_run("hubspot_notes", _context(gateway))
with pytest.raises(InconclusiveValidationError):
_run("hubspot_notes", _context(gateway))
gateway.list_associations.assert_called_once_with(
object_type=HubSpotObjectType.TICKETS,
@@ -245,7 +272,7 @@ def test_perm_sync_validation_probes_viewers_on_the_configured_types() -> None:
validate_perm_sync(_connector(gateway))
gateway.list_users.assert_called_once_with(limit=1)
gateway.list_users.assert_called_once_with(limit=USERS_PAGE_SIZE)
gateway.get_record_viewers.assert_called_once_with(
portal_id="46399533",
object_type=HubSpotObjectType.DEALS,
@@ -259,3 +286,12 @@ def test_perm_sync_validation_fails_on_a_missing_users_scope() -> None:
with pytest.raises(InsufficientPermissionsError, match="settings.users.read"):
validate_perm_sync(_connector(gateway))
def test_perm_sync_validation_passes_on_an_empty_portal() -> None:
gateway: MagicMock = _gateway()
gateway.list_records.return_value = HubSpotPage(items=[])
validate_perm_sync(_connector(gateway))
gateway.get_record_viewers.assert_not_called()
@@ -0,0 +1,437 @@
"""The HubSpot checks' rows: the design's short lines and descriptions, the
run order, the form-run sign-in policy, the rate-limit stop, Error Details,
and what a check that tested nothing says."""
from collections.abc import Callable
from unittest.mock import MagicMock
from onyx.configs.constants import DocumentSource
from onyx.connectors.capability_checks.models import (
CapabilityCheck,
CapabilityCheckContext,
CapabilityCheckResult,
CapabilityCheckStatus,
CheckErrorKind,
CheckNotRunReason,
)
from onyx.connectors.capability_checks.runner import run_capability_checks
from onyx.connectors.cross_connector_utils.rate_limit_wrapper import (
RateLimitTriedTooManyTimesError,
)
from onyx.connectors.hubspot.capability_checks import (
build_hubspot_doc_permission_sync_checks,
build_hubspot_indexing_checks,
)
from onyx.connectors.hubspot.config import HubSpotObjectType
from onyx.connectors.hubspot.models import HubSpotPage, HubSpotRecord, HubSpotUser
from onyx.connectors.hubspot.source_operations import (
HubSpotApiError,
HubSpotConnectionError,
)
from onyx.connectors.models import ConnectorMissingCredentialError
from onyx.db.enums import AccessType
from tests.unit.onyx.connectors.hubspot.test_hubspot_capability_checks import (
_gateway,
_record,
_refusal,
)
_ALL_CHECKS: list[CapabilityCheck] = (
build_hubspot_indexing_checks() + build_hubspot_doc_permission_sync_checks()
)
_READ_TICKETS = "hubspot_tickets_readable"
def _run_all(
check_ids: list[str],
gateway: MagicMock,
object_types: list[str] | None = None,
*,
form_run: bool = False,
) -> dict[str, CapabilityCheckResult]:
checks: list[CapabilityCheck] = [
check for check in _ALL_CHECKS if check.check_id in check_ids
]
context = CapabilityCheckContext(
source=DocumentSource.HUBSPOT,
credential_json={"hubspot_access_token": "pat-na1-secret"},
connector_specific_config=(
None if object_types is None else {"object_types": object_types}
),
access_type=AccessType.SYNC,
source_operations=gateway,
)
return {
result.check_id: result
for result in run_capability_checks(checks, context, form_run=form_run)
}
def _run(
check_id: str,
gateway: MagicMock,
object_types: list[str] | None = None,
*,
form_run: bool = False,
) -> CapabilityCheckResult:
return _run_all([check_id], gateway, object_types, form_run=form_run)[check_id]
def _records_only_in(
*object_types: HubSpotObjectType,
) -> Callable[..., HubSpotPage[HubSpotRecord]]:
"""A list_records side effect: one record of each given type, none of
the others."""
def list_records(
variant: HubSpotObjectType, **_: object
) -> HubSpotPage[HubSpotRecord]:
return HubSpotPage(items=[_record("1")] if variant in object_types else [])
return list_records
# --- Sign in to HubSpot ---
def test_sign_in_names_the_account() -> None:
result = _run("hubspot_token", _gateway())
assert result.display_name == "Sign in to HubSpot"
assert result.status == CapabilityCheckStatus.PASSED
assert result.summary == "Signed in to account 46399533"
def test_a_rejected_token_says_so() -> None:
gateway = _gateway()
gateway.get_portal_id.side_effect = _refusal("portal info", 401)
result = _run("hubspot_token", gateway)
assert result.status == CapabilityCheckStatus.FAILED
assert result.error_kind == CheckErrorKind.REJECTED
assert result.summary == "HubSpot rejected the API token"
def test_a_denied_sign_in_says_the_token_cannot_read_the_account() -> None:
gateway = _gateway()
gateway.get_portal_id.side_effect = _refusal("portal info", 403)
result = _run("hubspot_token", gateway)
assert result.error_kind == CheckErrorKind.PERMISSION
assert result.summary == "Token cannot read the account"
def test_a_missing_token_says_so() -> None:
gateway = _gateway()
gateway.get_portal_id.side_effect = ConnectorMissingCredentialError("HubSpot")
result = _run("hubspot_token", gateway)
assert result.status == CapabilityCheckStatus.FAILED
assert result.summary == "Account has no API token"
def test_an_unreachable_hubspot_fails_only_a_form_run() -> None:
gateway = _gateway()
gateway.get_portal_id.side_effect = HubSpotConnectionError(
"portal info", timed_out=False, detail="refused"
)
form = _run("hubspot_token", gateway, form_run=True)
stored = _run("hubspot_token", gateway)
assert form.status == CapabilityCheckStatus.FAILED
assert stored.status == CapabilityCheckStatus.INCOMPLETE
assert form.summary == stored.summary == "Cannot reach HubSpot"
def test_a_sign_in_timeout_fails_only_a_form_run() -> None:
gateway = _gateway()
gateway.get_portal_id.side_effect = HubSpotConnectionError(
"portal info", timed_out=True, detail="read timed out"
)
form = _run("hubspot_token", gateway, form_run=True)
stored = _run("hubspot_token", gateway)
assert form.status == CapabilityCheckStatus.FAILED
assert stored.status == CapabilityCheckStatus.INCOMPLETE
assert form.summary == "Sign-in timed out"
def test_an_api_error_carries_the_request_and_body() -> None:
gateway = _gateway()
gateway.get_portal_id.side_effect = HubSpotApiError(
"portal info",
500,
None,
'{"status": "error"}',
method="GET",
url="https://api.hubapi.com/integrations/v1/me",
)
result = _run("hubspot_token", gateway)
assert result.status == CapabilityCheckStatus.INCOMPLETE
assert result.summary == "Unexpected response from HubSpot"
assert result.error_detail is not None
assert result.error_detail.status_code == 500
assert result.error_detail.request == "GET api.hubapi.com/integrations/v1/me"
assert result.error_detail.response_excerpt == '{"status": "error"}'
def test_a_rate_limit_is_incomplete_and_stops_the_run() -> None:
gateway = _gateway()
gateway.list_records.side_effect = RateLimitTriedTooManyTimesError("429")
results = _run_all(
["hubspot_token", _READ_TICKETS, "hubspot_companies_readable"], gateway
)
tickets = results[_READ_TICKETS]
assert tickets.status == CapabilityCheckStatus.INCOMPLETE
assert tickets.error_kind == CheckErrorKind.RATE_LIMITED
assert tickets.summary == "Rate limited by HubSpot"
assert (
results["hubspot_companies_readable"].not_run_reason
== CheckNotRunReason.RATE_LIMITED
)
def test_a_failed_sign_in_stops_the_checks_after_it() -> None:
gateway = _gateway()
gateway.get_portal_id.side_effect = _refusal("portal info", 401)
results = _run_all(
[
"hubspot_token",
_READ_TICKETS,
"hubspot_associations",
"hubspot_notes",
"hubspot_users",
"hubspot_record_viewers",
],
gateway,
)
for check_id in (
_READ_TICKETS,
"hubspot_associations",
"hubspot_notes",
"hubspot_users",
"hubspot_record_viewers",
):
assert results[check_id].status == CapabilityCheckStatus.SKIPPED
assert results[check_id].blocked_by == "hubspot_token"
gateway.list_records.assert_not_called()
# --- Read {type} ---
def test_a_read_check_counts_the_records() -> None:
gateway = _gateway()
gateway.search_records.return_value = HubSpotPage(items=[_record("1")], total=3)
result = _run(_READ_TICKETS, gateway)
assert result.display_name == "Read tickets"
assert result.summary == "Found 3 tickets"
def test_a_read_check_counts_one_record_in_the_singular() -> None:
gateway = _gateway()
gateway.search_records.return_value = HubSpotPage(items=[_record("1")], total=1)
assert _run("hubspot_companies_readable", gateway).summary == "Found 1 company"
def test_a_denied_read_names_the_scope() -> None:
gateway = _gateway()
gateway.list_records.side_effect = _refusal("deals listing", 403)
result = _run("hubspot_deals_readable", gateway)
assert result.status == CapabilityCheckStatus.FAILED
assert result.summary == "Token cannot read deals"
assert result.description is not None
assert result.description.endswith(
"In HubSpot, add `crm.objects.deals.read` to the private app's scopes."
)
def test_an_empty_type_is_skipped_when_another_has_records() -> None:
gateway = _gateway()
gateway.list_records.side_effect = _records_only_in(HubSpotObjectType.CONTACTS)
result = _run(_READ_TICKETS, gateway)
assert result.status == CapabilityCheckStatus.SKIPPED
assert result.not_run_reason == CheckNotRunReason.NOTHING_IN_SCOPE
assert result.summary == "No tickets in HubSpot"
def test_no_record_of_any_selected_type_fails_the_first_read() -> None:
gateway = _gateway()
gateway.list_records.return_value = HubSpotPage(items=[])
results = _run_all(
[_READ_TICKETS, "hubspot_companies_readable"],
gateway,
object_types=["companies", "tickets"],
)
first = results[_READ_TICKETS]
assert first.status == CapabilityCheckStatus.FAILED
assert first.summary == "No records in the selected types"
assert results["hubspot_companies_readable"].status == CapabilityCheckStatus.SKIPPED
# --- List linked records and Sample notes ---
def test_linked_records_name_the_types() -> None:
result = _run("hubspot_associations", _gateway())
assert result.display_name == "List linked records"
assert result.summary == "Listed contacts linked to tickets"
def test_linked_records_run_past_an_empty_type() -> None:
gateway = _gateway()
gateway.list_records.side_effect = _records_only_in(HubSpotObjectType.CONTACTS)
results = _run_all(
["hubspot_token", _READ_TICKETS, "hubspot_associations"], gateway
)
assert results[_READ_TICKETS].status == CapabilityCheckStatus.SKIPPED
associations = results["hubspot_associations"]
assert associations.status == CapabilityCheckStatus.PASSED
assert associations.summary == "Listed companies linked to contacts"
def test_a_denied_link_names_the_linked_type_and_scope() -> None:
gateway = _gateway()
gateway.list_associations.side_effect = _refusal("associations", 403)
result = _run("hubspot_associations", gateway)
assert result.status == CapabilityCheckStatus.FAILED
assert result.summary == "Cannot read linked contacts"
assert result.description is not None
assert "`crm.objects.contacts.read`" in result.description
def test_no_record_in_scope_skips_the_samples() -> None:
gateway = _gateway()
gateway.list_records.return_value = HubSpotPage(items=[])
results = _run_all(
["hubspot_associations", "hubspot_notes", "hubspot_record_viewers"], gateway
)
for result in results.values():
assert result.status == CapabilityCheckStatus.SKIPPED
assert result.not_run_reason == CheckNotRunReason.NOTHING_IN_SCOPE
assert result.summary == "No records in scope"
def test_notes_name_the_record_type() -> None:
result = _run("hubspot_notes", _gateway())
assert result.display_name == "Sample notes"
assert result.summary == "Read a note linked to a ticket"
def test_a_record_without_notes_is_incomplete() -> None:
gateway = _gateway()
gateway.list_associations.return_value = HubSpotPage(items=[])
result = _run("hubspot_notes", gateway)
assert result.status == CapabilityCheckStatus.INCOMPLETE
assert result.summary == "No notes found to test"
def test_denied_notes_say_so() -> None:
gateway = _gateway()
gateway.read_records.side_effect = _refusal("notes batch read", 403)
assert _run("hubspot_notes", gateway).summary == "Token cannot read notes"
# --- List users and Read record viewers ---
def test_users_are_counted() -> None:
result = _run("hubspot_users", _gateway())
assert result.display_name == "List users"
assert result.summary == "1 user found"
def test_a_full_page_of_users_is_a_lower_bound() -> None:
gateway = _gateway()
gateway.list_users.return_value = HubSpotPage(
items=[HubSpotUser(id=i) for i in range(100)], next_after="100"
)
assert _run("hubspot_users", gateway).summary == "100+ users found"
def test_denied_users_name_the_scope() -> None:
gateway = _gateway()
gateway.list_users.side_effect = _refusal("user listing", 403)
result = _run("hubspot_users", gateway)
assert result.summary == "Token cannot list users"
assert result.description is not None
assert "`settings.users.read`" in result.description
def test_viewers_are_counted_on_a_sample_record() -> None:
gateway = _gateway()
gateway.get_record_viewers.return_value = {"1": {7, 8}}
result = _run("hubspot_record_viewers", gateway)
assert result.display_name == "Read record viewers"
assert result.summary == "2 users can view a sample ticket"
def test_a_denied_sample_listing_fails_the_viewers_check() -> None:
gateway = _gateway()
gateway.list_records.side_effect = _refusal("tickets listing", 403)
result = _run("hubspot_record_viewers", gateway)
assert result.status == CapabilityCheckStatus.FAILED
assert result.summary == "Token cannot read tickets"
assert result.description is not None
assert result.description.startswith("Permission sync cannot sync tickets.")
def test_denied_viewers_say_so() -> None:
gateway = _gateway()
gateway.get_record_viewers.side_effect = _refusal("record viewers", 403)
result = _run("hubspot_record_viewers", gateway)
assert result.summary == "Token cannot read record viewers"
def test_failed_users_stop_the_viewers_check() -> None:
gateway = _gateway()
gateway.list_users.side_effect = _refusal("user listing", 403)
results = _run_all(["hubspot_users", "hubspot_record_viewers"], gateway)
viewers = results["hubspot_record_viewers"]
assert viewers.status == CapabilityCheckStatus.SKIPPED
assert viewers.not_run_reason == CheckNotRunReason.DEPENDENCY_FAILED
assert viewers.blocked_by == "hubspot_users"
gateway.get_record_viewers.assert_not_called()
@@ -22,7 +22,7 @@ def test_default_timeout_is_passed_to_the_sdk() -> None:
gateway, _ = _gateway()
sdk_fn = MagicMock(return_value="ok")
result = gateway._sdk("op", sdk_fn, key="value")
result = gateway._sdk("op", "GET", "/path", sdk_fn, key="value")
assert result == "ok"
sdk_fn.assert_called_once_with(
@@ -34,7 +34,7 @@ def test_explicit_timeout_is_preserved() -> None:
gateway, _ = _gateway()
sdk_fn = MagicMock(return_value="ok")
gateway._sdk("op", sdk_fn, _request_timeout=5)
gateway._sdk("op", "GET", "/path", sdk_fn, _request_timeout=5)
sdk_fn.assert_called_once_with(_request_timeout=5)
@@ -6,6 +6,8 @@ from typing import Any
from unittest.mock import MagicMock, patch
import pytest
import requests
import urllib3
from hubspot.crm.associations.v4.models import (
CollectionResponseMultiAssociatedObjectWithLabelForwardPaging,
MultiAssociatedObjectWithLabel,
@@ -31,6 +33,7 @@ from onyx.connectors.hubspot.source_operations import (
PERMITTED_USERS_BATCH_SIZE,
PERMITTED_USERS_PATH,
HubSpotApiError,
HubSpotConnectionError,
HubSpotSourceOperations,
iter_pages,
)
@@ -252,7 +255,7 @@ def test_notes_are_read_through_the_notes_client() -> None:
assert [i.id for i in request.inputs] == ["5"]
def test_an_sdk_refusal_carries_the_status_and_operation() -> None:
def test_an_sdk_refusal_carries_the_status_operation_and_request() -> None:
gateway, client = _gateway()
client.crm.deals.basic_api.get_page.side_effect = DealsApiException(
status=403, reason="Forbidden"
@@ -262,6 +265,81 @@ def test_an_sdk_refusal_carries_the_status_and_operation() -> None:
gateway.list_records(variant=HubSpotObjectType.DEALS, properties=["p"])
assert (refusal.value.status, refusal.value.operation) == (403, "deals listing")
assert (refusal.value.method, refusal.value.url, refusal.value.body) == (
"GET",
"https://api.hubapi.com/crm/v3/objects/deals",
"Forbidden",
)
def test_a_rest_refusal_carries_the_request_and_body() -> None:
gateway, _ = _gateway()
response = _response(403, {})
response.text = '{"category": "MISSING_SCOPES"}'
with (
patch(REQUESTS_GET, return_value=response),
pytest.raises(HubSpotApiError) as refusal,
):
gateway.list_users()
assert refusal.value.status_code == 403
assert refusal.value.method == "GET"
assert refusal.value.url == "https://api.hubapi.com/settings/users/2026-09"
assert refusal.value.body == '{"category": "MISSING_SCOPES"}'
@pytest.mark.parametrize(
"failure, timed_out",
[
(requests.ConnectTimeout("connect timed out"), True),
(requests.ConnectionError("refused"), False),
],
)
def test_a_rest_call_that_gets_no_answer_says_whether_it_timed_out(
failure: requests.RequestException, timed_out: bool
) -> None:
gateway, _ = _gateway()
with (
patch(REQUESTS_GET, side_effect=failure),
pytest.raises(HubSpotConnectionError) as error,
):
gateway.get_portal_id()
assert error.value.timed_out is timed_out
def test_an_sdk_call_that_times_out_says_so() -> None:
gateway, client = _gateway()
client.crm.deals.basic_api.get_page.side_effect = urllib3.exceptions.MaxRetryError(
MagicMock(),
"/crm/v3/objects/deals",
urllib3.exceptions.ReadTimeoutError(
MagicMock(), "/crm/v3/objects/deals", "read timed out"
),
)
with pytest.raises(HubSpotConnectionError) as error:
gateway.list_records(variant=HubSpotObjectType.DEALS, properties=["p"])
assert error.value.timed_out
def test_a_search_keeps_the_match_count() -> None:
gateway, client = _gateway()
client.crm.tickets.search_api.do_search.return_value = _sdk_page(
{"results": [_sdk_record("1")], "total": 42}
)
page = gateway.search_records(
variant=HubSpotObjectType.TICKETS,
properties=["p"],
modified_after=START,
modified_before=None,
)
assert page.total == 42
def test_the_credential_is_read_once() -> None: