unittest/jsonlog.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#

import json
import os
import subprocess

import fstools
from unittest.output import STATUS_AVAILABLE, STATUS_CHANGED, STATUS_CLEAN, STATUS_EXISTS, STATUS_IN_WORK, STATUS_MISSING, STATUS_OLD, STATUS_RELEASED, STATUS_UNKNOWN

JSONLOG_FNAME = 'unittest.json'


#
# MAIN KEYS
#
MAIN_KEY_COVERAGE_INFO = 'coverage_information'
MAIN_KEY_LOST_SOULS = "lost_souls"
MAIN_KEY_SPECIFICATION = 'specification'
MAIN_KEY_SYSTEM_INFO = 'system_information'
MAIN_KEY_TESTOBJECT_INFO = 'testobject_information'
MAIN_KEY_TESTRUNS = 'testrun_list'
MAIN_KEY_UNITTEST_INFO = 'unittest_information'
#
# SUBKEYS FOR MAIN_KEY_COVERAGE_INFO
#
COVI_KEY_NAME = 'name'
COVI_KEY_FILEPATH = 'filepath'
COVI_KEY_LINE_COVERAGE = 'line_coverage'
COVI_KEY_BRANCH_COVERAGE = 'branch_coverage'
COVI_KEY_FILES = 'files'
#
# SUBKEYS FOR MAIN_KEY_LOST_SOULS
#
LOST_ITEMLIST = "item_list"
LOST_TESTCASELIST = "testcase_list"
#
# SUBKEYS FOR MAIN_KEY_SPECIFICATION
#
SPEC_ITEM_DICT = 'item_dict'
#
# SUBKEYS FOR MAIN_KEY_SYSTEM_INFO
#
SYSI_ARCHITECTURE = "Architecture"
SYSI_DISTRIBUTION = "Distribution"
SYSI_HOSTNAME = "Hostname"
SYSI_KERNEL = "Kernel"
SYSI_MACHINE = "Machine"
SYSI_PATH = "Path"
SYSI_SYSTEM = "System"
SYSI_USERNAME = "Username"
#
# SUBKEYS FOR MAIN_KEY_TESTOBJECT_INFO
#
TOBI_DEPENDENCIES = "Dependencies"
TOBI_DESCRIPTION = "Description"
TOBI_NAME = "Name"
TOBI_STATE = "State"
TOBI_STATE_RELESED = 'Released'
TOBI_STATE_IN_DEVELOPMENT = 'In development'
TOBI_SUPP_INTERP = "Supported Interpreters"
TOBI_VERSION = "Version"
#
# SUBKEYS FOR MAIN_KEY_TESTRUNS
#
TRUN_TESTCASES = 'testcases'
#
# SUBKEYS FOR MAIN_KEY_UNITTEST_INFO
#
UTEI_VERSION = TOBI_VERSION


def get_lib_folder(ut_folder):
    return os.path.join(ut_folder, 'pylibs', os.path.basename(ut_folder))


def get_ut_config(ut_folder):
    return os.path.join(get_ut_src_folder(ut_folder), 'config.py')


def get_ut_testcase_folder(ut_folder):
    return os.path.join(get_ut_src_folder(ut_folder), 'tests')


def get_ut_testresult_folder(ut_folder):
    return os.path.join(get_ut_subfolder(ut_folder), 'testresults')


def get_lib_testresult_folder(ut_folder):
    return os.path.join(get_lib_folder(ut_folder), '_testresults_')


def get_lib_jsonlog(ut_folder):
    return os.path.join(get_lib_testresult_folder(ut_folder), JSONLOG_FNAME)


def get_ut_jsonlog(ut_folder):
    return os.path.join(get_ut_testresult_folder(ut_folder), JSONLOG_FNAME)


def get_ut_src_folder(ut_folder):
    return os.path.join(get_ut_subfolder(ut_folder), 'src')


def get_ut_subfolder(ut_folder):
    return os.path.join(ut_folder, 'unittest')


def module_uid(path):
    return fstools.uid_filelist(path, '*.py', rekursive=True)


def __get_release_state__(ut_folder, lib):
    if lib:
        fn = get_lib_jsonlog(ut_folder)
    else:
        fn = get_ut_jsonlog(ut_folder)
    try:
        with open(fn, 'r') as fh:
            ut_data = json.loads(fh.read())
    except IOError:
        return STATUS_MISSING
    else:
        ut_status = ut_data.get(MAIN_KEY_TESTOBJECT_INFO, {}).get(TOBI_STATE, 'unknown')
        if 'released' in ut_status.lower():
            return STATUS_RELEASED
        elif 'work' in ut_status.lower():
            return STATUS_IN_WORK
        else:
            return STATUS_UNKNOWN


def get_lib_release_state(ut_folder):
    return __get_release_state__(ut_folder, True)


def get_ut_release_state(ut_folder):
    return __get_release_state__(ut_folder, False)


def __get_testcase_integrity__(ut_folder, lib):
    if lib:
        fn = get_lib_jsonlog(ut_folder)
    else:
        fn = get_ut_jsonlog(ut_folder)
    try:
        with open(fn, 'r') as fh:
            ut_data = json.loads(fh.read())
    except IOError:
        return STATUS_MISSING
    else:
        tc_version = ut_data.get(MAIN_KEY_UNITTEST_INFO, {}).get(UTEI_VERSION)
        current_version = module_uid(get_ut_testcase_folder(ut_folder))
        if tc_version == current_version:
            return STATUS_CLEAN
        else:
            return STATUS_CHANGED


