<commit_before>var EventEmitter = require('events').EventEmitter;
var inherits = require('inherits');
var through = require('through');
var resumer = require('resumer');
var inspect = require('object-inspect');
var nextTick = typeof setImmediate !== 'undefined'
    ? setImmediate
    : process.nextTick;

module.exports = Results;
inherits(Results, EventEmitter);

function Results () {
    if (!(this instanceof Results)) return new Results;
    this.count = 0;
    this.fail = 0;
    this.pass = 0;
    this._stream = through();
    this.tests = [];
}

Results.prototype.createStream = function (opts) {
    if (!opts) opts = {};
    var self = this;
    var output, testId = 0;
    if (opts.objectMode) {
        output = through();
        self.on('_push', function ontest (t, extra) {
            if (!extra) extra = {};
            var id = testId++;
            t.once('prerun', function () {
                var row = {
                    type: 'test',
                    name: t.name,
                    id: id
                };
                if (has(extra, 'parent')) {
                    row.parent = extra.parent;
                }
                output.queue(row);
            });
            t.on('test', function (st) {
                ontest(st, { parent: id });
            });
            t.on('result', function (res) {
                res.test = id;
                output.queue(res);
            });
            t.on('end', function () {
                output.queue({ type: 'end', test: id });
            });
        });
        self.on('done', function () { output.queue(null) });
    }
    else {
        output = resumer();
        output.queue('TAP version 13\n');
        self._stream.pipe(output);
    }

    nextTick(function next() {
        var t;
        while ((t = getNextTest(self))) {
            t.run();
            if (!t.ended) return t.once('end', function(){ nextTick(next); });
        }
        self.emit('done');
    });

    return output;
};

Results.prototype.push = function (t) {
    var self = this;
    self.tests.push(t);
    self._watch(t);
    self.emit('_push', t);
};

Results.prototype.only = function (name) {
    if (this._only) {
        self.count ++;
        self.fail ++;
        write('not ok ' + self.count + ' already called .only()\n');
    }
    this._only = name;
};

Results.prototype._watch = function (t) {
    var self = this;
    var write = function (s) { self._stream.queue(s) };
    t.once('prerun', function () {
        write('# ' + t.name + '\n');
    });

    t.on('result', function (res) {
        if (res.type == 'assert') {
            self.count++;
            res.ok ? self.pass++ : self.fail++;
        }

        write(encodeResult(res, self.count));
    });

    t.on('test', function (st) { self._watch(st) });
};

Results.prototype.close = function () {
    var self = this;
    if (self.closed) self._stream.emit('error', new Error('ALREADY CLOSED'));
    self.closed = true;
    var write = function (s) { self._stream.queue(s) };

    write('\n1..' + self.count + '\n');
    write('# tests ' + self.count + '\n');
    write('# pass  ' + self.pass + '\n');
    if (self.fail) {
        write('# fail  ' + self.fail + '\n');
    } else {
        write('\n# ok\n');
    }

    self._stream.queue(null);
};

