diagnostics.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
# Copyright (C) 2020 Timm Fitschen <t.fitschen@indiscale.com>
# Copyright (C) 2020 IndiScale GmbH <info@indiscale.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
# ** end header
#
"""diagnostics.py

A script which returns a json representation of various parameters which might
be interesting for debugging the server-side scripting functionality and which
should not be executable for non-admin users.
"""

import sys

TEST_MODULES = [
    "linkahead",
    "numpy",
    "pandas",
    "validate_email"
]


def get_files():
    from os import walk
    from os.path import join
    result = []
    for p, dirs, files in walk("."):
        for f in files:
            result.append(join(p, f))
        for d in dirs:
            result.append(join(p, d))
    return result


def get_option(name, default=None):
    for arg in sys.argv:
        if arg.startswith("--{}=".format(name)):
            index = len(name) + 3
            return arg[index:]
    return default


def get_exit_code():
    return int(get_option("exit", 0))


def get_auth_token():
    return get_option("auth-token")


def get_query():
    return get_option("query")


def get_linkahead_info(auth_token):
    import linkahead as db
    result = dict()
    result["version"] = db.version.version

    try:
        db.configure_connection(
            auth_token=auth_token,
            password_method="auth_token")

        info = db.Info()

        result["info"] = str(info)
        result["username"] = info.user_info.name
        result["realm"] = info.user_info.realm
        result["roles"] = info.user_info.roles

        # execute a query and return the results
        query = get_query()
        if query is not None:
            query_result = db.execute_query(query)
            result["query"] = (query, str(query_result))

    except Exception as e:
        result["exception"] = str(e)
    return result


def test_imports(modules):
    result = dict()
    for m in modules:
        try:
            i = __import__(m)
            if hasattr(i, "__version__") and i.__version__ is not None:
                v = i.__version__
            else:
                v = "unknown version"
            result[m] = (True, v)
        except ImportError as e:
            result[m] = (False, str(e))
    return result


def main():
    try:
        import json
    except ImportError:
        print('{{"python_version":"{v}",'
              '"python_path":["{p}"]}}'.format(v=sys.version,
                                               p='","'.join(sys.path)))
        raise

    try:
        diagnostics = dict()
        diagnostics["python_version"] = sys.version
        diagnostics["python_path"] = sys.path
        diagnostics["call"] = sys.argv
        diagnostics["import"] = test_imports(TEST_MODULES)
        diagnostics["files"] = get_files()

        auth_token = get_auth_token()
        diagnostics["auth_token"] = auth_token

        if diagnostics["import"]["linkahead"][0] is True:
            diagnostics["linkahead"] = get_linkahead_info(auth_token)

    finally:
        json.dump(diagnostics, sys.stdout)

    sys.exit(get_exit_code())


if __name__ == "__main__":
    main()