作りかけの何か

create_template.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
""" Module docstring. FIXME
"""
import argparse
import json
from logging import getLogger,StreamHandler,Formatter,DEBUG,INFO
from os import path
import sys

__author__ = 'Kiyoshi Mizumaru'
__email__ = '[email protected]'

logger = getLogger(__name__)
formatter = Formatter('%(asctime)s %(levelname)s $(message)s')
handler = StreamHandler()
handler.setLevel(INFO)
handler.setFormatter(formatter)
logger.setLevel(INFO)
logger.addHandler(handler)


class JsonHash(object):
    """

    """
    def __init__(self, base_dir="."):
        """
        Create a JsonHash instance
        :param base_dir: base directory for input files
        """
        if path.isdir(base_dir):
            self.base_dir = base_dir
        else:
            raise ValueError("base_dir must be a directory")
        self.json_hash = {}

    def get(self, filename):
        """
        Get an object stored as JSON file
        :param filename: basestring
        :return: dictionary
        """
        if self.json_hash.has_key(filename):
            return self.json_hash[filename]
        else:
            path = "{0}/{1}".format(self.base_dir, filename)
            try:
                with open(path, 'r') as f:
                    res = json.load(f)
            except IOError as (errno, strerror):
                logger.debug("IOError: [Errno {0}] {1}: '{2}'".format(errno, strerror, path))
                raise KeyError(filename)
            self.json_hash[filename] = res
            return res


class TemplateCreator(object):
    """

    """
    content_array = [
        ("cm_version", "gf_cm_version"),
        ("cdh_version", "gf_cdh_version"),
        ("cm_host", "gf_cm_host"),
        ("cluster_os_version", None),
        ("customer", None),
        ("customer_shortname", None),
        ("engagement", None),
        ("consultant", None),
        ("sow_date", None),
        ("sow", None),
        ("environment", None),
        ("cluster_name", None),
        ("engagement_date", None),
        ("time_server", None),
        ("dns_server", None),
        ("db_host", None),
        ("db_type", None),
        ("java_version", None),
        ("edh_placement", None),
        ("core_switches", None),
        ("tor_switches", None),
        ("master_node", None),
        ("utility_node", None),
        ("data_node", None),
        ("kafka_node", None),
        ("gateway_node", None),
        ("network_specification", None),
        ("database", None),
        ("zookeeper_config", None),
        ("hdfs_config", None),
        ("yarn_config", None),
        ("impala_config", None),
        ("spark_config", None),
        ("hbase_config", None),
        ("solr_config", None),
        ("oozie_config", None),
        ("hue_config", None),
        ("kafka_config", None),
        ("flume_config", None),
        ("hive_config", None),
        ("sentry_config", None),
        ("teragen_terasort_timings", None),
        ("ldap_cm", None),
        ("ldap_navigator", None),
        ("ldap_hue", None),
        ("LDAP", None),
        ("tls_cm", None),
        ("tls_hue", None),
        ("tls_navigator", None),
        ("tls_hdfs", None),
        ("tls_httpfs", None),
        ("tls_yarn", None),
        ("tls_oozie", None),
        ("kts_kms_hosts", None),
        ("encryption_zones", None),
        ("sentry_ad_groups", None),
    ] # content_array = [
    # def load_jsonfiles(self):
    #     """
    #     Load JSON files generated by get_json.sh script
    #     :return:
    #     """
    #     def load_jsonfile(filename):
    #         with open(filename, 'r') as f:
    #             res = json.load(f)
    #         return res
    #
    #     tmp = load_jsonfile("{0}/cm_config.json".format(self.in_path))
    #     print(json.dumps(tmp, indent=2))
    #     # 'cm_deployment.json'
    #     # 'cm_service_config_mgmt*.json'
    #     # 'host_*.json'
    #     # 'service_config_*.json'

    def __init__(self, in_path, output=None, force=False):
        """
        Create a TemplateCreator instance
        :param in_path: input directory (output directory of get_json.sh script)
        :param output: output filename
        :param force: overwrite existing output file or not
        """
        if path.isdir(in_path):
            self.in_path = in_path
        else:
            raise ValueError("in_path must be a directory")
        # Python2 doesn't support typing like Union[basestring, None]
        if isinstance(output, basestring) or output is None:
            self.output = output
        else:
            raise TypeError("output must be a string")
        if isinstance(force, bool):
            self.force_output = force
        else:
            raise TypeError("force must be a boolean value")
        self.json_hash = JsonHash(self.in_path)

    def gf_cm_version(self):
        """
        Generator for "cm_version"
        :return: Cloudera Manager version string
        """
        cm_deployment = self.json_hash.get('cm_deployment.json')
        return cm_deployment['versionInfo']['version']

    def gf_cdh_version(self):
        """
        Generator for "cdh_version"
        :return: CDH version string
        """
        cm_deployment = self.json_hash.get('cm_deployment.json')
        if len(cm_deployment['clusters']) == 1:
            index = 0
        else:
            # Collect names from clusters
            name_list = [cluster['displayName'] for cluster in cm_deployment['clusters']]
            clusters = ", ".join(["{0}) {1}".format(i, c) for i, c in enumerate(name_list)])
            while True:
                try:
                    index = int(raw_input("Select a cluster {0}: ".format(clusters)))
                    if 0 <= index < len(cm_deployment['clusters']):
                        break
                    else:
                        print 'Input out of range, please type again'
                except ValueError:
                    print 'Invalid input, please type again'
        return cm_deployment['clusters'][index]['fullVersion']

    def gf_cm_host(self):
        """
        Generator for "cm_host"
        :return: Hostname which runs Cloudera Manager
        """
        return "NOT IMPLEMENTED, HOW CAN I GET THE HOSTNAME?"

    def run(self):
        scratch = {"default_content": None}
        content = {}
        for key, gf in self.content_array:
            if gf is not None:
                content[key] = getattr(self, gf)()
        scratch["default_content"] = content

        print(json.dumps(scratch, indent=2))


def main(argv=None):
    if argv is None:
        argv = sys.argv[1:]

    parser = argparse.ArgumentParser(description='Create a JSON template')
    parser.add_argument('-i', '--input', metavar='input_dir', required=True,
                        help='input directory name')
    parser.add_argument('-o', '--output', metavar='json_file',
                        help='output filename')
    parser.add_argument('-f', '--force', action='store_true', default=False,
                        help='overwrite existing output file')
    args = parser.parse_args(argv)

    app = TemplateCreator(args.input, args.output, args.force)
    app.run()


if __name__ == "__main__":
    sys.exit(main())

test_create_template.py

# -*- coding: utf-8 -*-
"""
UnitTests for create_template module
"""
import unittest
from create_template import JsonHash


class MyTestCase(unittest.TestCase):
    def test_jsonhash(self):
        json_hash = JsonHash()
        with self.assertRaises(KeyError):
            json_hash.get('should_not_exist_in_current_directory.json')
        dict_valid = json_hash.get('test.json')
        self.assertEqual(dict_valid['default_content']['cm_version'], '5.12.0')


if __name__ == '__main__':
    unittest.main()