function encodeResult (res, count) {
    if (res.type == 'message') {
        return '# ' + res.name.trim().replace(/^#\s*/, '') + '\n';
    }

    if (res.type != 'assert') {
        return ''; //silently ignoring everything else
    }

    var output = '';
    output += (res.ok ? 'ok ' : 'not ok ');
    output += count + ' ';
    output += res.name.trim().replace(/\s+/g, ' ') || '(unnamed assert)';

    if (res.skip) output += ' # SKIP';
    else if (res.todo) output += ' # TODO';

    output += '\n';
    if (res.ok) return output;

    var outer = '  ';
    var inner = outer + '  ';
    output += outer + '---\n';
    output += inner + 'operator: ' +
        (res.operator || '(unknown operator)') + '\n';

    if (has(res, 'expected') || has(res, 'actual')) {
        var ex = inspect(res.expected);
        var ac = inspect(res.actual);

        if (Math.max(ex.length, ac.length) > 65) {
            output += inner + 'expected:\n' + inner + '  ' + ex + '\n';
            output += inner + 'actual:\n' + inner + '  ' + ac + '\n';
        }
        else {
            output += inner + 'expected: ' + ex + '\n';
            output += inner + 'actual:   ' + ac + '\n';
        }
    }
    if (res.at) {
        output += inner + 'at: ' + res.at + '\n';
    }
    if (res.operator === 'error' && res.actual && res.actual.stack) {
        var lines = String(res.actual.stack).split('\n');
        output += inner + 'stack:\n';
        output += inner + '  ' + lines[0] + '\n';
        for (var i = 1; i < lines.length; i++) {
            output += inner + lines[i] + '\n';
        }
    }

    output += outer + '...\n';
    return output;
}

function getNextTest (results) {
    if (!results._only) {
        return results.tests.shift();
    }

    while (results.tests.length) {
        var t = results.tests.shift();
        if (t && t.name === results._only) {
            return t;
        }
    }

    return null;
}

function has (obj, prop) {
    return Object.prototype.hasOwnProperty.call(obj, prop);
}
<commit_msg>separation of _watchObjectStream as a function<commit_after>var EventEmitter = require('events').EventEmitter;
var inherits = require('inherits');
var through = require('through');
var resumer = require('resumer');
var inspect = require('object-inspect');
var nextTick = typeof setImmediate !== 'undefined'
    ? setImmediate
    : process.nextTick;

module.exports = Results;
inherits(Results, EventEmitter);

function Results () {
    if (!(this instanceof Results)) return new Results;
    this.count = 0;
    this.fail = 0;
    this.pass = 0;
    this._stream = through();
    this.tests = [];
}

Results.prototype.createStream = function (opts) {
    if (!opts) opts = {};
    var self = this;
    var output;
    if (opts.objectMode) {
        output = through();
        self._watchObjectStream(output);
    }
    else {
        output = resumer();
        output.queue('TAP version 13\n');
        self._stream.pipe(output);
    }

    nextTick(function next() {
        var t;
        while ((t = getNextTest(self))) {
            t.run();
            if (!t.ended) return t.once('end', function(){ nextTick(next); });
        }
        self.emit('done');
    });

    return output;
};

Results.prototype.push = function (t) {
    var self = this;
    self.tests.push(t);
    self._watch(t);
    self.emit('_push', t);
};

Results.prototype.only = function (name) {
    if (this._only) {
        self.count ++;
        self.fail ++;
        write('not ok ' + self.count + ' already called .only()\n');
    }
    this._only = name;
};

Results.prototype._watch = function (t) {
    var self = this;
    var write = function (s) { self._stream.queue(s) };
    t.once('prerun', function () {
        write('# ' + t.name + '\n');
    });

    t.on('result', function (res) {
        if (res.type == 'assert') {
            self.count++;
            res.ok ? self.pass++ : self.fail++;
        }

        write(encodeResult(res, self.count));
    });

    t.on('test', function (st) { self._watch(st) });
};


Results.prototype._watchObjectStream = function(output) {
    var self = this;
    var testId = 0;

    self.on('_push', function ontest (t, extra) {
        if (!extra) extra = {};
        var id = testId++;
        t.once('prerun', function () {
            var row = {
                type: 'test',
                name: t.name,
                id: id
            };
            if (has(extra, 'parent')) {
                row.parent = extra.parent;
            }
            output.queue(row);
        });
        t.on('test', function (st) {
            ontest(st, { parent: id });
        });
        t.on('result', function (res) {
            res.test = id;
            output.queue(res);
        });
        t.on('end', function () {
            output.queue({ type: 'end', test: id });
        });
    });

    self.on('done', function () {
        output.queue(null);
    });
};

Results.prototype.close = function () {
    var self = this;
    if (self.closed) self._stream.emit('error', new Error('ALREADY CLOSED'));
    self.closed = true;
    var write = function (s) { self._stream.queue(s) };

    write('\n1..' + self.count + '\n');
    write('# tests ' + self.count + '\n');
    write('# pass  ' + self.pass + '\n');
    if (self.fail) {
        write('# fail  ' + self.fail + '\n');
    } else {
        write('\n# ok\n');
    }

    self._stream.queue(null);
};

function encodeResult (res, count) {
    if (res.type == 'message') {
        return '# ' + res.name.trim().replace(/^#\s*/, '') + '\n';
    }

    if (res.type != 'assert') {
        return ''; //silently ignoring everything else
    }

    var output = '';
    output += (res.ok ? 'ok ' : 'not ok ');
    output += count + ' ';
    output += res.name.trim().replace(/\s+/g, ' ') || '(unnamed assert)';

    if (res.skip) output += ' # SKIP';
    else if (res.todo) output += ' # TODO';

    output += '\n';
    if (res.ok) return output;

    var outer = '  ';
    var inner = outer + '  ';
    output += outer + '---\n';
    output += inner + 'operator: ' +
        (res.operator || '(unknown operator)') + '\n';

    if (has(res, 'expected') || has(res, 'actual')) {
        var ex = inspect(res.expected);
        var ac = inspect(res.actual);

        if (Math.max(ex.length, ac.length) > 65) {
            output += inner + 'expected:\n' + inner + '  ' + ex + '\n';
            output += inner + 'actual:\n' + inner + '  ' + ac + '\n';
        }
        else {
            output += inner + 'expected: ' + ex + '\n';
            output += inner + 'actual:   ' + ac + '\n';
        }
    }
    if (res.at) {
        output += inner + 'at: ' + res.at + '\n';
    }
    if (res.operator === 'error' && res.actual && res.actual.stack) {
        var lines = String(res.actual.stack).split('\n');
        output += inner + 'stack:\n';
        output += inner + '  ' + lines[0] + '\n';
        for (var i = 1; i < lines.length; i++) {
            output += inner + lines[i] + '\n';
        }
    }

    output += outer + '...\n';
    return output;
}

function getNextTest (results) {
    if (!results._only) {
        return results.tests.shift();
    }

    while (results.tests.length) {
        var t = results.tests.shift();
        if (t && t.name === results._only) {
            return t;
        }
    }

    return null;
}

function has (obj, prop) {
    return Object.prototype.hasOwnProperty.call(obj, prop);
}

{'fraction_non_alphanumeric': 0.18603139221695922, 'fraction_numerical': 0.0020432803937958577, 'mean_word_length': 1.8501852832186343, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2075834', 'n_tokens_mistral': 3467, 'n_tokens_neox': 3396, 'n_words': 800}