<commit_before>#!/usr/bin/python
################################################################################
# "THE BEER-WARE LICENSE" (Revision 42):
# <thenoviceoof> wrote this file. As long as you retain this notice
# you can do whatever you want with this stuff. If we meet some day,
# and you think this stuff is worth it, you can buy me a beer in
# return
# <NAME> <thenoviceoof>
################################################################################

import re
import sys
import nose
import Queue
import curses
import threading
import exceptions

from nose.plugins import Plugin

################################################################################
# windowing stuff

def curses_main(scr, test_queue):
    '''
    The curses loop
    Poll for input, new tests
    '''
    tests = {
        'ok': [],
        'fail': [],
        'error': [],
    }

    curses.use_default_colors()
    curses.curs_set(0)
    curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_WHITE)
    # wait for a character for only 0.1s
    curses.halfdelay(1)

    size = None
    new_tests = False
    status_bar = None
    try:
        while 1:
            # handle input
            prev_size = size
            size = scr.getmaxyx()[1], scr.getmaxyx()[0]
            c = scr.getch()
            if c == ord('q'):
                return
            elif c == curses.KEY_RESIZE or size != prev_size:
                status_bar = None
            # handle any new tests
            try:
                t = test_queue.get(block=False)
                tests[t[0]].append(t[1])
                new_tests = True
            except Queue.Empty:
                new_tests = False

            if status_bar is None:
                status_bar = curses.newwin(1,size[0])
                msg = 'Running tests...'
                status_bar.bkgdset(ord(' '), curses.color_pair(1))
                status_bar.clear()
                status_bar.addstr(0,0, msg + ' ' * (size[0] - len(msg) - 1))
                status_bar.refresh()
            if new_tests:
                status_bar.clear()
                ss = ['total: {0}'.format(sum([len(t) for t in tests.values()]))]
                ss += ['{0}: {1}'.format(x,len(tests[x]))
                       for x in ['ok', 'error', 'fail']]
                counts = ' | '.join(ss)
                status_bar.addstr(0,0, counts)
                status_bar.refresh()

            # scr.addstr(1,0, str(len(tests['ok'])))
            # scr.refresh()
    except KeyboardInterrupt:
        return

def curses_run(test_queue):
    '''
    Little wrapper to facilitate thread running
    '''
    curses.wrapper(curses_main, test_queue)

################################################################################
# nose plugin

class BoogerPlugin(Plugin):
    def __init__(self, *args, **kwargs):
        super(BoogerPlugin, self).__init__(*args, **kwargs)

        self.test_queue = Queue.Queue()
        self.curses = threading.Thread(target=curses_run,
                                       args=(self.test_queue,))
        self.curses.start()

    ############################################################################
    # utils<commit_msg>Display the test (name) in the main window<commit_after>#!/usr/bin/python
################################################################################
# "THE BEER-WARE LICENSE" (Revision 42):
# <thenoviceoof> wrote this file. As long as you retain this notice
# you can do whatever you want with this stuff. If we meet some day,
# and you think this stuff is worth it, you can buy me a beer in
# return
# <NAME> <thenoviceoof>
################################################################################

import re
import sys
import nose
import Queue
import curses
import threading
import exceptions

from nose.plugins import Plugin

################################################################################
# windowing stuff

def curses_main(scr, test_queue):
    '''
    The curses loop
    Poll for input, new tests
    '''
    test_counts = {
        'ok': 0,
        'fail': 0,
        'error': 0,
    }
    # keep fail/error tests in a fixed list
    tests = []

    curses.use_default_colors()
    curses.curs_set(0)
    curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_WHITE)
    # wait for a character for only 0.1s
    curses.halfdelay(1)

    size = None
    new_tests = False
    status_bar = None
    try:
        while 1:
            # handle input
            prev_size = size
            size = scr.getmaxyx()[1], scr.getmaxyx()[0]
            c = scr.getch()
            if c == ord('q'):
                return
            elif c == curses.KEY_RESIZE or size != prev_size:
                status_bar = None
            # handle any new tests
            try:
                t = test_queue.get(block=False)
                test_counts[t[0]] += 1
                if t[0] != 'ok':
                    tests.append(t[1])
                new_tests = True
            except Queue.Empty:
                new_tests = False

            if status_bar is None:
                status_bar = curses.newwin(1,size[0])
                msg = 'Running tests...'
                status_bar.attrset(curses.A_BOLD)
                status_bar.bkgdset(ord(' '), curses.color_pair(1))
                status_bar.clear()
                status_bar.addstr(0,0, msg + ' ' * (size[0] - len(msg) - 1))
                status_bar.refresh()
            if new_tests:
                # update status bar
                status_bar.clear()
                ss = ['total: {0}'.format(sum([v for v in test_counts.values()]))]
                ss += ['{0}: {1}'.format(x,test_counts[x])
                       for x in ['ok', 'error', 'fail']]
                counts = ' | '.join(ss)
                status_bar.addstr(0,0, counts)
                status_bar.refresh()
                # update test list
                for i in range(len(tests)):
                    t = tests[i]
                    scr.addstr(i+1, 0, str(t))
                scr.refresh()
    except KeyboardInterrupt:
        return

def curses_run(test_queue):
    '''
    Little wrapper to facilitate thread running
    '''
    curses.wrapper(curses_main, test_queue)

################################################################################
# nose plugin

class BoogerPlugin(Plugin):
    def __init__(self, *args, **kwargs):
        super(BoogerPlugin, self).__init__(*args, **kwargs)

        self.test_queue = Queue.Queue()
        self.curses = threading.Thread(target=curses_run,
                                       args=(self.test_queue,))
        self.curses.start()

    ############################################################################
    # utils
{'fraction_non_alphanumeric': 0.21739774748073504, 'fraction_numerical': 0.008446947243627741, 'mean_word_length': 2.0719162494310424, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '438000', 'n_tokens_mistral': 1941, 'n_tokens_neox': 1837, 'n_words': 470}