fix(connectors): second-pass validation fixes and test coverage

Adversarial re-validation of all four connectors plus a combined-change
regression audit.

- bitbucket: stop declaring incremental sync (deletion reconciliation is
  disabled for incremental runs, so deleted files were never removed on the
  default code configuration); drop a wasted listing round-trip after the
  frontier drains; add 33 tests
- databricks: reject an explicit maxDocuments of 0, which meant unlimited;
  add 34 tests
- google-chat: correct the sender displayName documentation (user auth
  populates only name and type) and emit second-precision RFC-3339 in the
  message filter; 15 -> 24 tests
- workday: fix a crash when maxVersions is persisted as a number; refuse a
  configuration whose status filter Workday did not honor; cap the
  unresolved-name error; 18 -> 27 tests
- document the Google Chat service-account omission
- docs: list all four connectors and correct the connector count
This commit is contained in:
Waleed Latif
2026-08-20 14:17:04 -07:00
parent 6b9737fa85
commit d06ef0d666
12 changed files with 1738 additions and 65 deletions
@@ -14,23 +14,24 @@ Connectors continuously sync documents from external services into your knowledg
<Image src="/static/connectors/connectors-sources.png" alt="Connect Source picker showing a searchable list of available connectors including Airtable, Asana, Confluence, Discord, Dropbox, Fireflies, GitHub, and Gmail" width={800} height={500} />
Sim ships with 60 built-in connectors:
Sim ships with 64 built-in connectors:
| Category | Connectors |
|----------|-----------|
| **Productivity** | Notion, Confluence, Asana, Linear, Jira, Jira Service Management, Monday, Trello, ClickUp, Google Calendar, Google Sheets, Google Forms, Microsoft Excel, Typeform |
| **Cloud Storage** | Google Drive, Dropbox, OneDrive, SharePoint, Box, Amazon S3, SFTP |
| **Documents** | Google Docs, Google Slides, Mintlify, WordPress, Webflow, DocuSign |
| **Development** | GitHub, GitLab, Azure DevOps, Sentry |
| **Development** | GitHub, GitLab, Bitbucket, Azure DevOps, Sentry |
| **Communication** | Slack, Discord, Microsoft Teams, Google Chat, Reddit, X, YouTube |
| **Email** | Gmail, Outlook |
| **CRM** | HubSpot, Salesforce |
| **Support** | Intercom, ServiceNow, Zendesk, Zoho Desk |
| **Incident Management** | incident.io, Rootly, PagerDuty |
| **Data** | Airtable |
| **Data** | Airtable, Databricks |
| **Note-taking** | Obsidian |
| **Meetings** | Zoom, Google Meet, Gong, Grain, Granola, Fathom, Fireflies |
| **Recruiting** | Greenhouse, Ashby |
| **HR** | Workday Help |
| **Compliance** | Google Vault |
## Adding a Connector
@@ -58,6 +59,8 @@ Other connectors use **API keys** or **personal access tokens** instead. The set
| **PagerDuty** | REST API key from Integrations → API Access Keys |
| **SFTP** | Password or unencrypted private key, plus a required SHA-256 host key fingerprint (host, port, username, and root path are entered as config fields) |
| **Mintlify** | API key — optional for public documentation sites, which sync from `llms.txt` |
| **Databricks** | Personal access token from your workspace's user settings (the workspace host is entered as a config field) |
| **Workday Help** | Register an API client for integrations in your tenant, then enter the client secret and refresh token together as `clientSecret:refreshToken` (the client ID, tenant host, and tenant name are entered as config fields) |
<Callout type="info">
If you rotate an API key in the external service, update it in Sim as well — OAuth tokens refresh automatically, but API keys do not.
@@ -0,0 +1,664 @@
/**
* @vitest-environment node
*
* The load-bearing logic here is the repository tree walk: a `max_depth` listing
* that Bitbucket may answer with a `555` timeout, an opaque `next` cursor that
* cannot be re-cut, and a per-run frontier of directories the walk has not reached.
* Every path that shortens the listing has to leave `syncContext.listingCapped` set,
* because the sync engine hard-deletes whatever a full listing omits.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { bitbucketConnector } from '@/connectors/bitbucket/bitbucket'
const ACCESS_TOKEN = 'bitbucket-token'
const COMMIT = 'e7d158ff7ed5538c28f94cd97a9ad569680fc94e'
const CONFIG = { workspaceSlug: 'acme', repoSlug: 'widgets' }
const PR_CONFIG = { ...CONFIG, contentTypes: 'pullrequests' }
const mockFetch = vi.fn()
function jsonResponse(body: unknown, status = 200): Response {
return new Response(JSON.stringify(body), {
status,
headers: { 'Content-Type': 'application/json' },
})
}
const REPOSITORY = {
full_name: 'acme/widgets',
mainbranch: { name: 'main', target: { hash: COMMIT } },
links: { html: { href: 'https://bitbucket.org/acme/widgets' } },
}
function fileEntry(path: string, size = 10, attributes: string[] = []) {
return { type: 'commit_file', path, size, attributes, commit: { hash: COMMIT } }
}
function dirEntry(path: string) {
return { type: 'commit_directory', path, commit: { hash: COMMIT } }
}
function pullRequestFixture(id: number, overrides: Record<string, unknown> = {}) {
return {
id,
title: `PR ${id}`,
state: 'OPEN',
summary: { raw: `Body of ${id}`, markup: 'markdown', html: `<p>Body of ${id}</p>` },
author: { display_name: 'Ada Lovelace', nickname: 'ada' },
created_on: '2026-01-01T00:00:00.000000+00:00',
updated_on: '2026-02-01T00:00:00.000000+00:00',
links: { html: { href: `https://bitbucket.org/acme/widgets/pull-requests/${id}` } },
...overrides,
}
}
/** Routes by URL, because the number of lookups before a listing varies by phase. */
function mockApi(routes: Array<[RegExp, () => Response]>) {
mockFetch.mockImplementation(async (url: string) => {
for (const [pattern, respond] of routes) {
if (pattern.test(url)) return respond()
}
if (/\/repositories\/acme\/widgets$/.test(url)) return jsonResponse(REPOSITORY)
throw new Error(`unrouted request: ${url}`)
})
}
function requestedUrls(pattern: RegExp): string[] {
return mockFetch.mock.calls.map(([url]) => url as string).filter((url) => pattern.test(url))
}
beforeEach(() => {
vi.clearAllMocks()
vi.stubGlobal('fetch', mockFetch)
})
afterEach(() => {
vi.unstubAllGlobals()
})
describe('bitbucket repository file listing', () => {
it('lists text files as deferred stubs and drops binary, symlink, and submodule entries', async () => {
mockApi([
[
/\/src\//,
() =>
jsonResponse({
values: [
fileEntry('README.md'),
fileEntry('logo.png', 2048, ['binary']),
fileEntry('link.txt', 5, ['link']),
fileEntry('vendor', 40, ['subrepository']),
fileEntry('run.sh', 12, ['executable']),
],
}),
],
])
const syncContext: Record<string, unknown> = {}
const result = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
expect(result.documents.map((d) => d.externalId)).toEqual(['file:README.md', 'file:run.sh'])
expect(result.documents[0].contentDeferred).toBe(true)
expect(result.documents[0].content).toBe('')
expect(result.documents[0].contentHash).toBe(`bitbucket:file:acme/widgets:${COMMIT}:README.md`)
expect(result.hasMore).toBe(false)
expect(syncContext.listingCapped).toBeUndefined()
})
it('pins the listing to the resolved tip commit and asks for the documented depth', async () => {
mockApi([[/\/src\//, () => jsonResponse({ values: [] })]])
await bitbucketConnector.listDocuments(ACCESS_TOKEN, CONFIG, undefined, {})
const url = requestedUrls(/\/src\//)[0]
expect(url).toContain(`/src/${COMMIT}/`)
expect(url).toContain('max_depth=5')
expect(url).toContain('pagelen=100')
})
it('applies the path prefix and extension filters to files and prunes unrelated directories', async () => {
mockApi([
[
/\/src\//,
() =>
jsonResponse({
values: [
fileEntry('docs/guide.md'),
fileEntry('docs/guide.txt'),
fileEntry('src/index.md'),
dirEntry('docs/a/b/c/d'),
dirEntry('src/a/b/c/d'),
],
}),
],
])
const syncContext: Record<string, unknown> = {}
const result = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
{ ...CONFIG, pathPrefix: '/docs', fileExtensions: 'md' },
undefined,
syncContext
)
expect(result.documents.map((d) => d.externalId)).toEqual(['file:docs/guide.md'])
expect(syncContext.pendingDirs).toEqual(['docs/a/b/c/d'])
})
it('queues only the frontier directories the walk actually stopped at', async () => {
mockApi([
[
/\/src\//,
() =>
jsonResponse({
values: [dirEntry('a'), dirEntry('a/b'), dirEntry('a/b/c/d/e'), fileEntry('a/b/x.md')],
}),
],
])
const syncContext: Record<string, unknown> = {}
await bitbucketConnector.listDocuments(ACCESS_TOKEN, CONFIG, undefined, syncContext)
expect(syncContext.pendingDirs).toEqual(['a/b/c/d/e'])
})
it('walks the frontier on a following page and finishes without flagging the listing', async () => {
mockApi([
[
/\/src\/[a-f0-9]+\/\?/,
() => jsonResponse({ values: [dirEntry('a/b/c/d/e'), fileEntry('root.md')] }),
],
[
/\/src\/[a-f0-9]+\/a\/b\/c\/d\/e\/\?/,
() => jsonResponse({ values: [fileEntry('a/b/c/d/e/deep.md')] }),
],
])
const syncContext: Record<string, unknown> = {}
const first = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
expect(first.hasMore).toBe(true)
const second = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
first.nextCursor,
syncContext
)
expect(second.documents.map((d) => d.externalId)).toEqual(['file:a/b/c/d/e/deep.md'])
expect(second.hasMore).toBe(false)
expect(syncContext.listingCapped).toBeUndefined()
})
it('replays the opaque next link verbatim instead of rebuilding it', async () => {
const nextUrl =
'https://api.bitbucket.org/2.0/repositories/acme/widgets/src/abc/?page=2&opaque=xyz'
mockApi([
[/page=2/, () => jsonResponse({ values: [fileEntry('second.md')] })],
[/\/src\//, () => jsonResponse({ values: [fileEntry('first.md')], next: nextUrl })],
])
const syncContext: Record<string, unknown> = {}
const first = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
expect(first.hasMore).toBe(true)
await bitbucketConnector.listDocuments(ACCESS_TOKEN, CONFIG, first.nextCursor, syncContext)
expect(requestedUrls(/page=2/)).toEqual([nextUrl])
})
it('rejects a next cursor that does not point at the Bitbucket Cloud API', async () => {
mockApi([
[
/\/src\//,
() => jsonResponse({ values: [], next: 'https://evil.example.com/2.0/repositories/x' }),
],
])
const syncContext: Record<string, unknown> = {}
const first = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
await expect(
bitbucketConnector.listDocuments(ACCESS_TOKEN, CONFIG, first.nextCursor, syncContext)
).rejects.toThrow(/Bitbucket Cloud API/)
})
})
describe('bitbucket source listing timeouts', () => {
it('retries a first-page 555 at depth 1 and keeps the listing reconcilable', async () => {
mockApi([
[/max_depth=1/, () => jsonResponse({ values: [dirEntry('a'), fileEntry('root.md')] })],
[/max_depth=5/, () => new Response('timeout', { status: 555 })],
])
const syncContext: Record<string, unknown> = {}
const result = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
expect(result.documents.map((d) => d.externalId)).toEqual(['file:root.md'])
expect(syncContext.listingCapped).toBeUndefined()
/** The depth-1 response only reached one level, so `a` is now unexplored. */
expect(syncContext.pendingDirs).toEqual(['a'])
expect(result.hasMore).toBe(true)
})
it('flags the listing capped when even the depth-1 retry times out', async () => {
mockApi([[/\/src\//, () => new Response('timeout', { status: 555 })]])
const syncContext: Record<string, unknown> = {}
const result = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
expect(syncContext.listingCapped).toBe(true)
expect(result.documents).toEqual([])
expect(requestedUrls(/\/src\//)).toHaveLength(2)
})
it('flags the listing capped when a replayed next cursor times out, without re-cutting it', async () => {
const nextUrl =
'https://api.bitbucket.org/2.0/repositories/acme/widgets/src/abc/?page=2&max_depth=5'
mockApi([
[/page=2/, () => new Response('timeout', { status: 555 })],
[/\/src\//, () => jsonResponse({ values: [fileEntry('first.md')], next: nextUrl })],
])
const syncContext: Record<string, unknown> = {}
const first = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
const second = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
first.nextCursor,
syncContext
)
expect(syncContext.listingCapped).toBe(true)
expect(second.documents).toEqual([])
/** Exactly one attempt: an opaque cursor is never rebuilt at another depth. */
expect(requestedUrls(/page=2/)).toHaveLength(1)
})
it('leaves the listing reconcilable when a directory is simply absent', async () => {
mockApi([[/\/src\//, () => jsonResponse({ type: 'error' }, 404)]])
const syncContext: Record<string, unknown> = {}
const result = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
expect(syncContext.listingCapped).toBeUndefined()
expect(result.hasMore).toBe(false)
})
it('flags the listing capped when the token stops working mid-walk', async () => {
mockApi([[/\/src\//, () => jsonResponse({ type: 'error' }, 403)]])
const syncContext: Record<string, unknown> = {}
await bitbucketConnector.listDocuments(ACCESS_TOKEN, CONFIG, undefined, syncContext)
expect(syncContext.listingCapped).toBe(true)
})
})
describe('bitbucket maxItems cap', () => {
it('flags the listing capped when the cap truncates a page', async () => {
mockApi([
[
/\/src\//,
() => jsonResponse({ values: [fileEntry('a.md'), fileEntry('b.md'), fileEntry('c.md')] }),
],
])
const syncContext: Record<string, unknown> = {}
const result = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
{ ...CONFIG, maxItems: '2' },
undefined,
syncContext
)
expect(result.documents).toHaveLength(2)
expect(result.hasMore).toBe(false)
expect(syncContext.listingCapped).toBe(true)
})
it('leaves the listing reconcilable when no cap is configured', async () => {
mockApi([[/\/src\//, () => jsonResponse({ values: [fileEntry('a.md'), fileEntry('b.md')] })]])
const syncContext: Record<string, unknown> = {}
const result = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
expect(result.documents).toHaveLength(2)
expect(syncContext.listingCapped).toBeUndefined()
})
it('lets an oversized file ride along without consuming the cap', async () => {
mockApi([
[
/\/src\//,
() =>
jsonResponse({
values: [fileEntry('huge.md', 500 * 1024 * 1024), fileEntry('a.md'), fileEntry('b.md')],
}),
],
])
const syncContext: Record<string, unknown> = {}
const result = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
{ ...CONFIG, maxItems: '2' },
undefined,
syncContext
)
expect(result.documents.map((d) => d.externalId)).toEqual([
'file:huge.md',
'file:a.md',
'file:b.md',
])
expect(result.documents[0].skippedReason).toMatch(/size limit/)
})
})
describe('bitbucket pull request listing', () => {
it('builds the documented collection query', async () => {
mockApi([[/\/pullrequests/, () => jsonResponse({ values: [pullRequestFixture(7)] })]])
await bitbucketConnector.listDocuments(ACCESS_TOKEN, PR_CONFIG, undefined, {})
const url = requestedUrls(/\/pullrequests/)[0]
expect(url).toContain('pagelen=50')
/** `+` must reach Bitbucket percent-encoded or it is read as a space. */
expect(url).toContain('fields=%2Bvalues.summary')
expect(url).toContain('sort=-id')
expect(url).toContain('state=OPEN')
})
it('repeats the state parameter for each requested state', async () => {
mockApi([[/\/pullrequests/, () => jsonResponse({ values: [] })]])
await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
{ ...PR_CONFIG, pullRequestState: 'all' },
undefined,
{}
)
const url = requestedUrls(/\/pullrequests/)[0]
expect(url).toContain('state=OPEN')
expect(url).toContain('state=MERGED')
expect(url).toContain('state=DECLINED')
expect(url).toContain('state=SUPERSEDED')
})
it('renders the incremental BBQL filter with an explicit numeric UTC offset', async () => {
mockApi([[/\/pullrequests/, () => jsonResponse({ values: [] })]])
await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
PR_CONFIG,
undefined,
{},
new Date('2026-02-03T04:05:06.007Z')
)
const url = new URL(requestedUrls(/\/pullrequests/)[0])
expect(url.searchParams.get('q')).toBe('updated_on > 2026-02-03T04:05:06.007+00:00')
})
it('maps the author-typed body and falls back to the rendered description', async () => {
mockApi([
[
/\/pullrequests/,
() =>
jsonResponse({
values: [
pullRequestFixture(7),
pullRequestFixture(8, {
summary: undefined,
rendered: { description: { raw: 'Rendered body' } },
}),
],
}),
],
])
const result = await bitbucketConnector.listDocuments(ACCESS_TOKEN, PR_CONFIG, undefined, {})
expect(result.documents[0].content).toBe('PR 7\n\nBody of 7')
expect(result.documents[0].contentDeferred).toBe(false)
expect(result.documents[0].contentHash).toBe(
'bitbucket:pr:acme/widgets:7:2026-02-01T00:00:00.000000+00:00'
)
expect(result.documents[1].content).toBe('PR 8\n\nRendered body')
})
it('walks from the code phase into the pull request phase for combined content', async () => {
mockApi([
[/\/src\//, () => jsonResponse({ values: [fileEntry('a.md')] })],
[/\/pullrequests/, () => jsonResponse({ values: [pullRequestFixture(7)] })],
])
const syncContext: Record<string, unknown> = {}
const first = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
{ ...CONFIG, contentTypes: 'all' },
undefined,
syncContext
)
expect(first.hasMore).toBe(true)
const second = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
{ ...CONFIG, contentTypes: 'all' },
first.nextCursor,
syncContext
)
expect(second.documents.map((d) => d.externalId)).toEqual(['pr:7'])
expect(second.hasMore).toBe(false)
expect(syncContext.listingCapped).toBeUndefined()
})
})
describe('bitbucket getDocument', () => {
it('hydrates a file with the same contentHash the listing produced', async () => {
mockApi([[/\/src\//, () => jsonResponse({ values: [fileEntry('README.md')] })]])
const syncContext: Record<string, unknown> = {}
const listed = await bitbucketConnector.listDocuments(
ACCESS_TOKEN,
CONFIG,
undefined,
syncContext
)
mockApi([[/\/src\/[a-f0-9]+\/README\.md$/, () => new Response('# Widgets', { status: 200 })]])
const fetched = await bitbucketConnector.getDocument(
ACCESS_TOKEN,
CONFIG,
'file:README.md',
syncContext
)
expect(fetched?.contentHash).toBe(listed.documents[0].contentHash)
expect(fetched?.contentDeferred).toBe(false)
expect(fetched?.content).toBe('README.md\n\n# Widgets')
expect(fetched?.metadata?.size).toBe(9)
})
it('surfaces an LFS-managed file as skipped rather than following the media redirect', async () => {
mockApi([
[
/\/src\/[a-f0-9]+\/big\.bin$/,
() =>
new Response(null, {
status: 301,
headers: { Location: 'https://media.atlassian.com/file/abc' },
}),
],
])
const doc = await bitbucketConnector.getDocument(ACCESS_TOKEN, CONFIG, 'file:big.bin', {})
expect(doc?.skippedReason).toMatch(/LFS/)
expect(doc?.content).toBe('')
})
it('surfaces a file with a NUL byte as a skipped binary', async () => {
mockApi([
[
/\/src\/[a-f0-9]+\/data\.md$/,
() => new Response(Buffer.from([0x61, 0x00, 0x62]), { status: 200 }),
],
])
const doc = await bitbucketConnector.getDocument(ACCESS_TOKEN, CONFIG, 'file:data.md', {})
expect(doc?.skippedReason).toMatch(/Binary/)
})
it('returns null for a file the ref no longer carries', async () => {
mockApi([[/\/src\/[a-f0-9]+\/gone\.md$/, () => jsonResponse({ type: 'error' }, 404)]])
expect(
await bitbucketConnector.getDocument(ACCESS_TOKEN, CONFIG, 'file:gone.md', {})
).toBeNull()
})
it('rethrows a transient failure so the sync records it instead of reading it as a deletion', async () => {
mockApi([[/\/src\/[a-f0-9]+\/flaky\.md$/, () => jsonResponse({ type: 'error' }, 500)]])
await expect(
bitbucketConnector.getDocument(ACCESS_TOKEN, CONFIG, 'file:flaky.md', {})
).rejects.toThrow(/500/)
})
it('returns null for a pull request the repository no longer has', async () => {
mockApi([[/\/pullrequests\/9$/, () => jsonResponse({ type: 'error' }, 404)]])
expect(await bitbucketConnector.getDocument(ACCESS_TOKEN, CONFIG, 'pr:9', {})).toBeNull()
})
it('rejects an externalId with no recognized resource prefix', async () => {
mockApi([])
expect(await bitbucketConnector.getDocument(ACCESS_TOKEN, CONFIG, 'wiki:Home', {})).toBeNull()
})
})
describe('bitbucket tag mapping', () => {
it('maps every declared file tag from the listing metadata', async () => {
mockApi([[/\/src\//, () => jsonResponse({ values: [fileEntry('docs/guide.md', 42)] })]])
const result = await bitbucketConnector.listDocuments(ACCESS_TOKEN, CONFIG, undefined, {})
const tags = bitbucketConnector.mapTags?.(result.documents[0].metadata ?? {})
expect(tags).toEqual({
contentType: 'file',
title: 'guide.md',
repository: 'acme/widgets',
path: 'docs/guide.md',
size: 42,
})
})
it('maps every declared pull request tag from the listing metadata', async () => {
mockApi([[/\/pullrequests/, () => jsonResponse({ values: [pullRequestFixture(7)] })]])
const result = await bitbucketConnector.listDocuments(ACCESS_TOKEN, PR_CONFIG, undefined, {})
const tags = bitbucketConnector.mapTags?.(result.documents[0].metadata ?? {})
expect(tags).toEqual({
contentType: 'pull_request',
title: 'PR 7',
repository: 'acme/widgets',
state: 'OPEN',
author: 'Ada Lovelace',
createdAt: new Date('2026-01-01T00:00:00.000000+00:00'),
updatedAt: new Date('2026-02-01T00:00:00.000000+00:00'),
})
})
})
describe('bitbucket validateConfig', () => {
it('rejects a repository the credential cannot read', async () => {
mockApi([[/\/repositories\/acme\/widgets$/, () => jsonResponse({ type: 'error' }, 404)]])
expect(await bitbucketConnector.validateConfig(ACCESS_TOKEN, CONFIG)).toEqual({
valid: false,
error: 'Repository "acme/widgets" not found',
})
})
it('rejects a branch or tag that resolves to neither', async () => {
mockApi([[/\/refs\//, () => jsonResponse({ type: 'error' }, 404)]])
const result = await bitbucketConnector.validateConfig(ACCESS_TOKEN, {
...CONFIG,
ref: 'nope',
})
expect(result.valid).toBe(false)
expect(result.error).toMatch(/Branch or tag "nope" not found/)
})
it('accepts a tag when no branch by that name exists', async () => {
mockApi([
[/\/refs\/branches\//, () => jsonResponse({ type: 'error' }, 404)],
[/\/refs\/tags\//, () => jsonResponse({ name: 'v1.0', target: { hash: COMMIT } })],
])
expect(
await bitbucketConnector.validateConfig(ACCESS_TOKEN, { ...CONFIG, ref: 'v1.0' })
).toEqual({ valid: true })
})
it('rejects a non-positive maxItems before spending a request', async () => {
mockApi([])
expect(
await bitbucketConnector.validateConfig(ACCESS_TOKEN, { ...CONFIG, maxItems: '0' })
).toEqual({ valid: false, error: 'Max items must be a positive number' })
})
})
+13 -12
View File
@@ -774,11 +774,16 @@ export const bitbucketConnector: ConnectorConfig = {
/**
* Cursor that stays in the code phase to walk the next frontier directory.
* Falls through to the following phase when there is no syncContext to carry
* the frontier — see {@link pendingDirectories}.
* Falls through to the following phase when the frontier is empty, or when
* there is no syncContext to carry it — see {@link pendingDirectories}.
* Checking the frontier here matters on the skip paths (a timed-out or absent
* directory): claiming `hasMore` with nothing left to walk costs the sync
* engine a whole extra listing round-trip that can only return zero documents.
*/
const continueCode = (): { nextCursor?: string; hasMore: boolean } =>
syncContext ? { nextCursor: encodeCursor({ phase: 'code' }), hasMore: true } : advance('code')
const continueCode = (frontier: string[]): { nextCursor?: string; hasMore: boolean } =>
syncContext && frontier.length > 0
? { nextCursor: encodeCursor({ phase: 'code' }), hasMore: true }
: advance('code')
if (state.phase === 'code') {
const { commit, ref } = await resolveCommit(
@@ -841,7 +846,7 @@ export const bitbucketConnector: ConnectorConfig = {
dir,
})
if (syncContext) syncContext.listingCapped = true
const skipped = continueCode()
const skipped = continueCode(frontier)
return { documents: [], nextCursor: skipped.nextCursor, hasMore: skipped.hasMore }
}
if (response.status === 404) {
@@ -854,7 +859,7 @@ export const bitbucketConnector: ConnectorConfig = {
repository: repository.fullName,
dir,
})
const skipped = continueCode()
const skipped = continueCode(frontier)
return { documents: [], nextCursor: skipped.nextCursor, hasMore: skipped.hasMore }
}
if (response.status === 401 || response.status === 403) {
@@ -928,12 +933,8 @@ export const bitbucketConnector: ConnectorConfig = {
hasMore: true,
}
}
if (frontier.length > 0) {
const cont = continueCode()
return { documents: capped, nextCursor: cont.nextCursor, hasMore: cont.hasMore }
}
const adv = advance('code')
return { documents: capped, nextCursor: adv.nextCursor, hasMore: adv.hasMore }
const cont = continueCode(frontier)
return { documents: capped, nextCursor: cont.nextCursor, hasMore: cont.hasMore }
}
let url: string
+9 -5
View File
@@ -10,12 +10,16 @@ export const bitbucketConnectorMeta: ConnectorMeta = {
icon: BitbucketIcon,
/**
* Incremental sync applies to pull requests only, via a BBQL
* `q=updated_on > <timestamp>` filter derived from lastSyncAt. Repository files
* carry no per-file change timestamp in the source listing, so they are always
* re-listed in full and reconciled by content hash.
* Deliberately disabled. Incremental sync would only ever apply to pull requests,
* via a BBQL `q=updated_on > <timestamp>` filter derived from lastSyncAt; repository
* files carry no per-file change timestamp in the source listing and are always
* re-listed in full regardless. The sync engine disables deletion reconciliation for
* every incremental run, so declaring support would leave files deleted upstream
* indexed indefinitely on the default `code` configuration -- a complete listing whose
* deletions are never applied. Re-listing pull requests in full each run is the
* cheaper tradeoff.
*/
supportsIncrementalSync: true,
supportsIncrementalSync: false,
/**
* Bitbucket Cloud REST API 2.0 authenticates with an OAuth 2.0 bearer token, the
@@ -0,0 +1,650 @@
/**
* @vitest-environment node
*
* `GET /api/2.0/workspace/list` is unpaginated and returns one directory at a
* time, so the notebook side of this connector is a hand-rolled breadth-first
* walk whose pending-directory queue rides in the cursor. Everything that can
* silently shrink a listing lives in that walk — a container type the walk does
* not recognise, a directory the token cannot read, the document cap — and the
* sync engine hard-deletes whatever a listing leaves out unless
* `syncContext.listingCapped` says the listing was partial. Those paths are the
* bulk of what is asserted here, alongside the stub/`getDocument` hash equality
* that keeps every sync from re-indexing every notebook.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { databricksConnector } from '@/connectors/databricks/databricks'
import { databricksConnectorMeta } from '@/connectors/databricks/meta'
const TOKEN = 'dapi-personal-access-token'
const HOST = 'dbc-1234abcd-5678.cloud.databricks.com'
const ORIGIN = `https://${HOST}`
const NOTEBOOK_CONFIG = { workspaceHost: HOST, contentType: 'notebooks', rootPath: '/Shared' }
const QUERY_CONFIG = { workspaceHost: HOST, contentType: 'queries' }
const fetchMock = vi.fn()
function jsonResponse(body: unknown, status = 200): Response {
return new Response(JSON.stringify(body), {
status,
headers: { 'Content-Type': 'application/json' },
})
}
function dir(path: string) {
return { object_type: 'DIRECTORY', path, object_id: path.length }
}
function repo(path: string) {
return { object_type: 'REPO', path, object_id: path.length }
}
function notebook(path: string, modifiedAt?: number) {
return {
object_type: 'NOTEBOOK',
path,
object_id: path.length * 1000,
language: 'PYTHON',
...(modifiedAt === undefined ? {} : { modified_at: modifiedAt }),
}
}
function query(id: string, overrides: Record<string, unknown> = {}) {
return {
id,
display_name: `Query ${id}`,
query_text: `SELECT ${id}`,
catalog: 'main',
schema: 'default',
owner_user_name: 'ada@example.com',
lifecycle_state: 'ACTIVE',
update_time: '2026-01-01T00:00:00Z',
...overrides,
}
}
/** Reads the `path` query parameter off a recorded `workspace/list` call. */
function listedPaths(): string[] {
return fetchMock.mock.calls
.map(([url]) => new URL(url as string))
.filter((url) => url.pathname === '/api/2.0/workspace/list')
.map((url) => url.searchParams.get('path') ?? '')
}
function queryListingUrls(): URL[] {
return fetchMock.mock.calls
.map(([url]) => new URL(url as string))
.filter((url) => url.pathname === '/api/2.0/sql/queries')
}
/**
* Serves `workspace/list` from a path→objects map. An unmapped path is a 404,
* which is the Databricks answer for a path that no longer exists.
*/
function mockWorkspace(tree: Record<string, unknown[] | (() => Response)>) {
fetchMock.mockImplementation(async (url: string) => {
const parsed = new URL(url)
if (parsed.pathname === '/api/2.0/workspace/list') {
const path = parsed.searchParams.get('path') ?? ''
const entry = tree[path]
if (entry === undefined) {
return jsonResponse({ error_code: 'RESOURCE_DOES_NOT_EXIST', message: 'not found' }, 404)
}
/** A factory, not a value: a retried request must not re-read a used body. */
if (typeof entry === 'function') return entry()
return jsonResponse({ objects: entry })
}
throw new Error(`Unexpected request: ${url}`)
})
}
/** Wraps an error body so each (possibly retried) request gets a fresh Response. */
function errorResponse(body: Record<string, string>, status: number): () => Response {
return () => jsonResponse(body, status)
}
beforeEach(() => {
vi.clearAllMocks()
vi.stubGlobal('fetch', fetchMock)
})
afterEach(() => {
vi.unstubAllGlobals()
vi.useRealTimers()
})
describe('databricks notebook traversal', () => {
it('walks DIRECTORY and REPO containers and indexes only notebooks', async () => {
mockWorkspace({
'/Shared': [dir('/Shared/team'), repo('/Shared/git-folder')],
'/Shared/team': [
notebook('/Shared/team/etl', 1),
{ object_type: 'FILE', path: '/Shared/team/readme.md', object_id: 5 },
{ object_type: 'DASHBOARD', path: '/Shared/team/board.lvdash.json', object_id: 6 },
{ object_type: 'LIBRARY', path: '/Shared/team/lib.jar', object_id: 7 },
],
'/Shared/git-folder': [notebook('/Shared/git-folder/model', 2)],
})
const syncContext: Record<string, unknown> = {}
const result = await databricksConnector.listDocuments(
TOKEN,
NOTEBOOK_CONFIG,
undefined,
syncContext
)
expect(listedPaths().sort()).toEqual(['/Shared', '/Shared/git-folder', '/Shared/team'])
expect(result.documents.map((doc) => doc.externalId).sort()).toEqual([
'notebook:/Shared/git-folder/model',
'notebook:/Shared/team/etl',
])
expect(result.hasMore).toBe(false)
expect(syncContext.listingCapped).toBeUndefined()
})
/**
* `workspace/list` answers with the object itself when the path is not a
* directory. Re-queueing that self-reference would list the same path forever.
*/
it('does not re-queue a container that lists itself', async () => {
mockWorkspace({
'/Shared': [dir('/Shared'), notebook('/Shared/etl', 1)],
})
const result = await databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
expect(listedPaths()).toEqual(['/Shared'])
expect(result.documents).toHaveLength(1)
expect(result.hasMore).toBe(false)
})
it('carries the pending queue through the cursor without re-listing or duplicating', async () => {
/** 60 sibling directories: more than the 50 a single call will list. */
const children = Array.from({ length: 60 }, (_, i) => dir(`/Shared/d${i}`))
const tree: Record<string, unknown[]> = { '/Shared': children }
for (let i = 0; i < 60; i++) {
tree[`/Shared/d${i}`] = [notebook(`/Shared/d${i}/nb`, i + 1)]
}
mockWorkspace(tree)
const first = await databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
expect(first.hasMore).toBe(true)
expect(first.nextCursor).toBeDefined()
expect(first.documents).toHaveLength(49)
const firstPaths = listedPaths()
fetchMock.mockClear()
const second = await databricksConnector.listDocuments(
TOKEN,
NOTEBOOK_CONFIG,
first.nextCursor,
{}
)
expect(second.hasMore).toBe(false)
expect(second.documents).toHaveLength(11)
const secondPaths = listedPaths()
expect(secondPaths.some((path) => firstPaths.includes(path))).toBe(false)
const ids = [...first.documents, ...second.documents].map((doc) => doc.externalId)
expect(new Set(ids).size).toBe(ids.length)
expect(ids).toHaveLength(60)
})
})
describe('databricks notebook listingCapped', () => {
it('flags listingCapped when a subdirectory is unreadable', async () => {
mockWorkspace({
'/Shared': [dir('/Shared/open'), dir('/Shared/locked')],
'/Shared/open': [notebook('/Shared/open/nb', 1)],
'/Shared/locked': errorResponse({ error_code: 'PERMISSION_DENIED' }, 403),
})
const syncContext: Record<string, unknown> = {}
const result = await databricksConnector.listDocuments(
TOKEN,
NOTEBOOK_CONFIG,
undefined,
syncContext
)
expect(result.documents).toHaveLength(1)
expect(syncContext.listingCapped).toBe(true)
})
it('flags listingCapped when a subdirectory has been deleted mid-walk', async () => {
mockWorkspace({
'/Shared': [dir('/Shared/open'), dir('/Shared/gone')],
'/Shared/open': [notebook('/Shared/open/nb', 1)],
})
const syncContext: Record<string, unknown> = {}
await databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, syncContext)
expect(syncContext.listingCapped).toBe(true)
})
/**
* A denied root yields an empty listing, not a partial one. Reporting that as
* a successful sync of zero documents is worse than failing.
*/
it('throws rather than reporting an empty sync when the root itself is denied', async () => {
mockWorkspace({ '/Shared': errorResponse({ error_code: 'PERMISSION_DENIED' }, 403) })
await expect(
databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
).rejects.toThrow(/denied access to \/Shared/)
})
it('flags listingCapped when maxDocuments hides notebooks from the same page', async () => {
mockWorkspace({
'/Shared': [notebook('/Shared/a', 1), notebook('/Shared/b', 2), notebook('/Shared/c', 3)],
})
const syncContext: Record<string, unknown> = {}
const result = await databricksConnector.listDocuments(
TOKEN,
{ ...NOTEBOOK_CONFIG, maxDocuments: '2' },
undefined,
syncContext
)
expect(result.documents).toHaveLength(2)
expect(result.hasMore).toBe(false)
expect(syncContext.listingCapped).toBe(true)
})
it('leaves listingCapped unset when maxDocuments lands exactly on an exhausted walk', async () => {
mockWorkspace({
'/Shared': [notebook('/Shared/a', 1), notebook('/Shared/b', 2)],
})
const syncContext: Record<string, unknown> = {}
const result = await databricksConnector.listDocuments(
TOKEN,
{ ...NOTEBOOK_CONFIG, maxDocuments: '2' },
undefined,
syncContext
)
expect(result.documents).toHaveLength(2)
expect(result.hasMore).toBe(false)
expect(syncContext.listingCapped).toBeUndefined()
})
it('flags listingCapped when maxDocuments stops the walk with directories pending', async () => {
mockWorkspace({
'/Shared': [dir('/Shared/a'), dir('/Shared/b')],
'/Shared/a': [notebook('/Shared/a/nb', 1)],
'/Shared/b': [notebook('/Shared/b/nb', 2)],
})
const syncContext: Record<string, unknown> = { totalDocsFetched: 0 }
const result = await databricksConnector.listDocuments(
TOKEN,
{ ...NOTEBOOK_CONFIG, maxDocuments: '1' },
undefined,
syncContext
)
expect(result.documents).toHaveLength(1)
expect(syncContext.listingCapped).toBe(true)
})
})
describe('databricks notebook error semantics under concurrency', () => {
/**
* The walk lists siblings in parallel. A permission failure on one sibling has
* to stay a skip while a server failure on another still aborts the whole
* listing — a `Promise.all` that swallowed either would be invisible in the
* happy path.
*/
it('propagates a 500 on one sibling even when another sibling is a skippable 403', async () => {
mockWorkspace({
'/Shared': [dir('/Shared/locked'), dir('/Shared/broken')],
'/Shared/locked': errorResponse({ error_code: 'PERMISSION_DENIED' }, 403),
'/Shared/broken': errorResponse({ error_code: 'INTERNAL_ERROR', message: 'boom' }, 500),
})
await expect(
databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
).rejects.toThrow(/Failed to list Databricks workspace path \/Shared\/broken: 500/)
})
it('propagates a 401 rather than treating a bad token as an empty directory', async () => {
mockWorkspace({
'/Shared': [dir('/Shared/a')],
'/Shared/a': errorResponse({ error_code: 'PERMISSION_DENIED' }, 401),
})
await expect(
databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
).rejects.toThrow(/401/)
})
it('propagates a 429 after the shared retry budget is spent', async () => {
vi.useFakeTimers()
mockWorkspace({
'/Shared': [dir('/Shared/a')],
'/Shared/a': errorResponse({ error_code: 'REQUEST_LIMIT_EXCEEDED' }, 429),
})
const pending = databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
const assertion = expect(pending).rejects.toThrow(/429/)
await vi.runAllTimersAsync()
await assertion
})
})
describe('databricks saved SQL queries', () => {
it('drops trashed queries without suppressing deletion reconciliation', async () => {
fetchMock.mockResolvedValue(
jsonResponse({
results: [query('a'), query('b', { lifecycle_state: 'TRASHED' })],
})
)
const syncContext: Record<string, unknown> = {}
const result = await databricksConnector.listDocuments(
TOKEN,
QUERY_CONFIG,
undefined,
syncContext
)
expect(result.documents.map((doc) => doc.externalId)).toEqual(['query:a'])
expect(result.hasMore).toBe(false)
expect(syncContext.listingCapped).toBeUndefined()
})
it('flags listingCapped when the cap stops paging while a page token remains', async () => {
fetchMock.mockResolvedValue(
jsonResponse({ results: [query('a'), query('b')], next_page_token: 'tok' })
)
const syncContext: Record<string, unknown> = {}
const result = await databricksConnector.listDocuments(
TOKEN,
{ ...QUERY_CONFIG, maxDocuments: '2' },
undefined,
syncContext
)
expect(result.documents).toHaveLength(2)
expect(result.hasMore).toBe(false)
expect(syncContext.listingCapped).toBe(true)
})
it('leaves listingCapped unset when the cap lands on the last page', async () => {
fetchMock.mockResolvedValue(jsonResponse({ results: [query('a'), query('b')] }))
const syncContext: Record<string, unknown> = {}
await databricksConnector.listDocuments(
TOKEN,
{ ...QUERY_CONFIG, maxDocuments: '2' },
undefined,
syncContext
)
expect(syncContext.listingCapped).toBeUndefined()
})
/**
* An echoed token on an empty tail would re-request the same page until the
* engine truncates pagination, which permanently disables deletion
* reconciliation.
*/
it('stops paging on an empty page even when a token is echoed back', async () => {
fetchMock.mockResolvedValue(jsonResponse({ results: [], next_page_token: 'tok' }))
const result = await databricksConnector.listDocuments(TOKEN, QUERY_CONFIG, undefined, {})
expect(result.hasMore).toBe(false)
expect(result.nextCursor).toBeUndefined()
})
it('asks for the full page size when no cap is configured', async () => {
fetchMock.mockResolvedValue(jsonResponse({ results: [query('a')] }))
await databricksConnector.listDocuments(TOKEN, QUERY_CONFIG, undefined, {})
expect(queryListingUrls()[0].searchParams.get('page_size')).toBe('100')
})
it('shrinks the last page to what the cap still allows', async () => {
fetchMock.mockResolvedValue(jsonResponse({ results: [query('a')] }))
await databricksConnector.listDocuments(
TOKEN,
{ ...QUERY_CONFIG, maxDocuments: '10' },
undefined,
{ totalDocsFetched: 7 }
)
expect(queryListingUrls()[0].searchParams.get('page_size')).toBe('3')
})
/**
* `page_size=0` is not a valid request. The cap arithmetic must floor at one
* even when the running total has already reached (or overshot) the cap.
*/
it.each([
['exactly at the cap', 10],
['past the cap', 12],
])('never requests a non-positive page size when already %s', async (_label, fetched) => {
fetchMock.mockResolvedValue(jsonResponse({ results: [] }))
await databricksConnector.listDocuments(
TOKEN,
{ ...QUERY_CONFIG, maxDocuments: '10' },
undefined,
{ totalDocsFetched: fetched }
)
const pageSize = Number(queryListingUrls()[0].searchParams.get('page_size'))
expect(pageSize).toBeGreaterThan(0)
})
})
describe('databricks contentHash consistency', () => {
it('produces the same notebook hash from the listing stub and from getDocument', async () => {
mockWorkspace({ '/Shared': [notebook('/Shared/etl', 1700000000000)] })
const listed = await databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
const stub = listed.documents[0]
expect(stub.contentDeferred).toBe(true)
expect(stub.content).toBe('')
fetchMock.mockImplementation(async (url: string) => {
const parsed = new URL(url)
if (parsed.pathname === '/api/2.0/workspace/get-status') {
return jsonResponse(notebook('/Shared/etl', 1700000000000))
}
if (parsed.pathname === '/api/2.0/workspace/export') {
expect(parsed.searchParams.get('format')).toBe('SOURCE')
return jsonResponse({
content: Buffer.from('print("hi")', 'utf8').toString('base64'),
file_type: 'py',
})
}
throw new Error(`Unexpected request: ${url}`)
})
const hydrated = await databricksConnector.getDocument(TOKEN, NOTEBOOK_CONFIG, stub.externalId)
expect(hydrated?.contentHash).toBe(stub.contentHash)
expect(hydrated?.externalId).toBe(stub.externalId)
expect(hydrated?.contentDeferred).toBe(false)
expect(hydrated?.content).toBe('print("hi")')
expect(hydrated?.sourceUrl).toBe(`${ORIGIN}/#notebook/${'/Shared/etl'.length * 1000}`)
})
/**
* `modified_at` is documented as file-only, so a notebook routinely carries no
* timestamp at all. Its hash must still be stable across syncs — a clock- or
* fetch-derived component would make every sync re-index every notebook. The
* cost of that stability is that edits go unseen until a full resync, which is
* exactly what `rehydrateOnFullSync` covers.
*/
it('keeps an undated notebook hash stable across repeated listings', async () => {
mockWorkspace({ '/Shared': [notebook('/Shared/etl')] })
const first = await databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
const second = await databricksConnector.listDocuments(TOKEN, NOTEBOOK_CONFIG, undefined, {})
expect(first.documents[0].contentHash).toBe('databricks:notebook:/Shared/etl:')
expect(second.documents[0].contentHash).toBe(first.documents[0].contentHash)
})
it('produces the same query hash from the listing and from getDocument', async () => {
fetchMock.mockResolvedValue(jsonResponse({ results: [query('q1')] }))
const listed = await databricksConnector.listDocuments(TOKEN, QUERY_CONFIG, undefined, {})
const stub = listed.documents[0]
fetchMock.mockResolvedValue(jsonResponse(query('q1')))
const refetched = await databricksConnector.getDocument(TOKEN, QUERY_CONFIG, stub.externalId)
expect(refetched?.contentHash).toBe(stub.contentHash)
expect(refetched?.content).toBe(stub.content)
})
it('returns null for a notebook that has been deleted since the listing', async () => {
fetchMock.mockResolvedValue(
jsonResponse({ error_code: 'RESOURCE_DOES_NOT_EXIST', message: 'gone' }, 404)
)
await expect(
databricksConnector.getDocument(TOKEN, NOTEBOOK_CONFIG, 'notebook:/Shared/gone')
).resolves.toBeNull()
})
it('reports an oversize export as a skip rather than losing the notebook', async () => {
fetchMock.mockImplementation(async (url: string) => {
const parsed = new URL(url)
if (parsed.pathname === '/api/2.0/workspace/get-status') {
return jsonResponse(notebook('/Shared/big', 5))
}
return jsonResponse({ error_code: 'MAX_NOTEBOOK_SIZE_EXCEEDED', message: 'too big' }, 400)
})
const doc = await databricksConnector.getDocument(
TOKEN,
NOTEBOOK_CONFIG,
'notebook:/Shared/big'
)
expect(doc?.skippedReason).toMatch(/10MB size limit/)
expect(doc?.contentHash).toBe(`databricks:notebook:/Shared/big:5`)
})
})
describe('databricks configuration', () => {
it('declares rehydrateOnFullSync so undated notebook edits stay recoverable', () => {
expect(databricksConnectorMeta.rehydrateOnFullSync).toBe(true)
})
it('refuses a workspace host outside the Databricks-owned domains', async () => {
const result = await databricksConnector.validateConfig(TOKEN, {
workspaceHost: 'evil.example.com',
contentType: 'notebooks',
})
expect(result.valid).toBe(false)
expect(result.error).toMatch(/Databricks-hosted domain/)
expect(fetchMock).not.toHaveBeenCalled()
})
it('refuses to list against a host outside the allowlist', async () => {
await expect(
databricksConnector.listDocuments(
TOKEN,
{ ...NOTEBOOK_CONFIG, workspaceHost: 'evil.example.com' },
undefined,
{}
)
).rejects.toThrow(/Databricks-hosted domain/)
})
it('rejects a root path that names a notebook instead of a folder', async () => {
fetchMock.mockResolvedValue(jsonResponse(notebook('/Shared/etl', 1)))
const result = await databricksConnector.validateConfig(TOKEN, {
...NOTEBOOK_CONFIG,
rootPath: '/Shared/etl',
})
expect(result.valid).toBe(false)
expect(result.error).toMatch(/is a notebook, not a folder/)
})
it('accepts a Git folder as the notebook root', async () => {
fetchMock.mockResolvedValue(jsonResponse(repo('/Repos/ada/project')))
await expect(
databricksConnector.validateConfig(TOKEN, {
...NOTEBOOK_CONFIG,
rootPath: '/Repos/ada/project',
})
).resolves.toEqual({ valid: true })
})
it('normalizes a relative, trailing-slashed root path to an absolute one', async () => {
fetchMock.mockResolvedValue(jsonResponse(dir('/Shared/team')))
await databricksConnector.validateConfig(TOKEN, {
...NOTEBOOK_CONFIG,
rootPath: ' Shared/team/ ',
})
const url = new URL(fetchMock.mock.calls[0][0] as string)
expect(url.searchParams.get('path')).toBe('/Shared/team')
})
})
describe('databricks mapTags', () => {
it('maps notebook metadata to the declared tag slots', () => {
const tags = databricksConnector.mapTags?.({
language: 'PYTHON',
lastModified: '2026-01-01T00:00:00.000Z',
path: '/Shared/etl',
})
expect(tags?.language).toBe('PYTHON')
expect(tags?.lastModified).toBeInstanceOf(Date)
})
it('maps query metadata, including the joined tag array', () => {
const tags = databricksConnector.mapTags?.({
owner: 'ada@example.com',
catalog: 'main',
schema: 'default',
labels: ['finance', 'daily'],
lastModified: '2026-01-01T00:00:00Z',
})
expect(tags).toMatchObject({
owner: 'ada@example.com',
catalog: 'main',
schema: 'default',
labels: 'finance, daily',
})
})
it('emits only tag ids the connector declares', () => {
const declared = new Set(databricksConnectorMeta.tagDefinitions?.map((tag) => tag.id))
const produced = databricksConnector.mapTags?.({
language: 'SQL',
owner: 'ada@example.com',
catalog: 'main',
schema: 'default',
labels: ['a'],
lastModified: '2026-01-01T00:00:00Z',
})
for (const key of Object.keys(produced ?? {})) {
expect(declared.has(key)).toBe(true)
}
})
})
+2 -2
View File
@@ -717,8 +717,8 @@ export const databricksConnector: ConnectorConfig = {
const maxDocuments = sourceConfig.maxDocuments
if (maxDocuments !== undefined && maxDocuments !== null && String(maxDocuments).trim()) {
const parsed = Number(maxDocuments)
if (!Number.isFinite(parsed) || parsed < 0) {
return { valid: false, error: 'Max documents must be a non-negative number' }
if (!Number.isFinite(parsed) || parsed <= 0) {
return { valid: false, error: 'Max documents must be a positive number' }
}
}
@@ -73,6 +73,13 @@ function listFilter(): string | null {
return listUrl ? new URL(listUrl).searchParams.get('filter') : null
}
/** Search params of the first `spaces.messages.list` request. */
function messagesParams(): URLSearchParams {
const url = requestedUrls.find((requested) => requested.includes('/messages?'))
if (!url) throw new Error('no messages request was made')
return new URL(url).searchParams
}
describe('google-chat space scope', () => {
it('filters to named spaces only when spaceTypes is unset', async () => {
await googleChatConnector.listDocuments('token', {})
@@ -168,6 +175,21 @@ describe('google-chat listing caps', () => {
expect(syncContext.listingCapped).toBe(true)
})
it('stops paginating at the cap even while the source offers another page', async () => {
listNextPageToken = 'page-2'
const syncContext: Record<string, unknown> = {}
const result = await googleChatConnector.listDocuments(
'token',
{ maxSpaces: '1' },
undefined,
syncContext
)
expect(result.hasMore).toBe(false)
expect(result.nextCursor).toBeUndefined()
expect(syncContext.listingCapped).toBe(true)
})
it('leaves deletion reconciliation enabled when the source is exhausted at the cap', async () => {
const syncContext: Record<string, unknown> = {}
await googleChatConnector.listDocuments('token', { maxSpaces: '1' }, undefined, syncContext)
@@ -182,6 +204,103 @@ describe('google-chat listing caps', () => {
})
})
describe('google-chat message window', () => {
it('requests messages newest-first so the cap keeps the most recent conversation', async () => {
await googleChatConnector.getDocument('token', {}, SPACE_NAME)
expect(messagesParams().get('orderBy')).toBe('DESC')
})
it('renders the newest-first page back into chronological order', async () => {
const doc = await googleChatConnector.getDocument('token', {}, SPACE_NAME)
const content = doc?.content ?? ''
expect(content.indexOf('Ada Lovelace')).toBeLessThan(content.indexOf('Grace Hopper'))
})
it('bounds the message page to the remaining window rather than the API maximum', async () => {
await googleChatConnector.getDocument('token', { maxMessages: '5' }, SPACE_NAME)
expect(messagesParams().get('pageSize')).toBe('5')
})
it('sends the lookback cutoff as an RFC-3339 timestamp without fractional seconds', async () => {
await googleChatConnector.getDocument('token', { lookbackDays: '30' }, SPACE_NAME)
const filter = messagesParams().get('filter')
expect(filter).toMatch(/^createTime > "\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z"$/)
})
it('omits the filter entirely when no lookback window is configured', async () => {
await googleChatConnector.getDocument('token', {}, SPACE_NAME)
expect(messagesParams().get('filter')).toBeNull()
})
it('indexes a card-only message through its fallbackText', async () => {
fetchMock.mockImplementation(async (input) => {
const url = String(input)
requestedUrls.push(url)
if (url.includes('/messages?')) {
return jsonResponse({
messages: [
{
name: `${SPACE_NAME}/messages/card`,
sender: { name: 'users/9', type: 'BOT' },
createTime: '2026-02-01T09:00:00Z',
fallbackText: 'Deploy finished',
},
],
})
}
if (url.endsWith(`/${SPACE_NAME}`)) return jsonResponse(SPACE)
return jsonResponse({}, 404)
})
const doc = await googleChatConnector.getDocument('token', {}, SPACE_NAME)
expect(doc?.content).toContain('Deploy finished')
expect(doc?.metadata?.messageCount).toBe(1)
})
it('labels a sender by resource name when user auth omits displayName', async () => {
fetchMock.mockImplementation(async (input) => {
const url = String(input)
requestedUrls.push(url)
if (url.includes('/messages?')) {
return jsonResponse({
messages: [
{
name: `${SPACE_NAME}/messages/m1`,
sender: { name: 'users/104512345678', type: 'HUMAN' },
createTime: '2026-02-01T10:00:00Z',
text: 'Morning',
},
],
})
}
if (url.endsWith(`/${SPACE_NAME}`)) return jsonResponse(SPACE)
return jsonResponse({}, 404)
})
const doc = await googleChatConnector.getDocument('token', {}, SPACE_NAME)
expect(doc?.content).toContain('users/104512345678: Morning')
})
it('rehashes when the configured window changes so the stored transcript is refetched', async () => {
const base = await googleChatConnector.listDocuments('token', {}, undefined, {})
const narrower = await googleChatConnector.listDocuments(
'token',
{ maxMessages: '50' },
undefined,
{}
)
const windowed = await googleChatConnector.listDocuments(
'token',
{ lookbackDays: '30' },
undefined,
{}
)
expect(narrower.documents[0].contentHash).not.toBe(base.documents[0].contentHash)
expect(windowed.documents[0].contentHash).not.toBe(base.documents[0].contentHash)
})
})
describe('google-chat getDocument', () => {
it('returns null for a space that no longer exists', async () => {
await expect(googleChatConnector.getDocument('token', {}, 'spaces/MISSING')).resolves.toBeNull()
+23 -4
View File
@@ -45,9 +45,16 @@ interface SpacesListResponse {
}
/**
* The author of a message. With user authentication the API populates
* `displayName`; a Chat app authenticating as a user only receives `name` and
* `type`, so `name` is the fallback.
* The author of a message.
*
* This connector authenticates as a user, and the User reference states that
* "if your Chat app authenticates as a user, the output for a User resource
* only populates the user's `name` and `type`" — so `displayName` is expected
* to be absent here and `name` (a `users/{id}` resource name) is the label that
* actually reaches the transcript. Resolving human names would need
* `spaces.members.list` and the `chat.memberships.readonly` scope, which this
* connector does not request. `displayName` is still read first so the
* transcript improves for free if a call ever does carry it.
*
* https://developers.google.com/workspace/chat/api/reference/rest/v1/User
*/
@@ -235,6 +242,18 @@ async function fetchSpace(accessToken: string, name: string): Promise<Space | nu
return (await response.json()) as Space
}
/**
* An RFC-3339 timestamp without fractional seconds, matching the form every
* example in the `spaces.messages.list` filter reference uses. `toISOString()`
* alone emits milliseconds, and the API rejects a filter it cannot parse with
* `INVALID_ARGUMENT` rather than ignoring it.
*
* https://developers.google.com/workspace/chat/api/reference/rest/v1/spaces.messages/list
*/
function rfc3339(date: Date): string {
return `${date.toISOString().slice(0, 19)}Z`
}
/**
* Fetches the newest `maxMessages` messages of a space, optionally bounded by a
* lookback window. Messages are requested newest-first (`orderBy=DESC`) so the
@@ -248,7 +267,7 @@ async function fetchSpaceMessages(
): Promise<ChatMessage[]> {
const filter =
lookbackDays > 0
? `createTime > "${new Date(Date.now() - lookbackDays * MS_PER_DAY).toISOString()}"`
? `createTime > "${rfc3339(new Date(Date.now() - lookbackDays * MS_PER_DAY))}"`
: undefined
const collected: ChatMessage[] = []
+16 -5
View File
@@ -39,11 +39,22 @@ import type { ConnectorMeta } from '@/connectors/types'
*
* `/articleVersions` lists article *versions*, not articles: every row carries a
* `version` number and a `parentArticle` reference to the article it belongs to,
* and the service publishes no "latest version only" filter or sort order. An
* unfiltered sync can therefore index each historical revision as its own
* document. Status is the only server-side narrowing the API offers, so it is
* required and has no default: the operator picks the scope explicitly, and the
* option that indexes every revision says so on its label.
* and the service publishes no "latest version only" filter or sort order — the
* whole document has eleven paths and none of them addresses an article. An
* unfiltered sync therefore indexes each historical revision as its own
* document, so the status filter is required and has no default: the operator
* picks the scope explicitly, and the option that indexes every revision says so
* on its label.
*
* The three options are the ones the service's own prose enumerates — "an
* article version can have a status of Published, Draft, or Archived". What the
* `status` *query parameter* accepts for them is the one thing the published
* spec does not settle: it is an untyped `array` of `string` with no enum and no
* `x-workday-populated-by`, unlike the sibling `audience` parameter, whose model
* names `/values/common/audiences` as its value source. The connector resolves
* the chosen name against `/articleStatuses` and sends the Workday ID;
* `validateConfig` reads the filtered response back and refuses the
* configuration if the tenant answered with a version in another status.
*/
export const workdayConnectorMeta: ConnectorMeta = {
id: 'workday',
+133
View File
@@ -73,6 +73,12 @@ function mockApi(listing: unknown, options: { audiences?: unknown } = {}) {
})
}
function tokenExchanges(): string[] {
return mockFetch.mock.calls
.map(([url]) => url as string)
.filter((url) => url.includes('/ccx/oauth2/'))
}
function listingUrls(): string[] {
return mockFetch.mock.calls
.map(([url]) => url as string)
@@ -211,6 +217,60 @@ describe('workday listDocuments', () => {
)
})
it('accepts a tenant host written without a scheme', async () => {
mockApi({ total: 1, data: [versionFixture('a')] })
await workdayConnector.listDocuments(
ACCESS_TOKEN,
{ ...CONFIG, tenantUrl: 'wd5-impl-services1.workday.com' },
undefined,
{}
)
expect(listingUrls()[0]).toContain(
'https://wd5-impl-services1.workday.com/ccx/api/helpArticle/v1/acme_pt1/articleVersions?'
)
})
it('passes a Workday ID through instead of looking it up as a display name', async () => {
mockApi({ total: 1, data: [versionFixture('a')] })
await workdayConnector.listDocuments(
ACCESS_TOKEN,
{ ...CONFIG, audience: AUDIENCE_ID },
undefined,
{}
)
expect(listingUrls()[0]).toContain(`audience=${AUDIENCE_ID}`)
expect(
mockFetch.mock.calls.some(([url]) => (url as string).includes('/values/common/audiences'))
).toBe(false)
})
it('caps how many available values an unresolved-name error names', async () => {
const many = Array.from({ length: 30 }, (_, index) => ({
id: `${index}`.padStart(32, '0'),
descriptor: `Audience ${index}`,
}))
mockApi({ total: 1, data: [versionFixture('a')] }, { audiences: { total: 30, data: many } })
const error = await workdayConnector
.listDocuments(ACCESS_TOKEN, { ...CONFIG, audience: 'Nope' }, undefined, {})
.catch((thrown: Error) => thrown)
expect((error as Error).message).toContain('and 10 more')
expect((error as Error).message).not.toContain('Audience 25')
})
it('reads a version cap persisted as a number rather than a string', async () => {
mockApi({ total: 500, data: [versionFixture('a')] })
await workdayConnector.listDocuments(ACCESS_TOKEN, { ...CONFIG, maxVersions: 1 }, undefined, {})
expect(listingUrls()[0]).toContain('limit=1')
})
it('rejects a tenant host that is not a Workday domain', async () => {
mockApi({ total: 0, data: [] })
@@ -244,6 +304,79 @@ describe('workday credential failures', () => {
expect(result.error).toMatch(/Manage Refresh Tokens for Integrations/)
})
it('buys one bearer token for a whole sync run and reuses it across pages', async () => {
mockApi({ total: 500, data: [versionFixture('a')] })
const syncContext: Record<string, unknown> = {}
await workdayConnector.listDocuments(ACCESS_TOKEN, CONFIG, undefined, syncContext)
await workdayConnector.listDocuments(ACCESS_TOKEN, CONFIG, '100', syncContext)
expect(tokenExchanges()).toHaveLength(1)
})
it('re-authenticates once and replays the request when a cached token has expired', async () => {
let listingCalls = 0
mockFetch.mockImplementation(async (url: string) => {
if (url.includes('/ccx/oauth2/')) {
return jsonResponse({ access_token: 'bearer', expires_in: 3600 })
}
if (url.includes('/articleStatuses')) {
return jsonResponse({ total: 1, data: [{ id: PUBLISHED_ID, descriptor: 'Published' }] })
}
listingCalls += 1
return listingCalls === 1
? jsonResponse({ error: 'expired token' }, 401)
: jsonResponse({ total: 1, data: [versionFixture('a')] })
})
const result = await workdayConnector.listDocuments(ACCESS_TOKEN, CONFIG, undefined, {})
expect(result.documents).toHaveLength(1)
expect(listingCalls).toBe(2)
expect(tokenExchanges()).toHaveLength(2)
})
it('surfaces a persistent 401 instead of retrying the exchange forever', async () => {
mockFetch.mockImplementation(async (url: string) => {
if (url.includes('/ccx/oauth2/')) {
return jsonResponse({ access_token: 'bearer', expires_in: 3600 })
}
if (url.includes('/articleStatuses')) {
return jsonResponse({ total: 1, data: [{ id: PUBLISHED_ID, descriptor: 'Published' }] })
}
return jsonResponse({ error: 'still unauthorized' }, 401)
})
await expect(
workdayConnector.listDocuments(ACCESS_TOKEN, CONFIG, undefined, {})
).rejects.toThrow(/still unauthorized/)
expect(tokenExchanges()).toHaveLength(2)
})
it('rejects a version cap that is not a positive number', async () => {
mockApi({ total: 0, data: [] })
const result = await workdayConnector.validateConfig(ACCESS_TOKEN, {
...CONFIG,
maxVersions: '0',
})
expect(result.valid).toBe(false)
expect(result.error).toMatch(/positive number/)
})
it('refuses a tenant that answered the status filter with another status', async () => {
mockApi({
total: 1,
data: [{ ...versionFixture('a'), status: { id: 'other', descriptor: 'Draft' } }],
})
const result = await workdayConnector.validateConfig(ACCESS_TOKEN, CONFIG)
expect(result.valid).toBe(false)
expect(result.error).toMatch(/ignored the article status filter/)
})
it('rejects a credential that is not a clientSecret:refreshToken pair', async () => {
mockApi({ total: 0, data: [] })
+95 -34
View File
@@ -31,6 +31,9 @@ const VALUES_PAGE_SIZE = 1000
*/
const MAX_VALUE_LOOKUP_ROWS = 5000
/** How many available values an unresolved-name error is allowed to name. */
const MAX_ERROR_DESCRIPTORS = 20
/**
* Workday resolves REST services as `/ccx/api/{service}/{version}/{tenant}/...`,
* where `{service}/{version}` is the `basePath` the service's OpenAPI document
@@ -353,20 +356,47 @@ async function fetchInstances(
if (typeof body.total === 'number' && instances.length >= body.total) break
}
if (instances.length >= MAX_VALUE_LOOKUP_ROWS) {
logger.warn('Workday value lookup hit its row cap; names beyond it cannot be resolved', {
resource,
rows: instances.length,
})
}
return instances
}
/**
* Names the values the lookup did return, capped. An audience list runs to as
* many rows as {@link MAX_VALUE_LOOKUP_ROWS} allows, and the whole list would
* otherwise be interpolated into an error string the knowledge base UI renders.
*/
function availableSuffix(descriptors: string[]): string {
if (descriptors.length === 0) return ''
const shown = descriptors.slice(0, MAX_ERROR_DESCRIPTORS)
const omitted = descriptors.length - shown.length
return `. Available: ${shown.join(', ')}${omitted > 0 ? `, and ${omitted} more` : ''}`
}
/**
* Turns configured display names into the Workday IDs the `status` and
* `audience` query parameters take. A value already shaped like a Workday ID or
* a `Type=Value` reference ID is passed through, so an operator who has the ID
* to hand never has to match a descriptor exactly.
*/
function resolveInstanceIds(
async function resolveInstanceIds(
values: string[],
instances: WorkdayInstance[],
loadInstances: () => Promise<WorkdayInstance[]>,
label: string
): string[] {
): Promise<string[]> {
/**
* Every value already being an ID is the common case for an operator who
* pasted Workday IDs, and the lookup it would otherwise run reads the tenant's
* whole prompt-value collection.
*/
if (values.every((value) => WORKDAY_INSTANCE_ID.test(value))) return values
const instances = await loadInstances()
const idByDescriptor = new Map<string, string>()
const descriptors: string[] = []
for (const instance of instances) {
@@ -383,9 +413,7 @@ function resolveInstanceIds(
const resolved = idByDescriptor.get(value.toLowerCase())
if (resolved) return resolved
throw new Error(
`Your Workday tenant has no ${label} named "${value}"${
descriptors.length > 0 ? `. Available: ${descriptors.join(', ')}` : ''
}`
`Your Workday tenant has no ${label} named "${value}"${availableSuffix(descriptors)}`
)
})
}
@@ -419,16 +447,17 @@ async function resolveFilters(
const status =
statusChoice === ALL_STATUSES
? []
: resolveInstanceIds(
: await resolveInstanceIds(
[statusChoice],
await fetchInstances(
'/articleStatuses',
PAGE_SIZE,
accessToken,
wd,
syncContext,
retryOptions
),
() =>
fetchInstances(
'/articleStatuses',
PAGE_SIZE,
accessToken,
wd,
syncContext,
retryOptions
),
'article status'
)
@@ -436,16 +465,17 @@ async function resolveFilters(
const audience =
audienceNames.length === 0
? []
: resolveInstanceIds(
: await resolveInstanceIds(
audienceNames,
await fetchInstances(
'/values/common/audiences/',
VALUES_PAGE_SIZE,
accessToken,
wd,
syncContext,
retryOptions
),
() =>
fetchInstances(
'/values/common/audiences/',
VALUES_PAGE_SIZE,
accessToken,
wd,
syncContext,
retryOptions
),
'audience'
)
@@ -522,11 +552,20 @@ function buildListUrl(
return tenantResourceUrl(wd, '/articleVersions', query)
}
/**
* Reads the optional version cap: `0` when none was configured, the floored
* count when one was, and `-1` when a value was supplied that is not a positive
* number, which {@link workdayConnector.validateConfig} rejects.
*
* The field is a short-input so the value normally arrives as a string, but a
* `sourceConfig` persisted with a JSON number must not crash the sync — reading
* it as a string and calling `.trim()` on it would.
*/
function parseMaxVersions(value: unknown): number {
const raw = (value as string | undefined)?.trim()
if (!raw) return 0
const raw = typeof value === 'string' ? value.trim() : value
if (raw === '' || raw === null || raw === undefined) return 0
const parsed = Number(raw)
return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : 0
return Number.isFinite(parsed) && parsed > 0 ? Math.floor(parsed) : -1
}
export const workdayConnector: ConnectorConfig = {
@@ -546,7 +585,7 @@ export const workdayConnector: ConnectorConfig = {
throw new Error(`Invalid pagination cursor: ${cursor}`)
}
const maxVersions = parseMaxVersions(sourceConfig.maxVersions)
const maxVersions = Math.max(parseMaxVersions(sourceConfig.maxVersions), 0)
const alreadyFetched = offset
const remaining = maxVersions > 0 ? maxVersions - alreadyFetched : PAGE_SIZE
const limit = Math.min(PAGE_SIZE, Math.max(remaining, 0))
@@ -638,14 +677,12 @@ export const workdayConnector: ConnectorConfig = {
accessToken: string,
sourceConfig: Record<string, unknown>
): Promise<{ valid: boolean; error?: string }> => {
const maxVersions = (sourceConfig.maxVersions as string | undefined)?.trim()
if (maxVersions) {
const parsed = Number(maxVersions)
if (!Number.isFinite(parsed) || parsed <= 0) {
return { valid: false, error: 'Max article versions must be a positive number' }
}
if (parseMaxVersions(sourceConfig.maxVersions) < 0) {
return { valid: false, error: 'Max article versions must be a positive number' }
}
const statusChoice = typeof sourceConfig.status === 'string' ? sourceConfig.status.trim() : ''
try {
const wd = resolveTenant(sourceConfig)
@@ -680,6 +717,30 @@ export const workdayConnector: ConnectorConfig = {
}
}
/**
* The `helpArticle` OpenAPI document declares `status` as an untyped
* `array` of `string` with no enum, no `$ref` and — unlike the sibling
* `audience` parameter, whose model carries
* `x-workday-populated-by: /values/common/audiences` — no declared value
* source. Whether it binds to an `/articleStatuses` Workday ID cannot be
* settled from the published spec, so the one request validation already
* makes is read back: a tenant that ignored the filter answers with a
* version in some other status, and the connector says so at configuration
* time instead of silently indexing the wrong scope on every sync.
*/
const sample = ((await response.json()) as WorkdayCollection<WorkdayArticleVersion>).data?.[0]
const returnedStatus = sample?.status?.descriptor
if (
statusChoice !== ALL_STATUSES &&
returnedStatus &&
returnedStatus.toLowerCase() !== statusChoice.toLowerCase()
) {
return {
valid: false,
error: `Workday ignored the article status filter: the first article version it returned is "${returnedStatus}", not "${statusChoice}". Choose "Every status" to sync every revision, or ask your Workday administrator whether the Help Article REST API accepts a status filter in this tenant.`,
}
}
return { valid: true }
} catch (error) {
logger.warn('Workday connector validation failed', { error: getErrorMessage(error) })
+8
View File
@@ -287,6 +287,14 @@ export const OAUTH_PROVIDERS: Record<string, OAuthProviderConfig> = {
],
serviceAccountProviderId: 'google-service-account',
},
/**
* Deliberately declares no `serviceAccountProviderId`, unlike every sibling
* Google service. A Google service-account JWT cannot reach user-scoped Chat
* data without domain-wide delegation, so offering service-account auth here
* would surface a credential path that always fails. Enterprises that
* authenticate other Google connectors through a delegated service account must
* attach a per-user OAuth credential for Chat.
*/
'google-chat': {
name: 'Google Chat',
description: 'Read Google Chat spaces and messages the signed-in user can access.',