<commit_before>#!/usr/bin/env python

"""
``stbt auto-selftest`` captures the behaviour of Frame Objects (and other
helper functions that operate on a video-frame) by generating doctests. When
you change these helper functions, the generated doctests help to ensure that
they still behave correctly.

Usage:

    stbt auto-selftest generate [source_file.py ...]
    stbt auto-selftest validate

``stbt auto-selftest generate`` generates a doctest for every `FrameObject` in
your test-pack against any screenshots stored in ``selftest/screenshots``. This
results in a set of python files under ``selftest/auto_selftest`` which can be
inspected by (human) eye and validated either with ``python -m doctest
<filename>`` or more commonly with ``stbt auto-selftest validate``.

**auto-selftest checklist**

1. Take screenshots of the device-under-test and put them in the
   ``selftest/screenshots`` directory of your test-pack.
2. Run ``stbt auto-selftest generate`` after every change to your Frame
   Objects, or after adding a new screenshot.
3. View the effect of your changes with ``git diff``.
4. Commit the changes to your auto-selftests along with your changes to the
   Frame Objects.
5. Run ``stbt auto-selftest validate`` on every change from your Continuous
   Integration system.

Using auto-selftest makes it much easier to create, update and modify Frame
Objects. If you find a screen where your Frame Object doesn't behave properly,
add that screenshot to your selftest corpus, and fix the Frame Object;
auto-selftest will check that you haven't introduced a regression in the Frame
Object's behaviour against the other screenshots.

For more information and for more advanced usage see the example test file
(``tests/example.py``), the accompanying screenshots
(``selftest/screenshots``), and the generated doctest
(``selftest/auto_selftest/tests/example_selftest.py``) under
<https://github.com/stb-tester/stb-tester/tree/master/tests/auto-selftest-example-test-pack/>.

For more information on the background behind auto-selftests see
`Improve black-box testing agility: automatic self-regression tests
<https://stb-tester.com/blog/2015/09/24/automatic-self-regression-tests>`_.
"""

import argparse
import cStringIO
import errno
import fnmatch
import multiprocessing
import os
import re
import shutil
import signal
import StringIO
import sys
import tempfile
import time
import traceback
from collections import namedtuple
from textwrap import dedent, wrap

from _stbt.imgproc_cache import cache
from _stbt.utils import find_import_name, mkdir_p, rm_f

SCREENSHOTS_ROOT = "selftest/screenshots"


