update.py

#!/usr/bin/env python
"""Helper script that auto-updates all response data for integration tests"""
# TODO: rework implementation to use logging instead of print so we can see timestamps of operations
# TODO: add a nested progress bar to show number of remaining tests to fix
# TODO: add support for verbosity levels (ie: by default just show progress bars)
from time import sleep
import math
import sys
import shlex
import json
from pathlib import Path
from contextlib import redirect_stdout, redirect_stderr
from datetime import datetime
from dateutil import tz
import pytest
from pytest import ExitCode
from tqdm import trange
import humanize
import click
from click.exceptions import ClickException, Abort
from friendlypins.utils.rest_io import RestIO


CUR_PATH = Path(__file__).parent
DEFAULT_KEY_FILE = CUR_PATH.joinpath("key.txt")
DEBUG_LOG_FILE = CUR_PATH.joinpath("debug.log")
CASSETTE_PATH = CUR_PATH.joinpath("tests").joinpath("cassettes")
REPORT_FILE = CUR_PATH.joinpath(".report.json")
PREVIOUS_REPORT = None


def get_secret():
    """Loads authentication token for Pinterest

    Returns:
        str: authentication token parsed from the file
    """
    if not DEFAULT_KEY_FILE.exists():
        raise Exception("Authentication key must be stored in a file named " + DEFAULT_KEY_FILE.name)

    retval = DEFAULT_KEY_FILE.read_text().strip()
    if not retval or len(retval) < 10:
        raise Exception("Invalid authentication token")
    return retval


def load_report():
    """Loads unit test data from the latest pytest report

    Requires the pytest-json-report plugin
    Assumes the output is stored i a file named .report.json in the current folder

    Returns:
        dict: parsed report data
    """

    if not REPORT_FILE.exists():
        raise Exception("pytest report file not found: " + REPORT_FILE.name)

    retval = json.loads(REPORT_FILE.read_text())
    # Reformat our JSON report to make it easier to read
    REPORT_FILE.write_text(json.dumps(retval, indent=4))
    return retval

def analyse_report(report):
    """Analyses a pytest report, and displays summary information to the console

    Args:
        report (dict):
            pytest report data, as generated by the :meth:`load_report` method

    Returns:
        int: number of failing unit tests still remaining
    """
    global PREVIOUS_REPORT  # pylint: disable=global-statement

    if report["summary"]["total"] == 0:
        raise Exception("pytest report has no test results")

    current_failures = list()
    for cur_test in report["tests"]:
        if cur_test["outcome"] in ("passed", "skipped"):
            continue
        if "RateLimitException" not in str(cur_test) and "Network is disabled" not in str(cur_test):
            raise Exception("Unit test {0} has failed for unexpected reasons. See debug.log for details".format(
                cur_test["nodeid"]))
        current_failures.append(cur_test["nodeid"])

    click.secho(
        "{0} of the {1} selected tests were successful".format(
            report["summary"].get("passed", 0),
            report["summary"]["total"]
        ),
        fg="green"
    )

    if PREVIOUS_REPORT:
        fixed_tests = list()
        for cur_test in PREVIOUS_REPORT["tests"]:
            if cur_test["outcome"] in ("passed", "skipped"):
                continue
            if cur_test["nodeid"] not in current_failures:
                fixed_tests.append(cur_test["nodeid"])
        if fixed_tests:
            click.secho("Fixed the following {0} tests:".format(len(fixed_tests)), fg="green")
            for cur_test in fixed_tests:
                click.secho("\t{0}".format(cur_test), fg="green")

    PREVIOUS_REPORT = report
    return report["summary"].get("failed", 0)


def sanity_check(secret):
    """Makes sure there are no further mentions of our auth token anywhere in any cassette

    Args:
        secret (str):
            Auth token to detect

    Returns:
        bool:
            True if everything looks OK, False if there are still mentions of the auth token in 1 or more cassettes
    """
    matches = list()
    for cur_file in CASSETTE_PATH.rglob("*.yaml"):
        if secret in cur_file.read_text():
            matches.append(cur_file)
    if matches:
        click.secho("Found {0} cassettes that still mention auth token:".format(len(matches)), fg="red")
        for cur_match in matches:
            click.secho("\t{0}".format(cur_match.name), fg="red")
        return False
    click.secho("Cassettes look clean - no mentions of auth tokens!", fg="green")
    return True


