apify--crawlee/test/e2e/run.mjs

/* eslint-disable no-loop-func */
import { execSync } from 'node:child_process';
import { once } from 'node:events';
import { readdir } from 'node:fs/promises';
import { dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { isMainThread, Worker, workerData } from 'node:worker_threads';

import { colors, getApifyToken, clearPackages, clearStorage, SKIPPED_TEST_CLOSE_CODE } from './tools.mjs';

const basePath = dirname(fileURLToPath(import.meta.url));

process.env.APIFY_TOKEN ??= await getApifyToken();
process.env.APIFY_CONTAINER_URL ??= 'http://127.0.0.1';
process.env.APIFY_CONTAINER_PORT ??= '8000';

/**
 * Depending on STORAGE_IMPLEMENTATION the workflow of the tests slightly differs:
 *   - for 'MEMORY': the 'storage' folder should be removed after the test actor finishes;
 *   - for 'LOCAL': the 'apify_storage' folder should be removed after the test actor finishes;
 *   - for 'PLATFORM': SDK packages should be copied to respective test actor folders
 *      (and also should be removed after pushing the actor to platform and starting the test run there)
 *      to check the latest changes on the platform;
 * @default 'MEMORY'
 * @ignore
 */
process.env.STORAGE_IMPLEMENTATION ??= 'MEMORY';

// If any of the tests failed - we want to exit with a non-zero code
// so that the CI knows that e2e test suite has failed
let failure = false;

async function run() {
    if (!['LOCAL', 'MEMORY', 'PLATFORM'].includes(process.env.STORAGE_IMPLEMENTATION)) {
        throw new Error(`Unknown storage provided: '${process.env.STORAGE_IMPLEMENTATION}'`);
    }

    console.log(`Running E2E tests with storage implementation '${process.env.STORAGE_IMPLEMENTATION}'`);

    const paths = await readdir(basePath, { withFileTypes: true });
    const dirs = paths.filter((dirent) => dirent.isDirectory());

    for (const dir of dirs) {
        if (process.argv.length === 3 && dir.name !== process.argv[2]) {
            continue;
        }

        const now = Date.now();
        console.log(`${colors.yellow(`[${dir.name}] `)}${colors.grey('Test starting...')}`);
        const worker = new Worker(fileURLToPath(import.meta.url), {
            workerData: dir.name,
            stdout: true,
            stderr: true,
        });
        let seenFirst = false;
        const prefix = colors.yellow(`[${dir.name}] `);
        // Surface only lines that start with a structured `[…]` marker (init,
        // assertion, build, run, kv, test skipped, etc.). Everything else
        // (crawler INFO logs, per-URL request handler logs, npm warnings, …)
        // is noise on a green run; buffer it and re-emit from the exit
        // handler iff the test failed.
        const deferredOnSuccess = [];

        // Line-buffered streaming so prefixed lines stay intact across chunk boundaries.
        const streamLines = (stream, sink) => {
            let buffer = '';
            stream.on('data', (chunk) => {
                buffer += chunk.toString();
                const lines = buffer.split('\n');
                buffer = lines.pop();
                for (const line of lines) {
                    if (line === '') continue;

                    if (!line.startsWith('[')) {
                        if (!seenFirst) {
                            console.log(
                                `${colors.red('[fatal]')} test ${colors.yellow(
                                    `[${dir.name}]`,
                                )} did not call "initialize(import.meta.url)"!`,
                            );
                            worker.terminate();
                            return;
                        }
                        deferredOnSuccess.push(line);
                        continue;
                    }

                    seenFirst = true;
                    sink(`${prefix}${line}`);
                }
            });
            stream.on('end', () => {
                if (buffer !== '') sink(`${prefix}${buffer}`);
            });
        };

        streamLines(worker.stdout, (line) => console.log(line));
        streamLines(worker.stderr, (line) => console.error(line));

        worker.on('error', (err) => {
            // If the worker emits any error, we want to exit with a non-zero code
            failure = true;
            console.log(`${colors.red('[fatal]')} test ${colors.yellow(`[${dir.name}]`)} failed with error: ${err}`);
        });

        const exitHandler = async (code) => {
            if (code === SKIPPED_TEST_CLOSE_CODE) {
                console.log(`${prefix}${colors.grey('Test skipped')}`);
                return;
            }

            if (code !== 0) {
                for (const line of deferredOnSuccess) console.log(`${prefix}${line}`);
            }

            const took = (Date.now() - now) / 1000;
            const status = code === 0 ? 'success' : 'failure';
            const color = code === 0 ? 'green' : 'red';
            console.log(
                `${colors.yellow(`[${dir.name}] `)}${colors[color](
                    `Test finished with status: ${status} `,
                )}${colors.grey(`[took ${took}s]`)}`,
            );

            if (['MEMORY', 'LOCAL'].includes(process.env.STORAGE_IMPLEMENTATION)) {
                await clearStorage(`${basePath}/${dir.name}`);
            }

            if (process.env.STORAGE_IMPLEMENTATION === 'PLATFORM') {
                await clearPackages(`${basePath}/${dir.name}`);
            }

            if (status === 'failure') failure = true;
        };

        const { promise: waitForExit, resolve: markTestDone } = Promise.withResolvers();

        worker.on('exit', async (exitCode) => {
            try {
                await exitHandler(exitCode);
            } finally {
                markTestDone();
            }
        });

        await waitForExit;
    }
}

if (isMainThread) {
    try {
        if (process.env.STORAGE_IMPLEMENTATION === 'LOCAL') {
            console.log('Temporary installing @apify/storage-local');
            execSync(`yarn add -D "@apify/storage-local@^3.0.0"`, { stdio: 'inherit' });
        }
        if (process.env.STORAGE_IMPLEMENTATION !== 'PLATFORM') {
            console.log('Fetching Camoufox...');

            for (let attempt = 0; attempt < 5; attempt++) {
                try {
                    execSync(`npx camoufox-js fetch`, { stdio: 'inherit' });
                } catch (e) {
                    console.error('Failed to fetch Camoufox', e);
                    if (attempt === 4) throw e;
                    console.log(`Retrying to fetch Camoufox (attempt ${attempt + 2}/5)...`);
                    await new Promise((resolve) => setTimeout(resolve, 10e3));
                    continue;
                }
            }
        }
        await run();
    } catch (e) {
        failure = true;
        console.error(e);
    } finally {
        if (process.env.STORAGE_IMPLEMENTATION === 'LOCAL') {
            console.log('Removing temporary installation of @apify/storage-local');
            execSync(`yarn remove @apify/storage-local`, { stdio: 'inherit' });
        }
    }

    // We want to exit with non-zero code if any of the tests failed
    if (failure) process.exit(1);
} else {
    await import(`${basePath}/${workerData}/test.mjs`);
}