CommunitySolidServer/test/integration/LockingResourceStore.test.ts

import { RootContainerInitializer } from '../../src/init/RootContainerInitializer';
import { BasicRepresentation } from '../../src/ldp/representation/BasicRepresentation';
import type { Representation } from '../../src/ldp/representation/Representation';
import { InMemoryDataAccessor } from '../../src/storage/accessors/InMemoryDataAccessor';
import { DataAccessorBasedStore } from '../../src/storage/DataAccessorBasedStore';
import { LockingResourceStore } from '../../src/storage/LockingResourceStore';
import type { ResourceStore } from '../../src/storage/ResourceStore';
import { APPLICATION_OCTET_STREAM } from '../../src/util/ContentTypes';
import { InternalServerError } from '../../src/util/errors/InternalServerError';
import { SingleRootIdentifierStrategy } from '../../src/util/identifiers/SingleRootIdentifierStrategy';
import type { ExpiringResourceLocker } from '../../src/util/locking/ExpiringResourceLocker';
import type { ResourceLocker } from '../../src/util/locking/ResourceLocker';
import { SingleThreadedResourceLocker } from '../../src/util/locking/SingleThreadedResourceLocker';
import { WrappedExpiringResourceLocker } from '../../src/util/locking/WrappedExpiringResourceLocker';
import { guardedStreamFrom } from '../../src/util/StreamUtil';
import { BASE } from './Config';

jest.useFakeTimers();

describe('A LockingResourceStore', (): void => {
  let path: string;
  let store: LockingResourceStore;
  let locker: ResourceLocker;
  let expiringLocker: ExpiringResourceLocker;
  let source: ResourceStore;
  let getRepresentationSpy: jest.SpyInstance;

  beforeEach(async(): Promise<void> => {
    jest.clearAllMocks();

    const base = 'http://test.com/';
    path = `${base}path`;
    source = new DataAccessorBasedStore(new InMemoryDataAccessor(base), new SingleRootIdentifierStrategy(base));

    // Initialize store
    const initializer = new RootContainerInitializer({ store: source, baseUrl: BASE });
    await initializer.handleSafe();

    locker = new SingleThreadedResourceLocker();
    expiringLocker = new WrappedExpiringResourceLocker(locker, 1000);

    store = new LockingResourceStore(source, expiringLocker);

    // Spy on a real ResourceLocker and ResourceStore instance
    getRepresentationSpy = jest.spyOn(source, 'getRepresentation');
    getRepresentationSpy.mockReturnValue(new Promise((resolve): any => resolve({ data:
        guardedStreamFrom([ 1, 2, 3 ]) } as Representation)));

    // Make sure something is in the store before we read from it in our tests.
    await source.setRepresentation({ path }, new BasicRepresentation([ 1, 2, 3 ], APPLICATION_OCTET_STREAM));
  });

  it('destroys the stream when nothing is read after 1000ms.', async(): Promise<void> => {
    const representation = await store.getRepresentation({ path }, {});
    const errorCallback = jest.fn();
    representation.data.on('error', errorCallback);

    // Wait 1000ms and read
    jest.advanceTimersByTime(1000);
    await new Promise(setImmediate);
    expect(representation.data.destroyed).toBe(true);

    // Verify a timeout error was thrown
    expect(errorCallback).toHaveBeenCalledTimes(1);
    expect(errorCallback).toHaveBeenLastCalledWith(new InternalServerError(`Lock expired after 1000ms on ${path}`));

    // Verify the lock was acquired and released at the right time
    expect(getRepresentationSpy).toHaveBeenCalledTimes(1);
  });

  it('destroys the stream when pauses between reads exceed 1000ms.', async(): Promise<void> => {
    const representation = await store.getRepresentation({ path }, {});
    const errorCallback = jest.fn();
    representation.data.on('error', errorCallback);

    // Wait 750ms and read
    jest.advanceTimersByTime(750);
    expect(representation.data.destroyed).toBe(false);
    representation.data.read();

    // Wait 750ms and read
    jest.advanceTimersByTime(750);
    expect(representation.data.destroyed).toBe(false);
    representation.data.read();

    // Wait 1000ms and watch the stream be destroyed
    jest.advanceTimersByTime(1000);
    await new Promise(setImmediate);
    expect(representation.data.destroyed).toBe(true);

    // Verify a timeout error was thrown
    expect(errorCallback).toHaveBeenCalledTimes(1);
    expect(errorCallback).toHaveBeenLastCalledWith(new InternalServerError(`Lock expired after 1000ms on ${path}`));

    // Verify the lock was acquired and released at the right time
    expect(getRepresentationSpy).toHaveBeenCalledTimes(1);
  });
});