def main(argv):
    parser = argparse.ArgumentParser(
        prog="stbt auto-selftest",
        epilog=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    subparsers = parser.add_subparsers(dest="command")

    subparser_generate = subparsers.add_parser(
        'generate', help="Regenerate auto-selftests from screenshots")
    subparser_generate.add_argument(
        "source_files", nargs="*", help="""Python source file(s) to search for
        FrameObjects (defaults to all python files in the test-pack)""")

    subparsers.add_parser('validate', help='Run (and check) the auto-selftests')

    cmdline_args = parser.parse_args(argv[1:])

    root = _find_test_pack_root()
    if root is None:
        sys.stderr.write(
            "error: This command must be run within a test pack. Couldn't find "
            "a .stbt.conf in this or any parent directory.\n")
        return 1

    os.chdir(root)

    if cmdline_args.command == 'generate':
        return generate(cmdline_args.source_files)
    elif cmdline_args.command == 'validate':
        return validate()
    else:
        assert False


def generate(source_files):
    tmpdir, modules = generate_into_tmpdir(source_files)
    target_dir = os.path.join(os.curdir, "selftest/auto_selftest")
    try:
        if source_files:
            # Only replace the selftests for the specified files.
            for f in modules:
                newfile = os.path.join(tmpdir, selftest_filename(f.filename))
                target = os.path.join(target_dir, selftest_filename(f.filename))
                if os.path.exists(newfile):
                    mkdir_p(os.path.dirname(target))
                    os.rename(newfile, target)
                else:
                    rm_f(target)
                    if f.error:
                        sys.stderr.write(
                            "error: '%s' isn't a valid source file.\n"
                            % f.filename)
                        return 1
                    sys.stderr.write(
                        "warning: '%s' doesn't define any selftests.\n"
                        % f.filename)
            prune_empty_directories(target_dir)
        else:
            # Replace all selftests, deleting selftests for source files that
            # no longer exist.
            if os.path.exists(target_dir):
                shutil.rmtree(target_dir)
            os.rename(tmpdir, target_dir)

    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


def validate():
    import filecmp
    tmpdir, _ = generate_into_tmpdir()
    try:
        orig_files = _recursive_glob(
            '*.py', "%s/selftest/auto_selftest" % os.curdir)
        new_files = _recursive_glob('*.py', tmpdir)
        if orig_files != new_files:
            return 1
        _, mismatch, errors = filecmp.cmpfiles(
            tmpdir, "%s/selftest/auto_selftest" % os.curdir, orig_files)
        if mismatch or errors:
            return 1
        else:
            return 0
    finally:
        shutil.rmtree(tmpdir)


def is_valid_python_identifier(x):
    return bool(re.match('^[a-zA-Z_][a-zA-Z0-9_]*$', x))


def prune_empty_directories(dir_):
    for root, _dirs, files in os.walk(dir_, topdown=False):
        if len(files) == 0 and root != dir_:
            try:
                os.rmdir(root)
            except OSError as e:
                if e.errno not in [errno.EEXIST, errno.ENOTEMPTY]:
                    raise


def init_worker():
    signal.signal(signal.SIGINT, signal.SIG_IGN)


def iterate_with_progress(sequence, width=20, stream=sys.stderr):
    stream.write('\n')
    total = len(sequence)
    for n, v in enumerate(sequence):
        stream.write(_progress_line(width, n, total) +
                     " - Processing %s\r" % v)
        yield v
    stream.write(_progress_line(width, total, total) + "\n")


def _progress_line(width, n, total):
    ANSI_ERASE_LINE = '\033[K'
    progress = (n * width) // total
    return '\r' + ANSI_ERASE_LINE + '[%s] %3d / %d' % (
        '#' * progress + ' ' * (width - progress), n, total)


def generate_into_tmpdir(source_files=None):
    start_time = time.time()

    selftest_dir = "%s/selftest" % os.curdir
    mkdir_p(selftest_dir)
    # We use this process pool for sandboxing rather than concurrency:
    pool = multiprocessing.Pool(
        processes=1, maxtasksperchild=1, initializer=init_worker)
    tmpdir = tempfile.mkdtemp(dir=selftest_dir, prefix="auto_selftest")
    modules = []
    try:
        if not source_files:
            source_files = valid_source_files(_recursive_glob('*.py'))
        perf_log = []
        test_file_count = 0
        for module_filename in iterate_with_progress(source_files):
            outname = os.path.join(tmpdir, selftest_filename(module_filename))
            barename = re.sub('.py$', '_bare.py', outname)
            mkdir_p(os.path.dirname(outname))

            module = pool.apply(inspect_module, (module_filename,))
            modules.append(module)
            test_line_count = write_bare_doctest(module, barename)
            if test_line_count:
                test_file_count += 1
                perf_log.extend(pool.apply_async(
                    update_doctests, (barename, outname)).get(timeout=60 * 60))
                os.unlink(barename)

        if test_file_count > 0:
            with open('%s/README' % tmpdir, 'w') as f:
                f.write("\n".join(wrap(
                    "This directory contains self-tests generated by `stbt "
                    "auto-selftest`. Do not modify by hand. Any files "<commit_msg>stbt auto-selftest: Pre-import stbt & dependencies for better performance<commit_after>#!/usr/bin/env python

"""
``stbt auto-selftest`` captures the behaviour of Frame Objects (and other
helper functions that operate on a video-frame) by generating doctests. When
you change these helper functions, the generated doctests help to ensure that
they still behave correctly.

Usage:

    stbt auto-selftest generate [source_file.py ...]
    stbt auto-selftest validate

``stbt auto-selftest generate`` generates a doctest for every `FrameObject` in
your test-pack against any screenshots stored in ``selftest/screenshots``. This
results in a set of python files under ``selftest/auto_selftest`` which can be
inspected by (human) eye and validated either with ``python -m doctest
<filename>`` or more commonly with ``stbt auto-selftest validate``.

**auto-selftest checklist**

1. Take screenshots of the device-under-test and put them in the
   ``selftest/screenshots`` directory of your test-pack.
2. Run ``stbt auto-selftest generate`` after every change to your Frame
   Objects, or after adding a new screenshot.
3. View the effect of your changes with ``git diff``.
4. Commit the changes to your auto-selftests along with your changes to the
   Frame Objects.
5. Run ``stbt auto-selftest validate`` on every change from your Continuous
   Integration system.

Using auto-selftest makes it much easier to create, update and modify Frame
Objects. If you find a screen where your Frame Object doesn't behave properly,
add that screenshot to your selftest corpus, and fix the Frame Object;
auto-selftest will check that you haven't introduced a regression in the Frame
Object's behaviour against the other screenshots.

For more information and for more advanced usage see the example test file
(``tests/example.py``), the accompanying screenshots
(``selftest/screenshots``), and the generated doctest
(``selftest/auto_selftest/tests/example_selftest.py``) under
<https://github.com/stb-tester/stb-tester/tree/master/tests/auto-selftest-example-test-pack/>.

For more information on the background behind auto-selftests see
`Improve black-box testing agility: automatic self-regression tests
<https://stb-tester.com/blog/2015/09/24/automatic-self-regression-tests>`_.
"""

import argparse
import cStringIO
import errno
import fnmatch
import multiprocessing
import os
import re
import shutil
import signal
import StringIO
import sys
import tempfile
import time
import traceback
from collections import namedtuple
from textwrap import dedent, wrap

from _stbt.imgproc_cache import cache
from _stbt.utils import find_import_name, mkdir_p, rm_f

SCREENSHOTS_ROOT = "selftest/screenshots"


def main(argv):
    parser = argparse.ArgumentParser(
        prog="stbt auto-selftest",
        epilog=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    subparsers = parser.add_subparsers(dest="command")

    subparser_generate = subparsers.add_parser(
        'generate', help="Regenerate auto-selftests from screenshots")
    subparser_generate.add_argument(
        "source_files", nargs="*", help="""Python source file(s) to search for
        FrameObjects (defaults to all python files in the test-pack)""")

    subparsers.add_parser('validate', help='Run (and check) the auto-selftests')

    cmdline_args = parser.parse_args(argv[1:])

    root = _find_test_pack_root()
    if root is None:
        sys.stderr.write(
            "error: This command must be run within a test pack. Couldn't find "
            "a .stbt.conf in this or any parent directory.\n")
        return 1

    os.chdir(root)

    if cmdline_args.command == 'generate':
        return generate(cmdline_args.source_files)
    elif cmdline_args.command == 'validate':
        return validate()
    else:
        assert False


def generate(source_files):
    tmpdir, modules = generate_into_tmpdir(source_files)
    target_dir = os.path.join(os.curdir, "selftest/auto_selftest")
    try:
        if source_files:
            # Only replace the selftests for the specified files.
            for f in modules:
                newfile = os.path.join(tmpdir, selftest_filename(f.filename))
                target = os.path.join(target_dir, selftest_filename(f.filename))
                if os.path.exists(newfile):
                    mkdir_p(os.path.dirname(target))
                    os.rename(newfile, target)
                else:
                    rm_f(target)
                    if f.error:
                        sys.stderr.write(
                            "error: '%s' isn't a valid source file.\n"
                            % f.filename)
                        return 1
                    sys.stderr.write(
                        "warning: '%s' doesn't define any selftests.\n"
                        % f.filename)
            prune_empty_directories(target_dir)
        else:
            # Replace all selftests, deleting selftests for source files that
            # no longer exist.
            if os.path.exists(target_dir):
                shutil.rmtree(target_dir)
            os.rename(tmpdir, target_dir)

    finally:
        shutil.rmtree(tmpdir, ignore_errors=True)


def validate():
    import filecmp
    tmpdir, _ = generate_into_tmpdir()
    try:
        orig_files = _recursive_glob(
            '*.py', "%s/selftest/auto_selftest" % os.curdir)
        new_files = _recursive_glob('*.py', tmpdir)
        if orig_files != new_files:
            return 1
        _, mismatch, errors = filecmp.cmpfiles(
            tmpdir, "%s/selftest/auto_selftest" % os.curdir, orig_files)
        if mismatch or errors:
            return 1
        else:
            return 0
    finally:
        shutil.rmtree(tmpdir)


def is_valid_python_identifier(x):
    return bool(re.match('^[a-zA-Z_][a-zA-Z0-9_]*$', x))


def prune_empty_directories(dir_):
    for root, _dirs, files in os.walk(dir_, topdown=False):
        if len(files) == 0 and root != dir_:
            try:
                os.rmdir(root)
            except OSError as e:
                if e.errno not in [errno.EEXIST, errno.ENOTEMPTY]:
                    raise


def init_worker():
    signal.signal(signal.SIGINT, signal.SIG_IGN)


def iterate_with_progress(sequence, width=20, stream=sys.stderr):
    stream.write('\n')
    total = len(sequence)
    for n, v in enumerate(sequence):
        stream.write(_progress_line(width, n, total) +
                     " - Processing %s\r" % v)
        yield v
    stream.write(_progress_line(width, total, total) + "\n")


def _progress_line(width, n, total):
    ANSI_ERASE_LINE = '\033[K'
    progress = (n * width) // total
    return '\r' + ANSI_ERASE_LINE + '[%s] %3d / %d' % (
        '#' * progress + ' ' * (width - progress), n, total)


def generate_into_tmpdir(source_files=None):
    start_time = time.time()

    # Importing stbt + gstreamer bindings + opencv is slow.
    # multiprocessing.Pool uses `fork` so by importing here, these modules
    # won't be imported from scratch in each subprocess.
    import stbt  # pylint:disable=unused-variable

    selftest_dir = "%s/selftest" % os.curdir
    mkdir_p(selftest_dir)
    # We use this process pool for sandboxing rather than concurrency:
    pool = multiprocessing.Pool(
        processes=1, maxtasksperchild=1, initializer=init_worker)
    tmpdir = tempfile.mkdtemp(dir=selftest_dir, prefix="auto_selftest")
    modules = []
    try:
        if not source_files:
            source_files = valid_source_files(_recursive_glob('*.py'))
        perf_log = []
        test_file_count = 0
        for module_filename in iterate_with_progress(source_files):
            outname = os.path.join(tmpdir, selftest_filename(module_filename))
            barename = re.sub('.py$', '_bare.py', outname)
            mkdir_p(os.path.dirname(outname))

            module = pool.apply(inspect_module, (module_filename,))
            modules.append(module)
            test_line_count = write_bare_doctest(module, barename)
            if test_line_count:
                test_file_count += 1
                perf_log.extend(pool.apply_async(
                    update_doctests, (barename, outname)).get(timeout=60 * 60))
                os.unlink(barename)

        if test_file_count > 0:
            with open('%s/README' % tmpdir, 'w') as f:
                f.write("\n".join(wrap(
                    "This directory contains self-tests generated by `stbt "
                    "auto-selftest`. Do not modify by hand. Any files "
{'fraction_non_alphanumeric': 0.09537485789505176, 'fraction_numerical': 0.0044276910189672706, 'mean_word_length': 2.6413943355119827, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 4, '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': '3180489', 'n_tokens_mistral': 5080, 'n_tokens_neox': 4928, 'n_words': 1418}