feat(core): Introduce object store service (#7225)

Depends on https://github.com/n8n-io/n8n/pull/7220 | Story:
[PAY-840](https://linear.app/n8n/issue/PAY-840/introduce-object-store-service-and-manager-for-binary-data)

This PR introduces an object store service for Enterprise edition. Note
that the service is tested but currently unused - it will be integrated
soon as a binary data manager, and later for execution data.
`amazonaws.com` in the host is temporarily hardcoded until we integrate
the service and test against AWS, Cloudflare and Backblaze, in the next
PR.

This is ready for review - the PR it depends on is approved and waiting
for CI.

---------

Co-authored-by: कारतोफ्फेलस्क्रिप्ट™ <aditya@netroy.in>
This commit is contained in:
Iván Ovejero 2023-09-27 09:42:35 +02:00 committed by GitHub
parent 2c4e25c06b
commit fa845453bb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 12628 additions and 4086 deletions

View file

@ -505,9 +505,9 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
const date = new Date();
date.setHours(date.getHours() - 1);
const executionIds = (
const workflowIdsAndExecutionIds = (
await this.find({
select: ['id'],
select: ['workflowId', 'id'],
where: {
deletedAt: LessThanOrEqual(DateUtils.mixedDateToUtcDatetimeString(date)),
},
@ -519,14 +519,16 @@ export class ExecutionRepository extends Repository<ExecutionEntity> {
*/
withDeleted: true,
})
).map(({ id }) => id);
).map(({ id: executionId, workflowId }) => ({ workflowId, executionId }));
const executionIds = workflowIdsAndExecutionIds.map((o) => o.executionId);
if (executionIds.length === 0) {
this.logger.debug('Found no executions to hard-delete from database');
return;
}
await this.binaryDataService.deleteManyByExecutionIds(executionIds);
await this.binaryDataService.deleteMany(workflowIdsAndExecutionIds);
this.logger.debug(`Hard-deleting ${executionIds.length} executions from database`, {
executionIds,

View file

@ -34,6 +34,7 @@
"bin"
],
"devDependencies": {
"@types/aws4": "^1.5.1",
"@types/concat-stream": "^2.0.0",
"@types/cron": "~1.7.1",
"@types/crypto-js": "^4.0.1",
@ -41,13 +42,15 @@
"@types/lodash": "^4.14.195",
"@types/mime-types": "^2.1.0",
"@types/request-promise-native": "~1.0.15",
"@types/uuid": "^8.3.2"
"@types/uuid": "^8.3.2",
"@types/xml2js": "^0.4.11"
},
"peerDependencies": {
"n8n-nodes-base": "workspace:*"
},
"dependencies": {
"@n8n/client-oauth2": "workspace:*",
"aws4": "^1.8.0",
"axios": "^0.21.1",
"concat-stream": "^2.0.0",
"cron": "~1.7.2",
@ -64,6 +67,7 @@
"pretty-bytes": "^5.6.0",
"qs": "^6.10.1",
"typedi": "^0.10.0",
"uuid": "^8.3.2"
"uuid": "^8.3.2",
"xml2js": "^0.5.0"
}
}

View file

@ -1,13 +1,13 @@
/* eslint-disable @typescript-eslint/naming-convention */
import { readFile, stat } from 'fs/promises';
import concatStream from 'concat-stream';
import prettyBytes from 'pretty-bytes';
import { Service } from 'typedi';
import { BINARY_ENCODING, LoggerProxy as Logger, IBinaryData } from 'n8n-workflow';
import { FileSystemManager } from './FileSystem.manager';
import { UnknownBinaryDataManager, InvalidBinaryDataMode } from './errors';
import { LogCatch } from '../decorators/LogCatch.decorator';
import { areValidModes } from './utils';
import { areValidModes, toBuffer } from './utils';
import type { Readable } from 'stream';
import type { BinaryData } from './types';
@ -15,8 +15,6 @@ import type { INodeExecutionData } from 'n8n-workflow';
@Service()
export class BinaryDataService {
private availableModes: BinaryData.Mode[] = [];
private mode: BinaryData.Mode = 'default';
private managers: Record<string, BinaryData.Manager> = {};
@ -24,10 +22,10 @@ export class BinaryDataService {
async init(config: BinaryData.Config) {
if (!areValidModes(config.availableModes)) throw new InvalidBinaryDataMode();
this.availableModes = config.availableModes;
this.mode = config.mode;
if (this.availableModes.includes('filesystem')) {
if (config.availableModes.includes('filesystem')) {
const { FileSystemManager } = await import('./FileSystem.manager');
this.managers.filesystem = new FileSystemManager(config.localStoragePath);
await this.managers.filesystem.init();
@ -80,7 +78,7 @@ export class BinaryDataService {
const manager = this.managers[this.mode];
if (!manager) {
const buffer = await this.binaryToBuffer(bufferOrStream);
const buffer = await this.toBuffer(bufferOrStream);
binaryData.data = buffer.toString(BINARY_ENCODING);
binaryData.fileSize = prettyBytes(buffer.length);
@ -106,11 +104,8 @@ export class BinaryDataService {
return binaryData;
}
async binaryToBuffer(body: Buffer | Readable) {
return new Promise<Buffer>((resolve) => {
if (Buffer.isBuffer(body)) resolve(body);
else body.pipe(concatStream(resolve));
});
async toBuffer(bufferOrStream: Buffer | Readable) {
return toBuffer(bufferOrStream);
}
async getAsStream(binaryDataId: string, chunkSize?: number) {
@ -141,12 +136,12 @@ export class BinaryDataService {
return this.getManager(mode).getMetadata(fileId);
}
async deleteManyByExecutionIds(executionIds: string[]) {
async deleteMany(ids: BinaryData.IdsForDeletion) {
const manager = this.managers[this.mode];
if (!manager) return;
await manager.deleteManyByExecutionIds(executionIds);
await manager.deleteMany(ids);
}
@LogCatch((error) =>

View file

@ -59,7 +59,7 @@ export class FileSystemManager implements BinaryData.Manager {
bufferOrStream: Buffer | Readable,
{ mimeType, fileName }: BinaryData.PreWriteMetadata,
) {
const fileId = this.createFileId(executionId);
const fileId = this.toFileId(executionId);
const filePath = this.getPath(fileId);
await fs.writeFile(filePath, bufferOrStream);
@ -77,10 +77,11 @@ export class FileSystemManager implements BinaryData.Manager {
return fs.rm(filePath);
}
async deleteManyByExecutionIds(executionIds: string[]) {
async deleteMany(ids: BinaryData.IdsForDeletion) {
const executionIds = ids.map((o) => o.executionId);
const set = new Set(executionIds);
const fileNames = await fs.readdir(this.storagePath);
const deletedIds = [];
for (const fileName of fileNames) {
const executionId = fileName.match(EXECUTION_ID_EXTRACTOR)?.[1];
@ -89,12 +90,8 @@ export class FileSystemManager implements BinaryData.Manager {
const filePath = this.resolvePath(fileName);
await Promise.all([fs.rm(filePath), fs.rm(`${filePath}.metadata`)]);
deletedIds.push(executionId);
}
}
return deletedIds;
}
async copyByFilePath(
@ -103,7 +100,7 @@ export class FileSystemManager implements BinaryData.Manager {
filePath: string,
{ mimeType, fileName }: BinaryData.PreWriteMetadata,
) {
const newFileId = this.createFileId(executionId);
const newFileId = this.toFileId(executionId);
await fs.cp(filePath, this.getPath(newFileId));
@ -114,12 +111,14 @@ export class FileSystemManager implements BinaryData.Manager {
return { fileId: newFileId, fileSize };
}
async copyByFileId(_workflowId: string, executionId: string, fileId: string) {
const newFileId = this.createFileId(executionId);
async copyByFileId(_workflowId: string, executionId: string, sourceFileId: string) {
const targetFileId = this.toFileId(executionId);
const sourcePath = this.resolvePath(sourceFileId);
const targetPath = this.resolvePath(targetFileId);
await fs.copyFile(this.resolvePath(fileId), this.resolvePath(newFileId));
await fs.copyFile(sourcePath, targetPath);
return newFileId;
return targetFileId;
}
async rename(oldFileId: string, newFileId: string) {
@ -136,7 +135,7 @@ export class FileSystemManager implements BinaryData.Manager {
// private methods
// ----------------------------------
private createFileId(executionId: string) {
private toFileId(executionId: string) {
return [executionId, uuid()].join('');
}

View file

@ -20,6 +20,8 @@ export namespace BinaryData {
export type PreWriteMetadata = Omit<Metadata, 'fileSize'>;
export type IdsForDeletion = Array<{ workflowId: string; executionId: string }>;
export interface Manager {
init(): Promise<void>;
@ -35,7 +37,7 @@ export namespace BinaryData {
getAsStream(fileId: string, chunkSize?: number): Promise<Readable>;
getMetadata(fileId: string): Promise<Metadata>;
copyByFileId(workflowId: string, executionId: string, fileId: string): Promise<string>;
copyByFileId(workflowId: string, executionId: string, sourceFileId: string): Promise<string>;
copyByFilePath(
workflowId: string,
executionId: string,
@ -44,7 +46,7 @@ export namespace BinaryData {
): Promise<WriteResult>;
deleteOne(fileId: string): Promise<void>;
deleteManyByExecutionIds(executionIds: string[]): Promise<string[]>;
deleteMany(ids: IdsForDeletion): Promise<void>;
rename(oldFileId: string, newFileId: string): Promise<void>;
}

View file

@ -1,5 +1,7 @@
import fs from 'fs/promises';
import fs from 'node:fs/promises';
import type { Readable } from 'node:stream';
import type { BinaryData } from './types';
import concatStream from 'concat-stream';
/**
* Modes for storing binary data:
@ -20,3 +22,10 @@ export async function ensureDirExists(dir: string) {
await fs.mkdir(dir, { recursive: true });
}
}
export async function toBuffer(body: Buffer | Readable) {
return new Promise<Buffer>((resolve) => {
if (Buffer.isBuffer(body)) resolve(body);
else body.pipe(concatStream(resolve));
});
}

View file

@ -775,7 +775,7 @@ export async function proxyRequestToAxios(
if (Buffer.isBuffer(responseData) || responseData instanceof Readable) {
responseData = await Container.get(BinaryDataService)
.binaryToBuffer(responseData)
.toBuffer(responseData)
.then((buffer) => buffer.toString('utf-8'));
}
@ -2589,7 +2589,7 @@ const getBinaryHelperFunctions = (
getBinaryStream,
getBinaryMetadata,
binaryToBuffer: async (body: Buffer | Readable) =>
Container.get(BinaryDataService).binaryToBuffer(body),
Container.get(BinaryDataService).toBuffer(body),
prepareBinaryData: async (binaryData, filePath, mimeType) =>
prepareBinaryData(binaryData, executionId!, workflowId, filePath, mimeType),
setBinaryDataBuffer: async (data, binaryData) =>
@ -2851,7 +2851,7 @@ export function getExecuteFunctions(
return dataProxy.getDataProxy();
},
binaryToBuffer: async (body: Buffer | Readable) =>
Container.get(BinaryDataService).binaryToBuffer(body),
Container.get(BinaryDataService).toBuffer(body),
async putExecutionToWait(waitTill: Date): Promise<void> {
runExecutionData.waitTill = waitTill;
if (additionalData.setExecutionStatus) {

View file

@ -0,0 +1,232 @@
/* eslint-disable @typescript-eslint/naming-convention */
import { createHash } from 'node:crypto';
import axios from 'axios';
import { Service } from 'typedi';
import { sign } from 'aws4';
import { isStream, parseXml } from './utils';
import { ExternalStorageRequestFailed } from './errors';
import type { AxiosRequestConfig, Method } from 'axios';
import type { Request as Aws4Options, Credentials as Aws4Credentials } from 'aws4';
import type { ListPage, ObjectStore, RawListPage } from './types';
import type { Readable } from 'stream';
import type { BinaryData } from '..';
@Service()
export class ObjectStoreService {
private credentials: Aws4Credentials;
constructor(
private bucket: { region: string; name: string },
credentials: { accountId: string; secretKey: string },
) {
this.credentials = {
accessKeyId: credentials.accountId,
secretAccessKey: credentials.secretKey,
};
}
get host() {
return `${this.bucket.name}.s3.${this.bucket.region}.amazonaws.com`;
}
/**
* Confirm that the configured bucket exists and the caller has permission to access it.
*
* @doc https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadBucket.html
*/
async checkConnection() {
return this.request('HEAD', this.host);
}
/**
* Upload an object to the configured bucket.
*
* @doc https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html
*/
async put(filename: string, buffer: Buffer, metadata: BinaryData.PreWriteMetadata = {}) {
const headers: Record<string, string | number> = {
'Content-Length': buffer.length,
'Content-MD5': createHash('md5').update(buffer).digest('base64'),
};
if (metadata.fileName) headers['x-amz-meta-filename'] = metadata.fileName;
if (metadata.mimeType) headers['Content-Type'] = metadata.mimeType;
return this.request('PUT', this.host, `/${filename}`, { headers, body: buffer });
}
/**
* Download an object as a stream or buffer from the configured bucket.
*
* @doc https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html
*/
async get(path: string, { mode }: { mode: 'buffer' }): Promise<Buffer>;
async get(path: string, { mode }: { mode: 'stream' }): Promise<Readable>;
async get(path: string, { mode }: { mode: 'stream' | 'buffer' }) {
const { data } = await this.request('GET', this.host, path, {
responseType: mode === 'buffer' ? 'arraybuffer' : 'stream',
});
if (mode === 'stream' && isStream(data)) return data;
if (mode === 'buffer' && Buffer.isBuffer(data)) return data;
throw new TypeError(`Expected ${mode} but received ${typeof data}.`);
}
/**
* Retrieve metadata for an object in the configured bucket.
*
* @doc https://docs.aws.amazon.com/AmazonS3/latest/userguide/UsingMetadata.html
*/
async getMetadata(path: string) {
type Response = {
headers: {
'content-length': string;
'content-type'?: string;
'x-amz-meta-filename'?: string;
} & Record<string, string | number>;
};
const response: Response = await this.request('HEAD', this.host, path);
return response.headers;
}
/**
* Delete an object in the configured bucket.
*
* @doc https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html
*/
async deleteOne(path: string) {
return this.request('DELETE', this.host, `/${encodeURIComponent(path)}`);
}
/**
* Delete objects with a common prefix in the configured bucket.
*
* @doc https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html
*/
async deleteMany(prefix: string) {
const objects = await this.list(prefix);
const innerXml = objects.map(({ key }) => `<Object><Key>${key}</Key></Object>`).join('\n');
const body = ['<Delete>', innerXml, '</Delete>'].join('\n');
const headers = {
'Content-Type': 'application/xml',
'Content-Length': body.length,
'Content-MD5': createHash('md5').update(body).digest('base64'),
};
return this.request('POST', this.host, '/?delete', { headers, body });
}
/**
* List objects with a common prefix in the configured bucket.
*
* @doc https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html
*/
async list(prefix: string) {
const items = [];
let isTruncated;
let nextPageToken;
do {
const listPage = await this.getListPage(prefix, nextPageToken);
if (listPage.contents?.length > 0) items.push(...listPage.contents);
isTruncated = listPage.isTruncated;
nextPageToken = listPage.nextContinuationToken;
} while (isTruncated && nextPageToken);
return items;
}
/**
* Fetch a page of objects with a common prefix in the configured bucket. Max 1000 per page.
*/
async getListPage(prefix: string, nextPageToken?: string) {
const bucketlessHost = this.host.split('.').slice(1).join('.');
const qs: Record<string, string | number> = { 'list-type': 2, prefix };
if (nextPageToken) qs['continuation-token'] = nextPageToken;
const { data } = await this.request('GET', bucketlessHost, `/${this.bucket.name}`, { qs });
if (typeof data !== 'string') {
throw new TypeError(`Expected XML string but received ${typeof data}`);
}
const { listBucketResult: page } = await parseXml<RawListPage>(data);
if (!page.contents) return { ...page, contents: [] };
// `explicitArray: false` removes array wrapper on single item array, so restore it
if (!Array.isArray(page.contents)) page.contents = [page.contents];
// remove null prototype - https://github.com/Leonidas-from-XIV/node-xml2js/issues/670
page.contents.forEach((item) => {
Object.setPrototypeOf(item, Object.prototype);
});
return page as ListPage;
}
private toPath(rawPath: string, qs?: Record<string, string | number>) {
const path = rawPath.startsWith('/') ? rawPath : `/${rawPath}`;
if (!qs) return path;
const qsParams = Object.entries(qs)
.map(([key, value]) => `${key}=${value}`)
.join('&');
return path.concat(`?${qsParams}`);
}
private async request<T = unknown>(
method: Method,
host: string,
rawPath = '',
{ qs, headers, body, responseType }: ObjectStore.RequestOptions = {},
) {
const path = this.toPath(rawPath, qs);
const optionsToSign: Aws4Options = {
method,
service: 's3',
region: this.bucket.region,
host,
path,
};
if (headers) optionsToSign.headers = headers;
if (body) optionsToSign.body = body;
const signedOptions = sign(optionsToSign, this.credentials);
const config: AxiosRequestConfig = {
method,
url: `https://${host}${path}`,
headers: signedOptions.headers,
};
if (body) config.data = body;
if (responseType) config.responseType = responseType;
try {
return await axios.request<T>(config);
} catch (error) {
throw new ExternalStorageRequestFailed(error, config);
}
}
}

View file

@ -0,0 +1,8 @@
import { AxiosRequestConfig } from 'axios';
export class ExternalStorageRequestFailed extends Error {
constructor(error: unknown, details: AxiosRequestConfig) {
super('Request to external object storage failed');
this.cause = { error, details };
}
}

View file

@ -0,0 +1,32 @@
import type { ResponseType } from 'axios';
export type RawListPage = {
listBucketResult: {
name: string;
prefix: string;
keyCount: number;
maxKeys: number;
isTruncated: boolean;
nextContinuationToken?: string; // only if isTruncated is true
contents?: Item | Item[];
};
};
type Item = {
key: string;
lastModified: string;
eTag: string;
size: number; // bytes
storageClass: string;
};
export type ListPage = Omit<RawListPage['listBucketResult'], 'contents'> & { contents: Item[] };
export namespace ObjectStore {
export type RequestOptions = {
qs?: Record<string, string | number>;
headers?: Record<string, string | number>;
body?: string | Buffer;
responseType?: ResponseType;
};
}

View file

@ -0,0 +1,16 @@
import { Stream } from 'node:stream';
import { parseStringPromise } from 'xml2js';
import { firstCharLowerCase, parseBooleans, parseNumbers } from 'xml2js/lib/processors';
export function isStream(maybeStream: unknown): maybeStream is Stream {
return maybeStream instanceof Stream;
}
export async function parseXml<T>(xml: string): Promise<T> {
return parseStringPromise(xml, {
explicitArray: false,
ignoreAttrs: true,
tagNameProcessors: [firstCharLowerCase],
valueProcessors: [parseNumbers, parseBooleans],
}) as Promise<T>;
}

View file

@ -16,3 +16,4 @@ export * from './NodeExecuteFunctions';
export * from './WorkflowExecute';
export { NodeExecuteFunctions, UserSettings };
export * from './errors';
export { ObjectStoreService } from './ObjectStore/ObjectStore.service.ee';

View file

@ -44,6 +44,7 @@ describe('NodeExecuteFunctions', () => {
data: 'This should be overwritten by the actual payload in the response',
},
inputData,
'workflowId',
'executionId',
);
@ -95,6 +96,7 @@ describe('NodeExecuteFunctions', () => {
data: 'This should be overwritten with the name of the configured data manager',
},
inputData,
'workflowId',
'executionId',
);

View file

@ -0,0 +1,301 @@
import axios from 'axios';
import { ObjectStoreService } from '../src/ObjectStore/ObjectStore.service.ee';
import { Readable } from 'stream';
jest.mock('axios');
const mockAxios = axios as jest.Mocked<typeof axios>;
const MOCK_BUCKET = { region: 'us-east-1', name: 'test-bucket' };
const MOCK_CREDENTIALS = { accountId: 'mock-account-id', secretKey: 'mock-secret-key' };
const FAILED_REQUEST_ERROR_MESSAGE = 'Request to external object storage failed';
const EXPECTED_HOST = `${MOCK_BUCKET.name}.s3.${MOCK_BUCKET.region}.amazonaws.com`;
const MOCK_S3_ERROR = new Error('Something went wrong!');
const toMultipleDeletionXml = (filename: string) => `<Delete>
<Object><Key>${filename}</Key></Object>
</Delete>`;
describe('ObjectStoreService', () => {
let objectStoreService: ObjectStoreService;
beforeEach(() => {
objectStoreService = new ObjectStoreService(MOCK_BUCKET, MOCK_CREDENTIALS);
jest.restoreAllMocks();
});
describe('checkConnection()', () => {
it('should send a HEAD request to the correct host', async () => {
mockAxios.request.mockResolvedValue({ status: 200 });
await objectStoreService.checkConnection();
expect(mockAxios.request).toHaveBeenCalledWith(
expect.objectContaining({
method: 'HEAD',
url: `https://${EXPECTED_HOST}/`,
headers: expect.objectContaining({
'X-Amz-Content-Sha256': expect.any(String),
'X-Amz-Date': expect.any(String),
Authorization: expect.any(String),
}),
}),
);
});
it('should throw an error on request failure', async () => {
mockAxios.request.mockRejectedValue(MOCK_S3_ERROR);
const promise = objectStoreService.checkConnection();
await expect(promise).rejects.toThrowError(FAILED_REQUEST_ERROR_MESSAGE);
});
});
describe('getMetadata()', () => {
it('should send a HEAD request to the correct host and path', async () => {
const path = 'file.txt';
mockAxios.request.mockResolvedValue({ status: 200 });
await objectStoreService.getMetadata(path);
expect(mockAxios.request).toHaveBeenCalledWith(
expect.objectContaining({
method: 'HEAD',
url: `https://${EXPECTED_HOST}/${path}`,
headers: expect.objectContaining({
Host: EXPECTED_HOST,
'X-Amz-Content-Sha256': expect.any(String),
'X-Amz-Date': expect.any(String),
Authorization: expect.any(String),
}),
}),
);
});
it('should throw an error on request failure', async () => {
const path = 'file.txt';
mockAxios.request.mockRejectedValue(MOCK_S3_ERROR);
const promise = objectStoreService.getMetadata(path);
await expect(promise).rejects.toThrowError(FAILED_REQUEST_ERROR_MESSAGE);
});
});
describe('put()', () => {
it('should send a PUT request to upload an object', async () => {
const path = 'file.txt';
const buffer = Buffer.from('Test content');
const metadata = { fileName: path, mimeType: 'text/plain' };
mockAxios.request.mockResolvedValue({ status: 200 });
await objectStoreService.put(path, buffer, metadata);
expect(mockAxios.request).toHaveBeenCalledWith(
expect.objectContaining({
method: 'PUT',
url: `https://${EXPECTED_HOST}/${path}`,
headers: expect.objectContaining({
'Content-Length': buffer.length,
'Content-MD5': expect.any(String),
'x-amz-meta-filename': metadata.fileName,
'Content-Type': metadata.mimeType,
}),
data: buffer,
}),
);
});
it('should throw an error on request failure', async () => {
const path = 'file.txt';
const buffer = Buffer.from('Test content');
const metadata = { fileName: path, mimeType: 'text/plain' };
mockAxios.request.mockRejectedValue(MOCK_S3_ERROR);
const promise = objectStoreService.put(path, buffer, metadata);
await expect(promise).rejects.toThrowError(FAILED_REQUEST_ERROR_MESSAGE);
});
});
describe('get()', () => {
it('should send a GET request to download an object as a buffer', async () => {
const path = 'file.txt';
mockAxios.request.mockResolvedValue({ status: 200, data: Buffer.from('Test content') });
const result = await objectStoreService.get(path, { mode: 'buffer' });
expect(mockAxios.request).toHaveBeenCalledWith(
expect.objectContaining({
method: 'GET',
url: `https://${EXPECTED_HOST}/${path}`,
responseType: 'arraybuffer',
}),
);
expect(Buffer.isBuffer(result)).toBe(true);
});
it('should send a GET request to download an object as a stream', async () => {
const path = 'file.txt';
mockAxios.request.mockResolvedValue({ status: 200, data: new Readable() });
const result = await objectStoreService.get(path, { mode: 'stream' });
expect(mockAxios.request).toHaveBeenCalledWith(
expect.objectContaining({
method: 'GET',
url: `https://${EXPECTED_HOST}/${path}`,
responseType: 'stream',
}),
);
expect(result instanceof Readable).toBe(true);
});
it('should throw an error on request failure', async () => {
const path = 'file.txt';
mockAxios.request.mockRejectedValue(MOCK_S3_ERROR);
const promise = objectStoreService.get(path, { mode: 'buffer' });
await expect(promise).rejects.toThrowError(FAILED_REQUEST_ERROR_MESSAGE);
});
});
describe('deleteOne()', () => {
it('should send a DELETE request to delete an object', async () => {
const path = 'file.txt';
mockAxios.request.mockResolvedValue({ status: 204 });
await objectStoreService.deleteOne(path);
expect(mockAxios.request).toHaveBeenCalledWith(
expect.objectContaining({
method: 'DELETE',
url: `https://${EXPECTED_HOST}/${path}`,
}),
);
});
it('should throw an error on request failure', async () => {
const path = 'file.txt';
mockAxios.request.mockRejectedValue(MOCK_S3_ERROR);
const promise = objectStoreService.deleteOne(path);
await expect(promise).rejects.toThrowError(FAILED_REQUEST_ERROR_MESSAGE);
});
});
describe('deleteMany()', () => {
it('should send a POST request to delete multiple objects', async () => {
const prefix = 'test-dir/';
const fileName = 'file.txt';
const mockList = [
{
key: fileName,
lastModified: '2023-09-24T12:34:56Z',
eTag: 'abc123def456',
size: 456789,
storageClass: 'STANDARD',
},
];
objectStoreService.list = jest.fn().mockResolvedValue(mockList);
mockAxios.request.mockResolvedValue({ status: 204 });
await objectStoreService.deleteMany(prefix);
expect(mockAxios.request).toHaveBeenCalledWith(
expect.objectContaining({
method: 'POST',
url: `https://${EXPECTED_HOST}/?delete`,
headers: expect.objectContaining({
'Content-Type': 'application/xml',
'Content-Length': expect.any(Number),
'Content-MD5': expect.any(String),
}),
data: toMultipleDeletionXml(fileName),
}),
);
});
it('should throw an error on request failure', async () => {
const prefix = 'test-dir/';
mockAxios.request.mockRejectedValue(MOCK_S3_ERROR);
const promise = objectStoreService.deleteMany(prefix);
await expect(promise).rejects.toThrowError(FAILED_REQUEST_ERROR_MESSAGE);
});
});
describe('list()', () => {
it('should list objects with a common prefix', async () => {
const prefix = 'test-dir/';
const mockListPage = {
contents: [{ key: `${prefix}file1.txt` }, { key: `${prefix}file2.txt` }],
isTruncated: false,
};
objectStoreService.getListPage = jest.fn().mockResolvedValue(mockListPage);
mockAxios.request.mockResolvedValue({ status: 200 });
const result = await objectStoreService.list(prefix);
expect(result).toEqual(mockListPage.contents);
});
it('should consolidate pages', async () => {
const prefix = 'test-dir/';
const mockFirstListPage = {
contents: [{ key: `${prefix}file1.txt` }],
isTruncated: true,
nextContinuationToken: 'token1',
};
const mockSecondListPage = {
contents: [{ key: `${prefix}file2.txt` }],
isTruncated: false,
};
objectStoreService.getListPage = jest
.fn()
.mockResolvedValueOnce(mockFirstListPage)
.mockResolvedValueOnce(mockSecondListPage);
mockAxios.request.mockResolvedValue({ status: 200 });
const result = await objectStoreService.list(prefix);
expect(result).toEqual([...mockFirstListPage.contents, ...mockSecondListPage.contents]);
});
it('should throw an error on request failure', async () => {
const prefix = 'test-dir/';
mockAxios.request.mockRejectedValue(MOCK_S3_ERROR);
const promise = objectStoreService.list(prefix);
await expect(promise).rejects.toThrowError(FAILED_REQUEST_ERROR_MESSAGE);
});
});
});

File diff suppressed because it is too large Load diff