def get_lib_testcase_integrity(ut_folder):
    return __get_testcase_integrity__(ut_folder, True)


def get_ut_testcase_integrity(ut_folder):
    return __get_testcase_integrity__(ut_folder, False)


def __get_src_integrity__(ut_folder, lib):
    if lib:
        fn = get_lib_jsonlog(ut_folder)
    else:
        fn = get_ut_jsonlog(ut_folder)
    try:
        with open(fn, 'r') as fh:
            ut_data = json.loads(fh.read())
    except IOError:
        return STATUS_MISSING
    else:
        tested_version = ut_data.get(MAIN_KEY_TESTOBJECT_INFO, {}).get(TOBI_VERSION)
        current_version = module_uid(get_lib_folder(ut_folder))
        if tested_version == current_version:
            return STATUS_CLEAN
        else:
            return STATUS_CHANGED


def get_lib_src_integrity(ut_folder):
    return __get_src_integrity__(ut_folder, True)


def get_ut_src_integrity(ut_folder):
    return __get_src_integrity__(ut_folder, False)


def status_module(ut_folder):
    try:
        with open(get_lib_jsonlog(ut_folder), 'r') as fh:
            ut_lib = json.loads(fh.read())
    except IOError:
        return STATUS_MISSING
    else:
        try:
            with open(get_ut_jsonlog(ut_folder), 'r') as fh:
                ut_ut = json.loads(fh.read())
        except IOError:
            return STATUS_UNKNOWN
        else:
            tested_version = ut_lib.get(MAIN_KEY_TESTOBJECT_INFO, {}).get(TOBI_VERSION)
            current_version = module_uid(get_lib_folder(ut_folder))
            if ut_ut[MAIN_KEY_TESTOBJECT_INFO] != ut_lib[MAIN_KEY_TESTOBJECT_INFO] or ut_ut[MAIN_KEY_UNITTEST_INFO] != ut_lib[MAIN_KEY_UNITTEST_INFO] or tested_version != current_version:
                return STATUS_OLD
            else:
                ut_status = ut_lib.get(MAIN_KEY_TESTOBJECT_INFO, {}).get(TOBI_STATE, 'unknown')
                if 'released' in ut_status.lower():
                    return STATUS_RELEASED
                elif 'work' in ut_status.lower():
                    return STATUS_IN_WORK
                else:
                    return STATUS_UNKNOWN


def versions_module(ut_folder):
    try:
        with open(get_ut_jsonlog(ut_folder), 'r') as fh:
            ut = json.loads(fh.read())
    except IOError:
        return STATUS_UNKNOWN
    else:
        interpreters = ut.get(MAIN_KEY_TESTOBJECT_INFO, {}).get(TOBI_SUPP_INTERP, '')
        interpreters = interpreters.split(',')
        for i in range(len(interpreters)):
            interpreters[i] = interpreters[i].strip()
            interpreters[i] = interpreters[i][6:]
        return ', '.join(interpreters)


def __get_coverage__(ut_folder, lib):
    if lib:
        fn = get_lib_jsonlog(ut_folder)
    else:
        fn = get_ut_jsonlog(ut_folder)
    try:
        with open(fn, 'r') as fh:
            ut = json.loads(fh.read())
    except IOError:
        return None, None
    else:
        lcov = ut.get(MAIN_KEY_COVERAGE_INFO, [{}])[0].get(COVI_KEY_LINE_COVERAGE)
        bcov = ut.get(MAIN_KEY_COVERAGE_INFO, [{}])[0].get(COVI_KEY_BRANCH_COVERAGE)
        return lcov, bcov


def lib_coverage(ut_folder):
    return __get_coverage__(ut_folder, True)


def ut_coverage(ut_folder):
    return __get_coverage__(ut_folder, False)


def status_git(ut_folder):
    p = subprocess.Popen("git -C %s status 2> /dev/null" % ut_folder, stdout=subprocess.PIPE, shell=True)
    output = p.communicate()[0]
    p_status = p.wait()
    if p_status == 0:
        if b"nichts zu committen" in output and b"um lokale Commits zu publizieren" not in output:
            return STATUS_CLEAN
        else:
            return STATUS_CHANGED
    else:
        return STATUS_UNKNOWN


def status_doc(ut_folder):
    if os.path.exists(os.path.join(get_lib_folder(ut_folder), '_docs_', 'index.html')):
        return STATUS_AVAILABLE
    else:
        if os.path.exists(os.path.join(ut_folder, 'docs', 'index.rst')):
            return STATUS_IN_WORK
        else:
            return STATUS_MISSING


def status_spec(ut_folder):
    if os.path.exists(os.path.join(ut_folder, 'requirements', 'specification.reqif')):
        try:
            with open(get_ut_jsonlog(ut_folder), 'r') as fh:
                ut = json.loads(fh.read())
                if len(ut[MAIN_KEY_LOST_SOULS][LOST_ITEMLIST]) > 0 or len(ut[MAIN_KEY_LOST_SOULS][LOST_TESTCASELIST]) > 0:
                    return STATUS_IN_WORK
                else:
                    return STATUS_CLEAN
        except IOError:
            return STATUS_EXISTS
    else:
        return STATUS_MISSING