tasks.py

# -*- coding: utf-8 -*-

try:
    from invoke import ctask as task
except ImportError:
    from invoke import task

import os
import sys
import time

import requests
from requests.auth import HTTPBasicAuth


@task
def style(c):
    """
    Run PEP style checks against the codebase
    """

    print("Running PEP code style checks...")
    c.run('flake8 .')


@task
def reset_data(c, debug=False):
    """
    Reset the database to a known state.
    This is achieved by installing the InvenTree test fixture data.
    """
    # Reset the database to a known state
    print("Reset test database to a known state (this might take a little while...)")

    hide = None if debug else 'both'

    c.run("docker-compose -f test/docker-compose.yml run --rm inventree-py-test-server invoke dev.delete-data -f", hide=hide)
    c.run("docker-compose -f test/docker-compose.yml run --rm inventree-py-test-server invoke migrate", hide=hide)
    c.run("docker-compose -f test/docker-compose.yml run --rm inventree-py-test-server invoke dev.import-fixtures", hide=hide)


@task(post=[reset_data])
def update_image(c, debug=True, reset=True):
    """
    Update the InvenTree image to the latest version
    """

    print("Pulling latest InvenTree image from docker hub (maybe grab a coffee!)")

    hide = None if debug else 'both'

    c.run("docker-compose -f test/docker-compose.yml pull", hide=hide)
    c.run("docker-compose -f test/docker-compose.yml run --rm inventree-py-test-server invoke update --skip-backup --no-frontend --skip-static", hide=hide)

    if reset:
        reset_data(c)


@task
def check_server(c, host="http://localhost:12345", username="testuser", password="testpassword", debug=True, timeout=30):
    """
    Check that we can ping the server and get a token.
    A generous timeout is applied here, to give the server time to activate
    """

    auth = HTTPBasicAuth(username=username, password=password)

    url = f"{host}/api/user/token/"

    response = None

    while response is None:

        try:
            response = requests.get(url, auth=auth, timeout=0.5)
        except Exception as e:
            if debug:
                print("Error:", str(e))

        if response is None:

            if timeout > 0:
                if debug:
                    print(f"No response from server. {timeout} seconds remaining")
                timeout -= 1
                time.sleep(1)

            else:
                return False

    if response.status_code != 200:
        if debug:
            print(f"Invalid status code: ${response.status_code}")
        return False

    if 'token' not in response.text:
        if debug:
            print("Token not in returned response:")
            print(str(response.text))
        return False

    # We have confirmed that the server is available
    if debug:
        print(f"InvenTree server is available at {host}")

    return True


@task
def start_server(c, debug=False):
    """
    Launch the InvenTree server (in a docker container)
    """

    # Start the InvenTree server
    print("Starting InvenTree server instance")
    c.run('docker-compose -f test/docker-compose.yml up -d', hide=None if debug else 'both')

    print("Waiting for InvenTree server to respond:")

    count = 60

    while not check_server(c, debug=False) and count > 0:
        count -= 1
        time.sleep(1)

    if count == 0:
        print("No response from InvenTree server")
        sys.exit(1)
    else:
        print("InvenTree server is active!")


@task
def stop_server(c, debug=False):
    """
    Stop a running InvenTree test server docker container
    """

    # Stop the server
    c.run('docker-compose -f test/docker-compose.yml down', hide=None if debug else 'both')


@task(help={
    'source': 'Specify the source file to test',
    'update': 'If set, update the docker image before running tests',
    'reset': 'If set, reset test data to a known state',
    'host': 'Specify the InvenTree host address (default = http://localhost:12345)',
    'username': 'Specify the InvenTree username (default = testuser)',
    'password': 'Specify the InvenTree password (default = testpassword)',
    'noserver': 'If set, do not spin up the docker container',
}
)
def test(c, source=None, update=False, reset=False, debug=False, host=None, username=None, password=None, noserver=False):
    """
    Run the unit tests for the python bindings.
    Performs the following steps:

    - Ensure the docker container is up and running
    - Reset the database to a known state (if --reset flag is given)
    - Perform unit testing
    """

    # If a source file is provided, check that it actually exists
    if source:

        if not source.endswith('.py'):
            source += '.py'

        if not os.path.exists(source):
            source = os.path.join('test', source)

        if not os.path.exists(source):
            print(f"Error: Source file '{source}' does not exist")
            sys.exit(1)

    if not noserver:
        if update:
            # Pull down the latest InvenTree docker image
            update_image(c, debug=debug)

        if reset:
            stop_server(c, debug=debug)
            reset_data(c, debug=debug)

        # Launch the InvenTree server (in a docker container)
        start_server(c)

    # Set environment variables so test scripts can access them
    os.environ['INVENTREE_PYTHON_TEST_SERVER'] = host or 'http://localhost:12345'
    os.environ['INVENTREE_PYTHON_TEST_USERNAME'] = username or 'testuser'
    os.environ['INVENTREE_PYTHON_TEST_PASSWORD'] = password or 'testpassword'

    # Run unit tests

    # If a single source file is supplied, test *just* that file
    # Otherwise, test *all* files
    if source:
        print(f"Running tests for '{source}'")
        c.run(f'coverage run -m unittest {source}')
    else:
        # Automatically discover tests, and run only those
        c.run('coverage run -m unittest discover -s test/')