Merge pull request #1377 from 0xProject/feature/pipeline-cleanup-mega-rebase

Merge all pipeline code into development

22 files changed, 1196 insertions, 0 deletions

diff --git a/packages/pipeline/test/data_sources/ohlcv_external/crypto_compare_test.ts b/packages/pipeline/test/data_sources/ohlcv_external/crypto_compare_test.ts
new file mode 100644
index 000000000..cb374bbb1
--- /dev/null
+++ b/packages/pipeline/test/data_sources/ohlcv_external/crypto_compare_test.ts
@@ -0,0 +1,47 @@
+import * as chai from 'chai';
+import 'mocha';
+import * as R from 'ramda';
+
+import { CryptoCompareOHLCVSource } from '../../../src/data_sources/ohlcv_external/crypto_compare';
+import { TradingPair } from '../../../src/utils/get_ohlcv_trading_pairs';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('ohlcv_external data source (Crypto Compare)', () => {
+    describe('generateBackfillIntervals', () => {
+        it('generates pairs with intervals to query', () => {
+            const source = new CryptoCompareOHLCVSource();
+            const pair: TradingPair = {
+                fromSymbol: 'ETH',
+                toSymbol: 'ZRX',
+                latestSavedTime: new Date().getTime() - source.interval * 2,
+            };
+
+            const expected = [
+                pair,
+                R.merge(pair, { latestSavedTime: pair.latestSavedTime + source.interval }),
+                R.merge(pair, { latestSavedTime: pair.latestSavedTime + source.interval * 2 }),
+            ];
+
+            const actual = source.generateBackfillIntervals(pair);
+            expect(actual).deep.equal(expected);
+        });
+
+        it('returns single pair if no backfill is needed', () => {
+            const source = new CryptoCompareOHLCVSource();
+            const pair: TradingPair = {
+                fromSymbol: 'ETH',
+                toSymbol: 'ZRX',
+                latestSavedTime: new Date().getTime() - source.interval + 5000,
+            };
+
+            const expected = [pair];
+
+            const actual = source.generateBackfillIntervals(pair);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});
diff --git a/packages/pipeline/test/db_global_hooks.ts b/packages/pipeline/test/db_global_hooks.ts
new file mode 100644
index 000000000..dfee02c45
--- /dev/null
+++ b/packages/pipeline/test/db_global_hooks.ts
@@ -0,0 +1,9 @@
+import { setUpDbAsync, tearDownDbAsync } from './db_setup';
+
+before('set up database', async () => {
+    await setUpDbAsync();
+});
+
+after('tear down database', async () => {
+    await tearDownDbAsync();
+});
diff --git a/packages/pipeline/test/db_setup.ts b/packages/pipeline/test/db_setup.ts
new file mode 100644
index 000000000..bf31d15b6
--- /dev/null
+++ b/packages/pipeline/test/db_setup.ts
@@ -0,0 +1,174 @@
+import * as Docker from 'dockerode';
+import * as fs from 'fs';
+import * as R from 'ramda';
+import { Connection, ConnectionOptions, createConnection } from 'typeorm';
+
+import * as ormConfig from '../src/ormconfig';
+
+// The name of the image to pull and use for the container. This also affects
+// which version of Postgres we use.
+const DOCKER_IMAGE_NAME = 'postgres:11-alpine';
+// The name to use for the Docker container which will run Postgres.
+const DOCKER_CONTAINER_NAME = '0x_pipeline_postgres_test';
+// The port which will be exposed on the Docker container.
+const POSTGRES_HOST_PORT = '15432';
+// Number of milliseconds to wait for postgres to finish initializing after
+// starting the docker container.
+const POSTGRES_SETUP_DELAY_MS = 5000;
+
+/**
+ * Sets up the database for testing purposes. If the
+ * ZEROEX_DATA_PIPELINE_TEST_DB_URL env var is specified, it will create a
+ * connection using that url. Otherwise it will spin up a new Docker container
+ * with a Postgres database and then create a connection to that database.
+ */
+export async function setUpDbAsync(): Promise<void> {
+    const connection = await createDbConnectionOnceAsync();
+    await connection.runMigrations({ transaction: true });
+}
+
+/**
+ * Tears down the database used for testing. This completely destroys any data.
+ * If a docker container was created, it destroys that container too.
+ */
+export async function tearDownDbAsync(): Promise<void> {
+    const connection = await createDbConnectionOnceAsync();
+    for (const _ of connection.migrations) {
+        await connection.undoLastMigration({ transaction: true });
+    }
+    if (needsDocker()) {
+        const docker = initDockerOnce();
+        const postgresContainer = docker.getContainer(DOCKER_CONTAINER_NAME);
+        await postgresContainer.kill();
+        await postgresContainer.remove();
+    }
+}
+
+let savedConnection: Connection;
+
+/**
+ * The first time this is run, it creates and returns a new TypeORM connection.
+ * Each subsequent time, it returns the existing connection. This is helpful
+ * because only one TypeORM connection can be active at a time.
+ */
+export async function createDbConnectionOnceAsync(): Promise<Connection> {
+    if (savedConnection !== undefined) {
+        return savedConnection;
+    }
+
+    if (needsDocker()) {
+        await initContainerAsync();
+    }
+    const testDbUrl =
+        process.env.ZEROEX_DATA_PIPELINE_TEST_DB_URL ||
+        `postgresql://postgres@localhost:${POSTGRES_HOST_PORT}/postgres`;
+    const testOrmConfig = R.merge(ormConfig, { url: testDbUrl }) as ConnectionOptions;
+
+    savedConnection = await createConnection(testOrmConfig);
+    return savedConnection;
+}
+
+async function sleepAsync(ms: number): Promise<{}> {
+    return new Promise<{}>(resolve => setTimeout(resolve, ms));
+}
+
+let savedDocker: Docker;
+
+function initDockerOnce(): Docker {
+    if (savedDocker !== undefined) {
+        return savedDocker;
+    }
+
+    // Note(albrow): Code for determining the right socket path is partially
+    // based on https://github.com/apocas/dockerode/blob/8f3aa85311fab64d58eca08fef49aa1da5b5f60b/test/spec_helper.js
+    const isWin = require('os').type() === 'Windows_NT';
+    const socketPath = process.env.DOCKER_SOCKET || (isWin ? '//./pipe/docker_engine' : '/var/run/docker.sock');
+    const isSocket = fs.existsSync(socketPath) ? fs.statSync(socketPath).isSocket() : false;
+    if (!isSocket) {
+        throw new Error(`Failed to connect to Docker using socket path: "${socketPath}".
+
+The database integration tests need to be able to connect to a Postgres database. Make sure that Docker is running and accessible at the expected socket path. If Docker isn't working you have two options:
+
+    1) Set the DOCKER_SOCKET environment variable to a socket path that can be used to connect to Docker or
+    2) Set the ZEROEX_DATA_PIPELINE_TEST_DB_URL environment variable to connect directly to an existing Postgres database instead of trying to start Postgres via Docker
+`);
+    }
+    savedDocker = new Docker({
+        socketPath,
+    });
+    return savedDocker;
+}
+
+// Creates the container, waits for it to initialize, and returns it.
+async function initContainerAsync(): Promise<Docker.Container> {
+    const docker = initDockerOnce();
+
+    // Tear down any existing containers with the same name.
+    await tearDownExistingContainerIfAnyAsync();
+
+    // Pull the image we need.
+    await pullImageAsync(docker, DOCKER_IMAGE_NAME);
+
+    // Create the container.
+    const postgresContainer = await docker.createContainer({
+        name: DOCKER_CONTAINER_NAME,
+        Image: DOCKER_IMAGE_NAME,
+        ExposedPorts: {
+            '5432': {},
+        },
+        HostConfig: {
+            PortBindings: {
+                '5432': [
+                    {
+                        HostPort: POSTGRES_HOST_PORT,
+                    },
+                ],
+            },
+        },
+    });
+    await postgresContainer.start();
+    await sleepAsync(POSTGRES_SETUP_DELAY_MS);
+    return postgresContainer;
+}
+
+async function tearDownExistingContainerIfAnyAsync(): Promise<void> {
+    const docker = initDockerOnce();
+
+    // Check if a container with the desired name already exists. If so, this
+    // probably means we didn't clean up properly on the last test run.
+    const existingContainer = docker.getContainer(DOCKER_CONTAINER_NAME);
+    if (existingContainer != null) {
+        try {
+            await existingContainer.kill();
+        } catch {
+            // If this fails, it's fine. The container was probably already
+            // killed.
+        }
+        try {
+            await existingContainer.remove();
+        } catch {
+            // If this fails, it's fine. The container was probably already
+            // removed.
+        }
+    }
+}
+
+function needsDocker(): boolean {
+    return process.env.ZEROEX_DATA_PIPELINE_TEST_DB_URL === undefined;
+}
+
+// Note(albrow): This is partially based on
+// https://stackoverflow.com/questions/38258263/how-do-i-wait-for-a-pull
+async function pullImageAsync(docker: Docker, imageName: string): Promise<void> {
+    return new Promise<void>((resolve, reject) => {
+        docker.pull(imageName, {}, (err, stream) => {
+            if (err != null) {
+                reject(err);
+                return;
+            }
+            docker.modem.followProgress(stream, () => {
+                resolve();
+            });
+        });
+    });
+}
diff --git a/packages/pipeline/test/entities/block_test.ts b/packages/pipeline/test/entities/block_test.ts
new file mode 100644
index 000000000..503f284f0
--- /dev/null
+++ b/packages/pipeline/test/entities/block_test.ts
@@ -0,0 +1,23 @@
+import 'mocha';
+import 'reflect-metadata';
+
+import { Block } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+// tslint:disable:custom-no-magic-numbers
+describe('Block entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const block = new Block();
+        block.hash = '0x12345';
+        block.number = 1234567;
+        block.timestamp = 5432154321;
+        const blocksRepository = connection.getRepository(Block);
+        await testSaveAndFindEntityAsync(blocksRepository, block);
+    });
+});
diff --git a/packages/pipeline/test/entities/dex_trades_test.ts b/packages/pipeline/test/entities/dex_trades_test.ts
new file mode 100644
index 000000000..83aaeec8f
--- /dev/null
+++ b/packages/pipeline/test/entities/dex_trades_test.ts
@@ -0,0 +1,60 @@
+import { BigNumber } from '@0x/utils';
+import 'mocha';
+import * as R from 'ramda';
+import 'reflect-metadata';
+
+import { DexTrade } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const baseTrade = {
+    sourceUrl: 'https://bloxy.info/api/dex/trades',
+    txTimestamp: 1543447585938,
+    txDate: '2018-11-21',
+    txSender: '0x00923b9a074762b93650716333b3e1473a15048e',
+    smartContractId: 7091917,
+    smartContractAddress: '0x818e6fecd516ecc3849daf6845e3ec868087b755',
+    contractType: 'DEX/Kyber Network Proxy',
+    maker: '0xbf2179859fc6d5bee9bf9158632dc51678a4100c',
+    taker: '0xbf2179859fc6d5bee9bf9158632dc51678a4100d',
+    amountBuy: new BigNumber('1.011943163078103'),
+    makerFeeAmount: new BigNumber(0),
+    buyCurrencyId: 1,
+    buySymbol: 'ETH',
+    amountSell: new BigNumber('941.4997928436911'),
+    takerFeeAmount: new BigNumber(0),
+    sellCurrencyId: 16610,
+    sellSymbol: 'ELF',
+    makerAnnotation: '',
+    takerAnnotation: '',
+    protocol: 'Kyber Network Proxy',
+    sellAddress: '0xbf2179859fc6d5bee9bf9158632dc51678a4100e',
+};
+
+const tradeWithNullAddresses: DexTrade = R.merge(baseTrade, {
+    txHash: '0xb93a7faf92efbbb5405c9a73cd4efd99702fe27c03ff22baee1f1b1e37b3a0bf',
+    buyAddress: '0xbf2179859fc6d5bee9bf9158632dc51678a4100e',
+    sellAddress: '0xbf2179859fc6d5bee9bf9158632dc51678a4100f',
+});
+
+const tradeWithNonNullAddresses: DexTrade = R.merge(baseTrade, {
+    txHash: '0xb93a7faf92efbbb5405c9a73cd4efd99702fe27c03ff22baee1f1b1e37b3a0be',
+    buyAddress: null,
+    sellAddress: null,
+});
+
+// tslint:disable:custom-no-magic-numbers
+describe('DexTrade entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const trades = [tradeWithNullAddresses, tradeWithNonNullAddresses];
+        const tradesRepository = connection.getRepository(DexTrade);
+        for (const trade of trades) {
+            await testSaveAndFindEntityAsync(tradesRepository, trade);
+        }
+    });
+});
diff --git a/packages/pipeline/test/entities/exchange_cancel_event_test.ts b/packages/pipeline/test/entities/exchange_cancel_event_test.ts
new file mode 100644
index 000000000..f3b306d69
--- /dev/null
+++ b/packages/pipeline/test/entities/exchange_cancel_event_test.ts
@@ -0,0 +1,57 @@
+import 'mocha';
+import * as R from 'ramda';
+import 'reflect-metadata';
+
+import { ExchangeCancelEvent } from '../../src/entities';
+import { AssetType } from '../../src/types';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const baseCancelEvent = {
+    contractAddress: '0x4f833a24e1f95d70f028921e27040ca56e09ab0b',
+    logIndex: 1234,
+    blockNumber: 6276262,
+    rawData: '0x000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428',
+    transactionHash: '0x6dd106d002873746072fc5e496dd0fb2541b68c77bcf9184ae19a42fd33657fe',
+    makerAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+    takerAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+    feeRecipientAddress: '0xc370d2a5920344aa6b7d8d11250e3e861434cbdd',
+    senderAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+    orderHash: '0xab12ed2cbaa5615ab690b9da75a46e53ddfcf3f1a68655b5fe0d94c75a1aac4a',
+    rawMakerAssetData: '0xf47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+    makerAssetProxyId: '0xf47261b0',
+    makerTokenAddress: '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+    rawTakerAssetData: '0xf47261b0000000000000000000000000e41d2489571d322189246dafa5ebde1f4699f498',
+    takerAssetProxyId: '0xf47261b0',
+    takerTokenAddress: '0xe41d2489571d322189246dafa5ebde1f4699f498',
+};
+
+const erc20CancelEvent = R.merge(baseCancelEvent, {
+    makerAssetType: 'erc20' as AssetType,
+    makerTokenId: null,
+    takerAssetType: 'erc20' as AssetType,
+    takerTokenId: null,
+});
+
+const erc721CancelEvent = R.merge(baseCancelEvent, {
+    makerAssetType: 'erc721' as AssetType,
+    makerTokenId: '19378573',
+    takerAssetType: 'erc721' as AssetType,
+    takerTokenId: '63885673888',
+});
+
+// tslint:disable:custom-no-magic-numbers
+describe('ExchangeCancelEvent entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const events = [erc20CancelEvent, erc721CancelEvent];
+        const cancelEventRepository = connection.getRepository(ExchangeCancelEvent);
+        for (const event of events) {
+            await testSaveAndFindEntityAsync(cancelEventRepository, event);
+        }
+    });
+});
diff --git a/packages/pipeline/test/entities/exchange_cancel_up_to_event_test.ts b/packages/pipeline/test/entities/exchange_cancel_up_to_event_test.ts
new file mode 100644
index 000000000..aa34f8c1c
--- /dev/null
+++ b/packages/pipeline/test/entities/exchange_cancel_up_to_event_test.ts
@@ -0,0 +1,29 @@
+import { BigNumber } from '@0x/utils';
+import 'mocha';
+import 'reflect-metadata';
+
+import { ExchangeCancelUpToEvent } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+// tslint:disable:custom-no-magic-numbers
+describe('ExchangeCancelUpToEvent entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const cancelUpToEventRepository = connection.getRepository(ExchangeCancelUpToEvent);
+        const cancelUpToEvent = new ExchangeCancelUpToEvent();
+        cancelUpToEvent.blockNumber = 6276262;
+        cancelUpToEvent.contractAddress = '0x4f833a24e1f95d70f028921e27040ca56e09ab0b';
+        cancelUpToEvent.logIndex = 42;
+        cancelUpToEvent.makerAddress = '0xf6da68519f78b0d0bc93c701e86affcb75c92428';
+        cancelUpToEvent.orderEpoch = new BigNumber('123456789123456789');
+        cancelUpToEvent.rawData = '0x000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428';
+        cancelUpToEvent.senderAddress = '0xf6da68519f78b0d0bc93c701e86affcb75c92428';
+        cancelUpToEvent.transactionHash = '0x6dd106d002873746072fc5e496dd0fb2541b68c77bcf9184ae19a42fd33657fe';
+        await testSaveAndFindEntityAsync(cancelUpToEventRepository, cancelUpToEvent);
+    });
+});
diff --git a/packages/pipeline/test/entities/exchange_fill_event_test.ts b/packages/pipeline/test/entities/exchange_fill_event_test.ts
new file mode 100644
index 000000000..b2cb8c5e0
--- /dev/null
+++ b/packages/pipeline/test/entities/exchange_fill_event_test.ts
@@ -0,0 +1,62 @@
+import { BigNumber } from '@0x/utils';
+import 'mocha';
+import * as R from 'ramda';
+import 'reflect-metadata';
+
+import { ExchangeFillEvent } from '../../src/entities';
+import { AssetType } from '../../src/types';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const baseFillEvent = {
+    contractAddress: '0x4f833a24e1f95d70f028921e27040ca56e09ab0b',
+    blockNumber: 6276262,
+    logIndex: 102,
+    rawData: '0x000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428',
+    transactionHash: '0x6dd106d002873746072fc5e496dd0fb2541b68c77bcf9184ae19a42fd33657fe',
+    makerAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+    takerAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+    feeRecipientAddress: '0xc370d2a5920344aa6b7d8d11250e3e861434cbdd',
+    senderAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+    makerAssetFilledAmount: new BigNumber('10000000000000000'),
+    takerAssetFilledAmount: new BigNumber('100000000000000000'),
+    makerFeePaid: new BigNumber('0'),
+    takerFeePaid: new BigNumber('12345'),
+    orderHash: '0xab12ed2cbaa5615ab690b9da75a46e53ddfcf3f1a68655b5fe0d94c75a1aac4a',
+    rawMakerAssetData: '0xf47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+    makerAssetProxyId: '0xf47261b0',
+    makerTokenAddress: '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+    rawTakerAssetData: '0xf47261b0000000000000000000000000e41d2489571d322189246dafa5ebde1f4699f498',
+    takerAssetProxyId: '0xf47261b0',
+    takerTokenAddress: '0xe41d2489571d322189246dafa5ebde1f4699f498',
+};
+
+const erc20FillEvent = R.merge(baseFillEvent, {
+    makerAssetType: 'erc20' as AssetType,
+    makerTokenId: null,
+    takerAssetType: 'erc20' as AssetType,
+    takerTokenId: null,
+});
+
+const erc721FillEvent = R.merge(baseFillEvent, {
+    makerAssetType: 'erc721' as AssetType,
+    makerTokenId: '19378573',
+    takerAssetType: 'erc721' as AssetType,
+    takerTokenId: '63885673888',
+});
+
+// tslint:disable:custom-no-magic-numbers
+describe('ExchangeFillEvent entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const events = [erc20FillEvent, erc721FillEvent];
+        const fillEventsRepository = connection.getRepository(ExchangeFillEvent);
+        for (const event of events) {
+            await testSaveAndFindEntityAsync(fillEventsRepository, event);
+        }
+    });
+});
diff --git a/packages/pipeline/test/entities/ohlcv_external_test.ts b/packages/pipeline/test/entities/ohlcv_external_test.ts
new file mode 100644
index 000000000..8b995db50
--- /dev/null
+++ b/packages/pipeline/test/entities/ohlcv_external_test.ts
@@ -0,0 +1,35 @@
+import 'mocha';
+import 'reflect-metadata';
+
+import { OHLCVExternal } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const ohlcvExternal: OHLCVExternal = {
+    exchange: 'CCCAGG',
+    fromSymbol: 'ETH',
+    toSymbol: 'ZRX',
+    startTime: 1543352400000,
+    endTime: 1543356000000,
+    open: 307.41,
+    close: 310.08,
+    low: 304.6,
+    high: 310.27,
+    volumeFrom: 904.6,
+    volumeTo: 278238.5,
+    source: 'Crypto Compare',
+    observedTimestamp: 1543442338074,
+};
+
+// tslint:disable:custom-no-magic-numbers
+describe('OHLCVExternal entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const repository = connection.getRepository(OHLCVExternal);
+        await testSaveAndFindEntityAsync(repository, ohlcvExternal);
+    });
+});
diff --git a/packages/pipeline/test/entities/relayer_test.ts b/packages/pipeline/test/entities/relayer_test.ts
new file mode 100644
index 000000000..760ffb6f9
--- /dev/null
+++ b/packages/pipeline/test/entities/relayer_test.ts
@@ -0,0 +1,55 @@
+import 'mocha';
+import * as R from 'ramda';
+import 'reflect-metadata';
+
+import { Relayer } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const baseRelayer = {
+    uuid: 'e8d27d8d-ddf6-48b1-9663-60b0a3ddc716',
+    name: 'Radar Relay',
+    homepageUrl: 'https://radarrelay.com',
+    appUrl: null,
+    sraHttpEndpoint: null,
+    sraWsEndpoint: null,
+    feeRecipientAddresses: [],
+    takerAddresses: [],
+};
+
+const relayerWithUrls = R.merge(baseRelayer, {
+    uuid: 'e8d27d8d-ddf6-48b1-9663-60b0a3ddc717',
+    appUrl: 'https://app.radarrelay.com',
+    sraHttpEndpoint: 'https://api.radarrelay.com/0x/v2/',
+    sraWsEndpoint: 'wss://ws.radarrelay.com/0x/v2',
+});
+
+const relayerWithAddresses = R.merge(baseRelayer, {
+    uuid: 'e8d27d8d-ddf6-48b1-9663-60b0a3ddc718',
+    feeRecipientAddresses: [
+        '0xa258b39954cef5cb142fd567a46cddb31a670124',
+        '0xa258b39954cef5cb142fd567a46cddb31a670125',
+        '0xa258b39954cef5cb142fd567a46cddb31a670126',
+    ],
+    takerAddresses: [
+        '0xa258b39954cef5cb142fd567a46cddb31a670127',
+        '0xa258b39954cef5cb142fd567a46cddb31a670128',
+        '0xa258b39954cef5cb142fd567a46cddb31a670129',
+    ],
+});
+
+// tslint:disable:custom-no-magic-numbers
+describe('Relayer entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const relayers = [baseRelayer, relayerWithUrls, relayerWithAddresses];
+        const relayerRepository = connection.getRepository(Relayer);
+        for (const relayer of relayers) {
+            await testSaveAndFindEntityAsync(relayerRepository, relayer);
+        }
+    });
+});
diff --git a/packages/pipeline/test/entities/sra_order_test.ts b/packages/pipeline/test/entities/sra_order_test.ts
new file mode 100644
index 000000000..c43de8ce8
--- /dev/null
+++ b/packages/pipeline/test/entities/sra_order_test.ts
@@ -0,0 +1,84 @@
+import { BigNumber } from '@0x/utils';
+import 'mocha';
+import * as R from 'ramda';
+import 'reflect-metadata';
+import { Repository } from 'typeorm';
+
+import { SraOrder, SraOrdersObservedTimeStamp } from '../../src/entities';
+import { AssetType } from '../../src/types';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const baseOrder = {
+    sourceUrl: 'https://api.radarrelay.com/0x/v2',
+    exchangeAddress: '0x4f833a24e1f95d70f028921e27040ca56e09ab0b',
+    makerAddress: '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81',
+    takerAddress: '0x0000000000000000000000000000000000000000',
+    feeRecipientAddress: '0xa258b39954cef5cb142fd567a46cddb31a670124',
+    senderAddress: '0x0000000000000000000000000000000000000000',
+    makerAssetAmount: new BigNumber('1619310371000000000'),
+    takerAssetAmount: new BigNumber('8178335207070707070707'),
+    makerFee: new BigNumber('100'),
+    takerFee: new BigNumber('200'),
+    expirationTimeSeconds: new BigNumber('1538529488'),
+    salt: new BigNumber('1537924688891'),
+    signature: '0x1b5a5d672b0d647b5797387ccbb89d8',
+    rawMakerAssetData: '0xf47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+    makerAssetProxyId: '0xf47261b0',
+    makerTokenAddress: '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+    rawTakerAssetData: '0xf47261b000000000000000000000000042d6622dece394b54999fbd73d108123806f6a18',
+    takerAssetProxyId: '0xf47261b0',
+    takerTokenAddress: '0x42d6622dece394b54999fbd73d108123806f6a18',
+    metadataJson: '{"isThisArbitraryData":true,"powerLevel":9001}',
+};
+
+const erc20Order = R.merge(baseOrder, {
+    orderHashHex: '0x1bdbeb0d088a33da28b9ee6d94e8771452f90f4a69107da2fa75195d61b9a1c9',
+    makerAssetType: 'erc20' as AssetType,
+    makerTokenId: null,
+    takerAssetType: 'erc20' as AssetType,
+    takerTokenId: null,
+});
+
+const erc721Order = R.merge(baseOrder, {
+    orderHashHex: '0x1bdbeb0d088a33da28b9ee6d94e8771452f90f4a69107da2fa75195d61b9a1d0',
+    makerAssetType: 'erc721' as AssetType,
+    makerTokenId: '19378573',
+    takerAssetType: 'erc721' as AssetType,
+    takerTokenId: '63885673888',
+});
+
+// tslint:disable:custom-no-magic-numbers
+describe('SraOrder and SraOrdersObservedTimeStamp entities', () => {
+    // Note(albrow): SraOrder and SraOrdersObservedTimeStamp are tightly coupled
+    // and timestamps have a foreign key constraint such that they have to point
+    // to an existing SraOrder. For these reasons, we are testing them together
+    // in the same test.
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const orderRepository = connection.getRepository(SraOrder);
+        const timestampRepository = connection.getRepository(SraOrdersObservedTimeStamp);
+        const orders = [erc20Order, erc721Order];
+        for (const order of orders) {
+            await testOrderWithTimestampAsync(orderRepository, timestampRepository, order);
+        }
+    });
+});
+
+async function testOrderWithTimestampAsync(
+    orderRepository: Repository<SraOrder>,
+    timestampRepository: Repository<SraOrdersObservedTimeStamp>,
+    order: SraOrder,
+): Promise<void> {
+    await testSaveAndFindEntityAsync(orderRepository, order);
+    const timestamp = new SraOrdersObservedTimeStamp();
+    timestamp.exchangeAddress = order.exchangeAddress;
+    timestamp.orderHashHex = order.orderHashHex;
+    timestamp.sourceUrl = order.sourceUrl;
+    timestamp.observedTimestamp = 1543377376153;
+    await testSaveAndFindEntityAsync(timestampRepository, timestamp);
+}
diff --git a/packages/pipeline/test/entities/token_metadata_test.ts b/packages/pipeline/test/entities/token_metadata_test.ts
new file mode 100644
index 000000000..48e656644
--- /dev/null
+++ b/packages/pipeline/test/entities/token_metadata_test.ts
@@ -0,0 +1,39 @@
+import { BigNumber } from '@0x/utils';
+import 'mocha';
+import 'reflect-metadata';
+
+import { TokenMetadata } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const metadataWithoutNullFields: TokenMetadata = {
+    address: '0xe41d2489571d322189246dafa5ebde1f4699f498',
+    authority: 'https://website-api.0xproject.com/tokens',
+    decimals: new BigNumber(18),
+    symbol: 'ZRX',
+    name: '0x',
+};
+
+const metadataWithNullFields: TokenMetadata = {
+    address: '0xe41d2489571d322189246dafa5ebde1f4699f499',
+    authority: 'https://website-api.0xproject.com/tokens',
+    decimals: null,
+    symbol: null,
+    name: null,
+};
+
+// tslint:disable:custom-no-magic-numbers
+describe('TokenMetadata entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const tokenMetadata = [metadataWithoutNullFields, metadataWithNullFields];
+        const tokenMetadataRepository = connection.getRepository(TokenMetadata);
+        for (const tokenMetadatum of tokenMetadata) {
+            await testSaveAndFindEntityAsync(tokenMetadataRepository, tokenMetadatum);
+        }
+    });
+});
diff --git a/packages/pipeline/test/entities/token_order_test.ts b/packages/pipeline/test/entities/token_order_test.ts
new file mode 100644
index 000000000..c6057f5aa
--- /dev/null
+++ b/packages/pipeline/test/entities/token_order_test.ts
@@ -0,0 +1,31 @@
+import { BigNumber } from '@0x/utils';
+import 'mocha';
+
+import { TokenOrderbookSnapshot } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+const tokenOrderbookSnapshot: TokenOrderbookSnapshot = {
+    source: 'ddextest',
+    observedTimestamp: Date.now(),
+    orderType: 'bid',
+    price: new BigNumber(10.1),
+    baseAssetSymbol: 'ETH',
+    baseAssetAddress: '0x818e6fecd516ecc3849daf6845e3ec868087b755',
+    baseVolume: new BigNumber(143),
+    quoteAssetSymbol: 'ABC',
+    quoteAssetAddress: '0x00923b9a074762b93650716333b3e1473a15048e',
+    quoteVolume: new BigNumber(12.3234234),
+};
+
+describe('TokenOrderbookSnapshot entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const tokenOrderbookSnapshotRepository = connection.getRepository(TokenOrderbookSnapshot);
+        await testSaveAndFindEntityAsync(tokenOrderbookSnapshotRepository, tokenOrderbookSnapshot);
+    });
+});
diff --git a/packages/pipeline/test/entities/transaction_test.ts b/packages/pipeline/test/entities/transaction_test.ts
new file mode 100644
index 000000000..634844544
--- /dev/null
+++ b/packages/pipeline/test/entities/transaction_test.ts
@@ -0,0 +1,26 @@
+import { BigNumber } from '@0x/utils';
+import 'mocha';
+import 'reflect-metadata';
+
+import { Transaction } from '../../src/entities';
+import { createDbConnectionOnceAsync } from '../db_setup';
+import { chaiSetup } from '../utils/chai_setup';
+
+import { testSaveAndFindEntityAsync } from './util';
+
+chaiSetup.configure();
+
+// tslint:disable:custom-no-magic-numbers
+describe('Transaction entity', () => {
+    it('save/find', async () => {
+        const connection = await createDbConnectionOnceAsync();
+        const transactionRepository = connection.getRepository(Transaction);
+        const transaction = new Transaction();
+        transaction.blockHash = '0x6ff106d00b6c3746072fc06bae140fb2549036ba7bcf9184ae19a42fd33657fd';
+        transaction.blockNumber = 6276262;
+        transaction.gasPrice = new BigNumber(3000000);
+        transaction.gasUsed = new BigNumber(125000);
+        transaction.transactionHash = '0x6dd106d002873746072fc5e496dd0fb2541b68c77bcf9184ae19a42fd33657fe';
+        await testSaveAndFindEntityAsync(transactionRepository, transaction);
+    });
+});
diff --git a/packages/pipeline/test/entities/util.ts b/packages/pipeline/test/entities/util.ts
new file mode 100644
index 000000000..043a3b15d
--- /dev/null
+++ b/packages/pipeline/test/entities/util.ts
@@ -0,0 +1,25 @@
+import * as chai from 'chai';
+import 'mocha';
+
+import { Repository } from 'typeorm';
+
+const expect = chai.expect;
+
+/**
+ * First saves the given entity to the database, then finds it and makes sure
+ * that the found entity is exactly equal to the original one. This is a bare
+ * minimum basic test to make sure that the entity type definition and our
+ * database schema are aligned and that it is possible to save and find the
+ * entity.
+ * @param repository A TypeORM repository corresponding with the type of the entity.
+ * @param entity An instance of a TypeORM entity which will be saved/retrieved from the database.
+ */
+export async function testSaveAndFindEntityAsync<T>(repository: Repository<T>, entity: T): Promise<void> {
+    // Note(albrow): We are forced to use an 'as any' hack here because
+    // TypeScript complains about stack depth when checking the types.
+    await repository.save(entity as any);
+    const gotEntity = await repository.findOneOrFail({
+        where: entity,
+    });
+    expect(gotEntity).deep.equal(entity);
+}
diff --git a/packages/pipeline/test/parsers/bloxy/index_test.ts b/packages/pipeline/test/parsers/bloxy/index_test.ts
new file mode 100644
index 000000000..2b8d68f98
--- /dev/null
+++ b/packages/pipeline/test/parsers/bloxy/index_test.ts
@@ -0,0 +1,99 @@
+// tslint:disable:custom-no-magic-numbers
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import 'mocha';
+import * as R from 'ramda';
+
+import { BLOXY_DEX_TRADES_URL, BloxyTrade } from '../../../src/data_sources/bloxy';
+import { DexTrade } from '../../../src/entities';
+import { _parseBloxyTrade } from '../../../src/parsers/bloxy';
+import { _convertToExchangeFillEvent } from '../../../src/parsers/events';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+const baseInput: BloxyTrade = {
+    tx_hash: '0xb93a7faf92efbbb5405c9a73cd4efd99702fe27c03ff22baee1f1b1e37b3a0bf',
+    tx_time: '2018-11-21T09:06:28.000+00:00',
+    tx_date: '2018-11-21',
+    tx_sender: '0x00923b9a074762b93650716333b3e1473a15048e',
+    smart_contract_id: 7091917,
+    smart_contract_address: '0x818e6fecd516ecc3849daf6845e3ec868087b755',
+    contract_type: 'DEX/Kyber Network Proxy',
+    maker: '0x0000000000000000000000000000000000000001',
+    taker: '0x0000000000000000000000000000000000000002',
+    amountBuy: 1.011943163078103,
+    makerFee: 38.912083,
+    buyCurrencyId: 1,
+    buySymbol: 'ETH',
+    amountSell: 941.4997928436911,
+    takerFee: 100.39,
+    sellCurrencyId: 16610,
+    sellSymbol: 'ELF',
+    maker_annotation: 'random annotation',
+    taker_annotation: 'random other annotation',
+    protocol: 'Kyber Network Proxy',
+    buyAddress: '0xbf2179859fc6d5bee9bf9158632dc51678a4100d',
+    sellAddress: '0xbf2179859fc6d5bee9bf9158632dc51678a4100e',
+};
+
+const baseExpected: DexTrade = {
+    sourceUrl: BLOXY_DEX_TRADES_URL,
+    txHash: '0xb93a7faf92efbbb5405c9a73cd4efd99702fe27c03ff22baee1f1b1e37b3a0bf',
+    txTimestamp: 1542791188000,
+    txDate: '2018-11-21',
+    txSender: '0x00923b9a074762b93650716333b3e1473a15048e',
+    smartContractId: 7091917,
+    smartContractAddress: '0x818e6fecd516ecc3849daf6845e3ec868087b755',
+    contractType: 'DEX/Kyber Network Proxy',
+    maker: '0x0000000000000000000000000000000000000001',
+    taker: '0x0000000000000000000000000000000000000002',
+    amountBuy: new BigNumber('1.011943163078103'),
+    makerFeeAmount: new BigNumber('38.912083'),
+    buyCurrencyId: 1,
+    buySymbol: 'ETH',
+    amountSell: new BigNumber('941.4997928436911'),
+    takerFeeAmount: new BigNumber('100.39'),
+    sellCurrencyId: 16610,
+    sellSymbol: 'ELF',
+    makerAnnotation: 'random annotation',
+    takerAnnotation: 'random other annotation',
+    protocol: 'Kyber Network Proxy',
+    buyAddress: '0xbf2179859fc6d5bee9bf9158632dc51678a4100d',
+    sellAddress: '0xbf2179859fc6d5bee9bf9158632dc51678a4100e',
+};
+
+interface TestCase {
+    input: BloxyTrade;
+    expected: DexTrade;
+}
+
+const testCases: TestCase[] = [
+    {
+        input: baseInput,
+        expected: baseExpected,
+    },
+    {
+        input: R.merge(baseInput, { buyAddress: null, sellAddress: null }),
+        expected: R.merge(baseExpected, { buyAddress: null, sellAddress: null }),
+    },
+    {
+        input: R.merge(baseInput, {
+            buySymbol:
+                'RING\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000',
+        }),
+        expected: R.merge(baseExpected, { buySymbol: 'RING' }),
+    },
+];
+
+describe('bloxy', () => {
+    describe('_parseBloxyTrade', () => {
+        for (const [i, testCase] of testCases.entries()) {
+            it(`converts BloxyTrade to DexTrade entity (${i + 1}/${testCases.length})`, () => {
+                const actual = _parseBloxyTrade(testCase.input);
+                expect(actual).deep.equal(testCase.expected);
+            });
+        }
+    });
+});
diff --git a/packages/pipeline/test/parsers/ddex_orders/index_test.ts b/packages/pipeline/test/parsers/ddex_orders/index_test.ts
new file mode 100644
index 000000000..213100f44
--- /dev/null
+++ b/packages/pipeline/test/parsers/ddex_orders/index_test.ts
@@ -0,0 +1,66 @@
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import 'mocha';
+
+import { DdexMarket } from '../../../src/data_sources/ddex';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../../src/entities';
+import { aggregateOrders, parseDdexOrder } from '../../../src/parsers/ddex_orders';
+import { OrderType } from '../../../src/types';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('ddex_orders', () => {
+    describe('aggregateOrders', () => {
+        it('aggregates orders by price point', () => {
+            const input = [
+                { price: '1', amount: '20', orderId: 'testtest' },
+                { price: '1', amount: '30', orderId: 'testone' },
+                { price: '2', amount: '100', orderId: 'testtwo' },
+            ];
+            const expected = [['1', new BigNumber(50)], ['2', new BigNumber(100)]];
+            const actual = aggregateOrders(input);
+            expect(actual).deep.equal(expected);
+        });
+    });
+
+    describe('parseDdexOrder', () => {
+        it('converts ddexOrder to TokenOrder entity', () => {
+            const ddexOrder: [string, BigNumber] = ['0.5', new BigNumber(10)];
+            const ddexMarket: DdexMarket = {
+                id: 'ABC-DEF',
+                quoteToken: 'ABC',
+                quoteTokenDecimals: 5,
+                quoteTokenAddress: '0x0000000000000000000000000000000000000000',
+                baseToken: 'DEF',
+                baseTokenDecimals: 2,
+                baseTokenAddress: '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81',
+                minOrderSize: '0.1',
+                maxOrderSize: '1000',
+                pricePrecision: 1,
+                priceDecimals: 1,
+                amountDecimals: 0,
+            };
+            const observedTimestamp: number = Date.now();
+            const orderType: OrderType = 'bid';
+            const source: string = 'ddex';
+
+            const expected = new TokenOrder();
+            expected.source = 'ddex';
+            expected.observedTimestamp = observedTimestamp;
+            expected.orderType = 'bid';
+            expected.price = new BigNumber(0.5);
+            expected.baseAssetSymbol = 'DEF';
+            expected.baseAssetAddress = '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81';
+            expected.baseVolume = new BigNumber(5);
+            expected.quoteAssetSymbol = 'ABC';
+            expected.quoteAssetAddress = '0x0000000000000000000000000000000000000000';
+            expected.quoteVolume = new BigNumber(10);
+
+            const actual = parseDdexOrder(ddexMarket, observedTimestamp, orderType, source, ddexOrder);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});
diff --git a/packages/pipeline/test/parsers/events/index_test.ts b/packages/pipeline/test/parsers/events/index_test.ts
new file mode 100644
index 000000000..7e439ce39
--- /dev/null
+++ b/packages/pipeline/test/parsers/events/index_test.ts
@@ -0,0 +1,78 @@
+import { ExchangeFillEventArgs } from '@0x/contract-wrappers';
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import { LogWithDecodedArgs } from 'ethereum-types';
+import 'mocha';
+
+import { ExchangeFillEvent } from '../../../src/entities';
+import { _convertToExchangeFillEvent } from '../../../src/parsers/events';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('exchange_events', () => {
+    describe('_convertToExchangeFillEvent', () => {
+        it('converts LogWithDecodedArgs to ExchangeFillEvent entity', () => {
+            const input: LogWithDecodedArgs<ExchangeFillEventArgs> = {
+                logIndex: 102,
+                transactionIndex: 38,
+                transactionHash: '0x6dd106d002873746072fc5e496dd0fb2541b68c77bcf9184ae19a42fd33657fe',
+                blockHash: '',
+                blockNumber: 6276262,
+                address: '0x4f833a24e1f95d70f028921e27040ca56e09ab0b',
+                data:
+                    '0x000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428000000000000000000000000000000000000000000000000002386f26fc10000000000000000000000000000000000000000000000000000016345785d8a000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000000000001600000000000000000000000000000000000000000000000000000000000000024f47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000024f47261b0000000000000000000000000e41d2489571d322189246dafa5ebde1f4699f49800000000000000000000000000000000000000000000000000000000',
+                topics: [
+                    '0x0bcc4c97732e47d9946f229edb95f5b6323f601300e4690de719993f3c371129',
+                    '0x000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428',
+                    '0x000000000000000000000000c370d2a5920344aa6b7d8d11250e3e861434cbdd',
+                    '0xab12ed2cbaa5615ab690b9da75a46e53ddfcf3f1a68655b5fe0d94c75a1aac4a',
+                ],
+                event: 'Fill',
+                args: {
+                    makerAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+                    feeRecipientAddress: '0xc370d2a5920344aa6b7d8d11250e3e861434cbdd',
+                    takerAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+                    senderAddress: '0xf6da68519f78b0d0bc93c701e86affcb75c92428',
+                    makerAssetFilledAmount: new BigNumber('10000000000000000'),
+                    takerAssetFilledAmount: new BigNumber('100000000000000000'),
+                    makerFeePaid: new BigNumber('0'),
+                    takerFeePaid: new BigNumber('12345'),
+                    orderHash: '0xab12ed2cbaa5615ab690b9da75a46e53ddfcf3f1a68655b5fe0d94c75a1aac4a',
+                    makerAssetData: '0xf47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+                    takerAssetData: '0xf47261b0000000000000000000000000e41d2489571d322189246dafa5ebde1f4699f498',
+                },
+            };
+            const expected = new ExchangeFillEvent();
+            expected.contractAddress = '0x4f833a24e1f95d70f028921e27040ca56e09ab0b';
+            expected.blockNumber = 6276262;
+            expected.logIndex = 102;
+            expected.rawData =
+                '0x000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428000000000000000000000000000000000000000000000000002386f26fc10000000000000000000000000000000000000000000000000000016345785d8a000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000000000001600000000000000000000000000000000000000000000000000000000000000024f47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000024f47261b0000000000000000000000000e41d2489571d322189246dafa5ebde1f4699f49800000000000000000000000000000000000000000000000000000000';
+            expected.transactionHash = '0x6dd106d002873746072fc5e496dd0fb2541b68c77bcf9184ae19a42fd33657fe';
+            expected.makerAddress = '0xf6da68519f78b0d0bc93c701e86affcb75c92428';
+            expected.takerAddress = '0xf6da68519f78b0d0bc93c701e86affcb75c92428';
+            expected.feeRecipientAddress = '0xc370d2a5920344aa6b7d8d11250e3e861434cbdd';
+            expected.senderAddress = '0xf6da68519f78b0d0bc93c701e86affcb75c92428';
+            expected.makerAssetFilledAmount = new BigNumber('10000000000000000');
+            expected.takerAssetFilledAmount = new BigNumber('100000000000000000');
+            expected.makerFeePaid = new BigNumber('0');
+            expected.takerFeePaid = new BigNumber('12345');
+            expected.orderHash = '0xab12ed2cbaa5615ab690b9da75a46e53ddfcf3f1a68655b5fe0d94c75a1aac4a';
+            expected.rawMakerAssetData = '0xf47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2';
+            expected.makerAssetType = 'erc20';
+            expected.makerAssetProxyId = '0xf47261b0';
+            expected.makerTokenAddress = '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2';
+            expected.makerTokenId = null;
+            expected.rawTakerAssetData = '0xf47261b0000000000000000000000000e41d2489571d322189246dafa5ebde1f4699f498';
+            expected.takerAssetType = 'erc20';
+            expected.takerAssetProxyId = '0xf47261b0';
+            expected.takerTokenAddress = '0xe41d2489571d322189246dafa5ebde1f4699f498';
+            expected.takerTokenId = null;
+            const actual = _convertToExchangeFillEvent(input);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});
diff --git a/packages/pipeline/test/parsers/ohlcv_external/crypto_compare_test.ts b/packages/pipeline/test/parsers/ohlcv_external/crypto_compare_test.ts
new file mode 100644
index 000000000..118cafc5e
--- /dev/null
+++ b/packages/pipeline/test/parsers/ohlcv_external/crypto_compare_test.ts
@@ -0,0 +1,62 @@
+import * as chai from 'chai';
+import 'mocha';
+import * as R from 'ramda';
+
+import { CryptoCompareOHLCVRecord } from '../../../src/data_sources/ohlcv_external/crypto_compare';
+import { OHLCVExternal } from '../../../src/entities';
+import { OHLCVMetadata, parseRecords } from '../../../src/parsers/ohlcv_external/crypto_compare';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('ohlcv_external parser (Crypto Compare)', () => {
+    describe('parseRecords', () => {
+        const record: CryptoCompareOHLCVRecord = {
+            time: 200,
+            close: 100,
+            high: 101,
+            low: 99,
+            open: 98,
+            volumefrom: 1234,
+            volumeto: 4321,
+        };
+
+        const metadata: OHLCVMetadata = {
+            fromSymbol: 'ETH',
+            toSymbol: 'ZRX',
+            exchange: 'CCCAGG',
+            source: 'CryptoCompare',
+            observedTimestamp: new Date().getTime(),
+            interval: 100000,
+        };
+
+        const entity = new OHLCVExternal();
+        entity.exchange = metadata.exchange;
+        entity.fromSymbol = metadata.fromSymbol;
+        entity.toSymbol = metadata.toSymbol;
+        entity.startTime = 100000;
+        entity.endTime = 200000;
+        entity.open = record.open;
+        entity.close = record.close;
+        entity.low = record.low;
+        entity.high = record.high;
+        entity.volumeFrom = record.volumefrom;
+        entity.volumeTo = record.volumeto;
+        entity.source = metadata.source;
+        entity.observedTimestamp = metadata.observedTimestamp;
+
+        it('converts Crypto Compare OHLCV records to OHLCVExternal entity', () => {
+            const input = [record, R.merge(record, { time: 300 }), R.merge(record, { time: 400 })];
+            const expected = [
+                entity,
+                R.merge(entity, { startTime: 200000, endTime: 300000 }),
+                R.merge(entity, { startTime: 300000, endTime: 400000 }),
+            ];
+
+            const actual = parseRecords(input, metadata);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});
diff --git a/packages/pipeline/test/parsers/paradex_orders/index_test.ts b/packages/pipeline/test/parsers/paradex_orders/index_test.ts
new file mode 100644
index 000000000..1522806bf
--- /dev/null
+++ b/packages/pipeline/test/parsers/paradex_orders/index_test.ts
@@ -0,0 +1,54 @@
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import 'mocha';
+
+import { ParadexMarket, ParadexOrder } from '../../../src/data_sources/paradex';
+import { TokenOrderbookSnapshot as TokenOrder } from '../../../src/entities';
+import { parseParadexOrder } from '../../../src/parsers/paradex_orders';
+import { OrderType } from '../../../src/types';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('paradex_orders', () => {
+    describe('parseParadexOrder', () => {
+        it('converts ParadexOrder to TokenOrder entity', () => {
+            const paradexOrder: ParadexOrder = {
+                amount: '412',
+                price: '0.1245',
+            };
+            const paradexMarket: ParadexMarket = {
+                id: '2',
+                symbol: 'ABC/DEF',
+                baseToken: 'DEF',
+                quoteToken: 'ABC',
+                minOrderSize: '0.1',
+                maxOrderSize: '1000',
+                priceMaxDecimals: 5,
+                amountMaxDecimals: 5,
+                baseTokenAddress: '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81',
+                quoteTokenAddress: '0x0000000000000000000000000000000000000000',
+            };
+            const observedTimestamp: number = Date.now();
+            const orderType: OrderType = 'bid';
+            const source: string = 'paradex';
+
+            const expected = new TokenOrder();
+            expected.source = 'paradex';
+            expected.observedTimestamp = observedTimestamp;
+            expected.orderType = 'bid';
+            expected.price = new BigNumber(0.1245);
+            expected.baseAssetSymbol = 'DEF';
+            expected.baseAssetAddress = '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81';
+            expected.baseVolume = new BigNumber(412 * 0.1245);
+            expected.quoteAssetSymbol = 'ABC';
+            expected.quoteAssetAddress = '0x0000000000000000000000000000000000000000';
+            expected.quoteVolume = new BigNumber(412);
+
+            const actual = parseParadexOrder(paradexMarket, observedTimestamp, orderType, source, paradexOrder);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});
diff --git a/packages/pipeline/test/parsers/sra_orders/index_test.ts b/packages/pipeline/test/parsers/sra_orders/index_test.ts
new file mode 100644
index 000000000..ee2842ef3
--- /dev/null
+++ b/packages/pipeline/test/parsers/sra_orders/index_test.ts
@@ -0,0 +1,68 @@
+import { APIOrder } from '@0x/types';
+import { BigNumber } from '@0x/utils';
+import * as chai from 'chai';
+import 'mocha';
+
+import { SraOrder } from '../../../src/entities';
+import { _convertToEntity } from '../../../src/parsers/sra_orders';
+import { chaiSetup } from '../../utils/chai_setup';
+
+chaiSetup.configure();
+const expect = chai.expect;
+
+// tslint:disable:custom-no-magic-numbers
+describe('sra_orders', () => {
+    describe('_convertToEntity', () => {
+        it('converts ApiOrder to SraOrder entity', () => {
+            const input: APIOrder = {
+                order: {
+                    makerAddress: '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81',
+                    takerAddress: '0x0000000000000000000000000000000000000000',
+                    feeRecipientAddress: '0xa258b39954cef5cb142fd567a46cddb31a670124',
+                    senderAddress: '0x0000000000000000000000000000000000000000',
+                    makerAssetAmount: new BigNumber('1619310371000000000'),
+                    takerAssetAmount: new BigNumber('8178335207070707070707'),
+                    makerFee: new BigNumber('0'),
+                    takerFee: new BigNumber('0'),
+                    exchangeAddress: '0x4f833a24e1f95d70f028921e27040ca56e09ab0b',
+                    expirationTimeSeconds: new BigNumber('1538529488'),
+                    signature:
+                        '0x1b5a5d672b0d647b5797387ccbb89d822d5d2e873346b014f4ff816ff0783f2a7a0d2824d2d7042ec8ea375bc7f870963e1cb8248f1db03ddf125e27b5963aa11f03',
+                    salt: new BigNumber('1537924688891'),
+                    makerAssetData: '0xf47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+                    takerAssetData: '0xf47261b000000000000000000000000042d6622dece394b54999fbd73d108123806f6a18',
+                },
+                metaData: { isThisArbitraryData: true, powerLevel: 9001 },
+            };
+            const expected = new SraOrder();
+            expected.exchangeAddress = '0x4f833a24e1f95d70f028921e27040ca56e09ab0b';
+            expected.orderHashHex = '0x1bdbeb0d088a33da28b9ee6d94e8771452f90f4a69107da2fa75195d61b9a1c9';
+            expected.makerAddress = '0xb45df06e38540a675fdb5b598abf2c0dbe9d6b81';
+            expected.takerAddress = '0x0000000000000000000000000000000000000000';
+            expected.feeRecipientAddress = '0xa258b39954cef5cb142fd567a46cddb31a670124';
+            expected.senderAddress = '0x0000000000000000000000000000000000000000';
+            expected.makerAssetAmount = new BigNumber('1619310371000000000');
+            expected.takerAssetAmount = new BigNumber('8178335207070707070707');
+            expected.makerFee = new BigNumber('0');
+            expected.takerFee = new BigNumber('0');
+            expected.expirationTimeSeconds = new BigNumber('1538529488');
+            expected.salt = new BigNumber('1537924688891');
+            expected.signature =
+                '0x1b5a5d672b0d647b5797387ccbb89d822d5d2e873346b014f4ff816ff0783f2a7a0d2824d2d7042ec8ea375bc7f870963e1cb8248f1db03ddf125e27b5963aa11f03';
+            expected.rawMakerAssetData = '0xf47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2';
+            expected.makerAssetType = 'erc20';
+            expected.makerAssetProxyId = '0xf47261b0';
+            expected.makerTokenAddress = '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2';
+            expected.makerTokenId = null;
+            expected.rawTakerAssetData = '0xf47261b000000000000000000000000042d6622dece394b54999fbd73d108123806f6a18';
+            expected.takerAssetType = 'erc20';
+            expected.takerAssetProxyId = '0xf47261b0';
+            expected.takerTokenAddress = '0x42d6622dece394b54999fbd73d108123806f6a18';
+            expected.takerTokenId = null;
+            expected.metadataJson = '{"isThisArbitraryData":true,"powerLevel":9001}';
+
+            const actual = _convertToEntity(input);
+            expect(actual).deep.equal(expected);
+        });
+    });
+});
diff --git a/packages/pipeline/test/utils/chai_setup.ts b/packages/pipeline/test/utils/chai_setup.ts
new file mode 100644
index 000000000..1a8733093
--- /dev/null
+++ b/packages/pipeline/test/utils/chai_setup.ts
@@ -0,0 +1,13 @@
+import * as chai from 'chai';
+import chaiAsPromised = require('chai-as-promised');
+import ChaiBigNumber = require('chai-bignumber');
+import * as dirtyChai from 'dirty-chai';
+
+export const chaiSetup = {
+    configure(): void {
+        chai.config.includeStack = true;
+        chai.use(ChaiBigNumber());
+        chai.use(dirtyChai);
+        chai.use(chaiAsPromised);
+    },
+};