def run_tests(params):
    """Launches pytest to orchestrate a test run

    All output from the test runner will be hidden to keep the console clean

    Args:
        params (list of str):
            command line parameters to pass to the test runner
            these options will be combined with a default set defined internally

    Returns:
        int: return code produced by the test run
    """
    default_test_params = [
        "./tests",
        "-vv",
        "--json-report",
        "--key-file",
        DEFAULT_KEY_FILE.name
    ]

    with DEBUG_LOG_FILE.open("a") as debug_out:
        with redirect_stdout(debug_out):
            with redirect_stderr(sys.stdout):
                return pytest.main(default_test_params + params)


@click.command()
@click.option("--force", is_flag=True,
              help="Forces overwrite of all cassettes even if their tests are currently passing")
def main(force):
    """Regenerates vcrpy cassettes for integration tests, accounting for rate limits
    enforced by the Pinterest REST APIs
    """
    secret = get_secret()
    service = RestIO(secret)

    # Make sure we re-create our debug log for each run
    if DEBUG_LOG_FILE.exists():
        DEBUG_LOG_FILE.unlink()

    if force:
        # Regenerate all cassette data until we hit our rate limit
        click.secho("Regenerating all recorded cassettes")
        result = run_tests(shlex.split("--record-mode=rewrite"))
        num_failures = analyse_report(load_report())
    else:
        click.secho("Generating baseline...")
        # Start by generating a baseline state without using any API calls
        run_tests(shlex.split("--record-mode=none --block-network"))
        num_failures = analyse_report(load_report())
        if num_failures == 0:
            click.secho("All unit tests passed. Aborting rebuild.", fg="yellow")
            click.secho("To force a rebuild of all cassettes try --force", fg="yellow")
            return

        # The re-run any failed tests, forcing the cassettes to get regenerated
        # We append --lf to only rerun the tests that failed on the last pass
        click.secho("Rebuilding initial cassettes...")
        result = run_tests(shlex.split("--record-mode=rewrite --lf"))
        num_failures = analyse_report(load_report())

    iteration = 1
    while result == ExitCode.TESTS_FAILED and num_failures != 0:
        # check headers to see when the next token renewal is
        now = datetime.now(tz=tz.tzlocal())
        renewal = service.headers.time_to_refresh
        service.refresh_headers()
        wait_time = renewal - now
        minutes = math.ceil(wait_time.total_seconds() / 60)

        # if the rate limit has expired wait until the limit has been refreshed
        if minutes > 0:
            click.secho("Next renewal: {0}".format(renewal.astimezone(tz.tzlocal())))
            click.secho("Sleeping for {0} minutes...".format(minutes))
            # Give regular status updates to the user via a progress bar once every minute
            for _ in trange(minutes):
                sleep(60)

            # Give the API a few additional seconds before we try again to account for clock skew
            sleep(10)

        click.secho("Running test iteration " + str(iteration))
        # We append --lf to only rerun the tests that failed on the previous run
        result = run_tests(shlex.split("--record-mode=rewrite --lf"))

        # If the number of failing tests hasn't changed or has gotten worse, we are not making any progress
        # and thus we should exit to avoid a deadlock
        temp = analyse_report(load_report())
        if temp >= num_failures:
            raise Exception("Last unit test run had {0} failures and current run had {1}".format(num_failures, temp))
        num_failures = temp

        # repeat until all tests pass
        iteration += 1

    # return the final test run result to the caller
    if result != ExitCode.OK:
        raise ClickException("Regeneration failed for unexpected reason: " + str(result))


def _main(args):
    """Primary entry point function

    Args:
        args (list of str):
            command line arguments to pass to the command interpreter
    Returns:
        int:
            return code to pass back to the shell
    """
    start = datetime.now()
    try:
        main.main(args, standalone_mode=False)
    except Abort:
        click.secho("Operation aborted!", fg="yellow", bold=True)
    except Exception as err:  # pylint: disable=broad-except
        click.secho("Error: " + str(err), fg="red")
        return 1
    finally:
        if "--help" not in sys.argv:
            # display overall runtime for reference when performing update
            end = datetime.now()
            runtime = end - start
            click.secho("Operation complete. Total runtime: " + humanize.naturaldelta(runtime), fg="green")
    return 0


if __name__ == "__main__":
    sys.exit(_main(sys.argv[1:]))