2gis__k8s-handle-120

{
  "status": "ok",
  "stdout": "./k8s_handle/__init__.py:133:parser_target_config.add_argument('--skip-tags', action='append', required=False,\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "k8s_handle/__init__.py",
  "content": "    for resource in resources:\n        provisioner.run(resource)\n\n\nparser = argparse.ArgumentParser(description='CLI utility generate k8s resources by templates and apply it to cluster')\nsubparsers = parser.add_subparsers(dest=\"command\")\nsubparsers.required = True\n\nparser_target_config = argparse.ArgumentParser(add_help=False)\nparser_target_config.add_argument('-s', '--section', required=True, type=str, help='Section to deploy from config file')\nparser_target_config.add_argument('-c', '--config', required=False, help='Config file, default: config.yaml')\nparser_target_config.add_argument('--tags', action='append', required=False,\n                                  help='Only use templates tagged with these values')\nparser_target_config.add_argument('--skip-tags', action='append', required=False,\n                                  help='Only use templates whose tags do not match these values')\n\nparser_target_resource = argparse.ArgumentParser(add_help=False)\nparser_target_resource.add_argument('-r', '--resource', required=True, type=str,\n                                    help='Resource spec path, absolute (started with slash) or relative from TEMP_DIR')\n\nparser_deprecated = argparse.ArgumentParser(add_help=False)\nparser_deprecated.add_argument('--dry-run', required=False, action='store_true',\n                               help='Don\\'t run kubectl commands. Deprecated, use \"k8s-handle template\" instead')\n\nparser_provisioning = argparse.ArgumentParser(add_help=False)\nparser_provisioning.add_argument('--sync-mode', action='store_true', required=False, default=False,\n                                 help='Turn on sync mode and wait deployment ending')\nparser_provisioning.add_argument('--tries', type=int, required=False, default=360,\n                                 help='Count of tries to check deployment status')\nparser_provisioning.add_argument('--retry-delay', type=int, required=False, default=5,\n                                 help='Sleep between tries in seconds')\nparser_provisioning.add_argument('--strict', action='store_true', required=False,\n                                 help='Check existence of all env variables in config.yaml and stop if var is not set')\nparser_provisioning.add_argument('--use-kubeconfig', action='store_true', required=False,\n                                 help='Try to use kube config')\nparser_provisioning.add_argument('--k8s-handle-debug', action='store_true', required=False,\n                                 help='Show K8S client debug messages')\n\nparser_logs = argparse.ArgumentParser(add_help=False)\nparser_logs.add_argument('--show-logs', action='store_true', required=False, default=False, help='Show logs for jobs')\nparser_logs.add_argument('--tail-lines', type=int, required=False, help='Lines of recent log file to display')\n\narguments_connection = parser_provisioning.add_argument_group()\narguments_connection.add_argument('--k8s-master-uri', required=False, help='K8S master to connect to')\narguments_connection.add_argument('--k8s-ca-base64', required=False, help='base64-encoded K8S certificate authority')\narguments_connection.add_argument('--k8s-token', required=False, help='K8S token to use')\n\nparser_deploy = subparsers.add_parser(\n    'deploy',\n    parents=[parser_provisioning, parser_target_config, parser_logs, parser_deprecated],\n    help='Do attempt to create specs from templates and deploy K8S resources of the selected section')\nparser_deploy.set_defaults(func=handler_deploy)\n\nparser_apply = subparsers.add_parser('apply', parents=[parser_provisioning, parser_target_resource, parser_logs],\n                                     help='Do attempt to deploy K8S resource from the existing spec')\nparser_apply.set_defaults(func=handler_apply)\n\nparser_destroy = subparsers.add_parser('destroy',\n                                       parents=[parser_provisioning, parser_target_config, parser_deprecated],\n                                       help='Do attempt to destroy K8S resources of the selected section')\nparser_destroy.set_defaults(func=handler_destroy)\n\nparser_delete = subparsers.add_parser('delete', parents=[parser_provisioning, parser_target_resource],\n                                      help='Do attempt to destroy K8S resource from the existing spec')\nparser_delete.set_defaults(func=handler_delete)\n\nparser_template = subparsers.add_parser('render', parents=[parser_target_config],\n                                        help='Make resources from the template and config. '\n                                             'Created resources will be placed into the TEMP_DIR')\nparser_template.set_defaults(func=handler_render)\n\n\ndef main():\n    # INFO furiousassault: backward compatibility rough attempt\n    # must be removed later according to https://github.com/2gis/k8s-handle/issues/40\n    deprecation_warnings = 0\n    filtered_arguments = []\n\n    for argument in sys.argv[1:]:\n        if argument in ['--sync-mode=true', '--sync-mode=True', '--dry-run=true', '--dry-run=True']:\n            deprecation_warnings += 1",
  "start_line": 120,
  "end_line": 200,
  "total_lines": 256,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "k8s_handle/__init__.py",
  "content": "parser_destroy = subparsers.add_parser('destroy',\n                                       parents=[parser_provisioning, parser_target_config, parser_deprecated],\n                                       help='Do attempt to destroy K8S resources of the selected section')\nparser_destroy.set_defaults(func=handler_destroy)\n\nparser_delete = subparsers.add_parser('delete', parents=[parser_provisioning, parser_target_resource],\n                                      help='Do attempt to destroy K8S resource from the existing spec')\nparser_delete.set_defaults(func=handler_delete)\n\nparser_template = subparsers.add_parser('render', parents=[parser_target_config],\n                                        help='Make resources from the template and config. '\n                                             'Created resources will be placed into the TEMP_DIR')\nparser_template.set_defaults(func=handler_render)\n\n\ndef main():\n    # INFO furiousassault: backward compatibility rough attempt\n    # must be removed later according to https://github.com/2gis/k8s-handle/issues/40\n    deprecation_warnings = 0\n    filtered_arguments = []\n\n    for argument in sys.argv[1:]:\n        if argument in ['--sync-mode=true', '--sync-mode=True', '--dry-run=true', '--dry-run=True']:\n            deprecation_warnings += 1\n            filtered_arguments.append(argument.split('=')[0])\n            continue\n\n        if argument in ['--sync-mode=false', '--sync-mode=False', '--dry-run=false', '--dry-run=False']:\n            deprecation_warnings += 1\n            continue\n\n        filtered_arguments.append(argument)\n\n    args, unrecognized_args = parser.parse_known_args(filtered_arguments)\n\n    if deprecation_warnings or unrecognized_args:\n        log.warning(\"Explicit true/false arguments to --sync-mode and --dry-run keys are deprecated \"\n                    \"and will be discontinued in the future. Use these keys without arguments instead.\")\n\n    args_dict = vars(args)\n    settings.CHECK_STATUS_TRIES = args_dict.get('tries')\n    settings.CHECK_DAEMONSET_STATUS_TRIES = args_dict.get('tries')\n    settings.CHECK_STATUS_TIMEOUT = args_dict.get('retry_delay')\n    settings.CHECK_DAEMONSET_STATUS_TIMEOUT = args_dict.get('retry_delay')\n    settings.GET_ENVIRON_STRICT = args_dict.get('strict')\n    settings.COUNT_LOG_LINES = args_dict.get('tail_lines')\n    settings.CONFIG_FILE = args_dict.get('config') or settings.CONFIG_FILE\n\n    try:\n        args.func(args_dict)\n    except templating.TemplateRenderingError as e:\n        log.error('Template generation error: {}'.format(e))\n        sys.exit(1)\n    except InvalidYamlError as e:\n        log.error('{}'.format(e))\n        sys.exit(1)\n    except DeprecationError as e:\n        log.warning('Deprecation warning: {}'.format(e))\n    except RuntimeError as e:\n        log.error('RuntimeError: {}'.format(e))\n        sys.exit(1)\n    except ResourceNotAvailableError as e:\n        log.error('Resource not available: {}'.format(e))\n        sys.exit(1)\n    except ProvisioningError:\n        sys.exit(1)\n\n    print(r'''\n                         _(_)_                          wWWWw   _\n             @@@@       (_)@(_)   vVVVv     _     @@@@  (___) _(_)_\n            @@()@@ wWWWw  (_)\\    (___)   _(_)_  @@()@@   Y  (_)@(_)\n             @@@@  (___)     `|/    Y    (_)@(_)  @@@@   \\|/   (_)\n              /      Y       \\|    \\|/    /(_)    \\|      |/      |\n           \\ |     \\ |/       | / \\ | /  \\|/       |/    \\|      \\|/\n            \\|//    \\|///    \\|//  \\|/// \\|///    \\|//    |//    \\|//\n       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^''')\n\n\nif __name__ == '__main__':\n    main()",
  "start_line": 177,
  "end_line": 256,
  "total_lines": 256,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "43:def handler_render(args):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "k8s_handle/__init__.py",
  "content": "    _handler_apply_delete(args, COMMAND_DESTROY)\n\n\ndef handler_render(args):\n    context = config.load_context_section(args.get('section'))\n    templating.Renderer(\n        settings.TEMPLATES_DIR,\n        args.get('tags'),\n        args.get('skip_tags')\n    ).generate_by_context(context)\n\n\ndef _handler_deploy_destroy(args, command):\n    context = config.load_context_section(args.get('section'))\n    resources = templating.Renderer(\n        settings.TEMPLATES_DIR,\n        args.get('tags'),\n        args.get('skip_tags')\n    ).generate_by_context(context)\n\n    if args.get('dry_run'):\n        return\n\n    _handler_provision(\n        command,\n        resources,\n        config.PriorityEvaluator(args, context, os.environ),\n        args.get('use_kubeconfig'),\n        args.get('sync_mode'),\n        args.get('show_logs')\n    )\n\n\ndef _handler_apply_delete(args, command):\n    _handler_provision(\n        command,\n        [os.path.join(settings.TEMP_DIR, args.get('resource'))],\n        config.PriorityEvaluator(args, {}, os.environ),\n        args.get('use_kubeconfig'),\n        args.get('sync_mode'),\n        args.get('show_logs')\n    )\n\n\ndef _handler_provision(command, resources, priority_evaluator, use_kubeconfig, sync_mode, show_logs):\n    kubeconfig_namespace = None\n\n    if priority_evaluator.environment_deprecated():\n        log.warning(\"K8S_HOST and K8S_CA environment variables support is deprecated \"\n                    \"and will be discontinued in the future. Use K8S_MASTER_URI and K8S_CA_BASE64 instead.\")\n\n    # INFO rvadim: https://github.com/kubernetes-client/python/issues/430#issuecomment-359483997\n    if use_kubeconfig:\n        try:\n            load_kube_config()\n            kubeconfig_namespace = list_kube_config_contexts()[1].get('context').get('namespace')\n        except Exception as e:\n            raise RuntimeError(e)\n    else:\n        client.Configuration.set_default(priority_evaluator.k8s_client_configuration())\n",
  "start_line": 40,
  "end_line": 100,
  "total_lines": 256,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "k8s_handle/templating.py",
  "content": "import base64\nimport glob\nimport itertools\nimport logging\nimport os\nimport re\nfrom hashlib import sha256\n\nimport yaml\nfrom jinja2 import Environment, FileSystemLoader, StrictUndefined\nfrom jinja2.exceptions import TemplateNotFound, TemplateSyntaxError, UndefinedError\n\nfrom k8s_handle import settings\nfrom k8s_handle.exceptions import TemplateRenderingError\n\nlog = logging.getLogger(__name__)\n\n\ndef get_template_contexts(file_path):\n    try:\n        with open(file_path) as f:\n            try:\n                contexts = yaml.safe_load_all(f.read())\n            except Exception as e:\n                raise RuntimeError('Unable to load yaml file: {}, {}'.format(file_path, e))\n\n            for context in contexts:\n                if context is None:\n                    continue  # Skip empty YAML documents\n                if 'kind' not in context or context['kind'] is None:\n                    raise RuntimeError('Field \"kind\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'metadata' not in context or context['metadata'] is None:\n                    raise RuntimeError('Field \"metadata\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'name' not in context['metadata'] or context['metadata']['name'] is None:\n                    raise RuntimeError('Field \"metadata->name\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'spec' in context:\n                    # INFO: Set replicas = 1 by default for replaces cases in Deployment and StatefulSet\n                    if 'replicas' not in context['spec'] or context['spec']['replicas'] is None:\n                        if context['kind'] in ['Deployment', 'StatefulSet']:\n                            context['spec']['replicas'] = 1\n                yield context\n    except FileNotFoundError as e:\n        raise RuntimeError(e)\n\n\ndef b64decode(string):\n    res = base64.decodebytes(string.encode())\n    return res.decode()\n\n\ndef b64encode(string):\n    res = base64.b64encode(string.encode())\n    return res.decode()\n\n\ndef hash_sha256(string):\n    res = sha256()\n    res.update(string.encode('utf-8'))\n    return res.hexdigest()\n\n\ndef to_yaml(data, flow_style=True, width=99999):\n    return yaml.safe_dump(data, default_flow_style=flow_style, width=width)\n\n\ndef get_env(templates_dir):\n    # https://stackoverflow.com/questions/9767585/insert-static-files-literally-into-jinja-templates-without-parsing-them\n    def include_file(path):\n        path = os.path.join(templates_dir, '../', path)\n        output = []\n        for file_path in glob.glob(path):\n            with open(file_path, 'r') as f:\n                output.append(f.read())\n        return '\\n'.join(output)\n\n    env = Environment(\n        undefined=StrictUndefined,\n        loader=FileSystemLoader([templates_dir]))\n\n    env.filters['b64decode'] = b64decode\n    env.filters['b64encode'] = b64encode\n    env.filters['hash_sha256'] = hash_sha256\n    env.filters['to_yaml'] = to_yaml\n    env.globals['include_file'] = include_file\n\n    log.debug('Available templates in path {}: {}'.format(templates_dir, env.list_templates()))\n    return env\n\n\nclass Renderer:\n    def __init__(self, templates_dir, tags=None, tags_skip=None):\n        self._templates_dir = templates_dir\n        self._tags = tags\n        self._tags_skip = tags_skip\n        self._env = get_env(self._templates_dir)\n\n    def _iterate_entries(self, entries, tags=None):\n        if tags is None:\n            tags = set()\n\n        for entry in entries:\n            entry[\"tags\"] = self._get_template_tags(entry).union(tags)\n\n            if \"group\" not in entry.keys():\n                if not self._evaluate_tags(entry.get(\"tags\"), self._tags, self._tags_skip):\n                    continue\n                yield entry\n\n            for nested_entry in self._iterate_entries(entry.get(\"group\", []), entry.get(\"tags\")):\n                yield nested_entry\n\n    def _preprocess_templates(self, templates):\n        output = []\n        for template in templates:\n            tags = template.get('tags', [])\n            new_templates = []\n            try:\n                regex = re.compile(template.get('template'))\n                new_templates = list(\n                    map(lambda x: {'template': x, 'tags': tags}, filter(regex.search, self._env.list_templates())))\n            except Exception as e:\n                log.warning(f'Exception during preprocess {template}, {e}, passing it as is')\n\n            if len(new_templates) == 0:\n                output.append(template)\n            else:\n                output += new_templates\n        return output\n\n    def generate_by_context(self, context):\n        if context is None:\n            raise RuntimeError('Can\\'t generate templates from None context')\n\n        templates = self._preprocess_templates(context.get('templates', []))\n        if len(templates) == 0:\n            templates = context.get('kubectl', [])\n            if len(templates) == 0:\n                return\n\n        output = []\n        for template in self._iterate_entries(templates):\n            try:\n                path = self._generate_file(template, settings.TEMP_DIR, context)\n                log.info('File \"{}\" successfully generated'.format(path))\n                output.append(path)\n            except TemplateNotFound as e:\n                raise TemplateRenderingError(\n                    \"Processing {}: template {} hasn't been found\".format(template['template'], e.name))\n            except (UndefinedError, TemplateSyntaxError) as e:\n                raise TemplateRenderingError('Unable to render {}, due to: {}'.format(template, e))",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 201,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "k8s_handle/templating.py",
  "content": "import base64\nimport glob\nimport itertools\nimport logging\nimport os\nimport re\nfrom hashlib import sha256\n\nimport yaml\nfrom jinja2 import Environment, FileSystemLoader, StrictUndefined\nfrom jinja2.exceptions import TemplateNotFound, TemplateSyntaxError, UndefinedError\n\nfrom k8s_handle import settings\nfrom k8s_handle.exceptions import TemplateRenderingError\n\nlog = logging.getLogger(__name__)\n\n\ndef get_template_contexts(file_path):\n    try:\n        with open(file_path) as f:\n            try:\n                contexts = yaml.safe_load_all(f.read())\n            except Exception as e:\n                raise RuntimeError('Unable to load yaml file: {}, {}'.format(file_path, e))\n\n            for context in contexts:\n                if context is None:\n                    continue  # Skip empty YAML documents\n                if 'kind' not in context or context['kind'] is None:\n                    raise RuntimeError('Field \"kind\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'metadata' not in context or context['metadata'] is None:\n                    raise RuntimeError('Field \"metadata\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'name' not in context['metadata'] or context['metadata']['name'] is None:\n                    raise RuntimeError('Field \"metadata->name\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'spec' in context:\n                    # INFO: Set replicas = 1 by default for replaces cases in Deployment and StatefulSet\n                    if 'replicas' not in context['spec'] or context['spec']['replicas'] is None:\n                        if context['kind'] in ['Deployment', 'StatefulSet']:\n                            context['spec']['replicas'] = 1\n                yield context\n    except FileNotFoundError as e:\n        raise RuntimeError(e)\n\n\ndef b64decode(string):\n    res = base64.decodebytes(string.encode())\n    return res.decode()\n\n\ndef b64encode(string):\n    res = base64.b64encode(string.encode())\n    return res.decode()\n\n\ndef hash_sha256(string):\n    res = sha256()\n    res.update(string.encode('utf-8'))\n    return res.hexdigest()\n\n\ndef to_yaml(data, flow_style=True, width=99999):\n    return yaml.safe_dump(data, default_flow_style=flow_style, width=width)\n\n\ndef get_env(templates_dir):\n    # https://stackoverflow.com/questions/9767585/insert-static-files-literally-into-jinja-templates-without-parsing-them\n    def include_file(path):\n        path = os.path.join(templates_dir, '../', path)\n        output = []\n        for file_path in glob.glob(path):\n            with open(file_path, 'r') as f:\n                output.append(f.read())\n        return '\\n'.join(output)\n\n    env = Environment(\n        undefined=StrictUndefined,\n        loader=FileSystemLoader([templates_dir]))\n\n    env.filters['b64decode'] = b64decode\n    env.filters['b64encode'] = b64encode\n    env.filters['hash_sha256'] = hash_sha256\n    env.filters['to_yaml'] = to_yaml\n    env.globals['include_file'] = include_file\n\n    log.debug('Available templates in path {}: {}'.format(templates_dir, env.list_templates()))\n    return env\n\n\nclass Renderer:\n    def __init__(self, templates_dir, tags=None, tags_skip=None):\n        self._templates_dir = templates_dir\n        self._tags = tags\n        self._tags_skip = tags_skip\n        self._env = get_env(self._templates_dir)\n\n    def _iterate_entries(self, entries, tags=None):\n        if tags is None:\n            tags = set()\n\n        for entry in entries:\n            entry[\"tags\"] = self._get_template_tags(entry).union(tags)\n\n            if \"group\" not in entry.keys():\n                if not self._evaluate_tags(entry.get(\"tags\"), self._tags, self._tags_skip):\n                    continue\n                yield entry\n\n            for nested_entry in self._iterate_entries(entry.get(\"group\", []), entry.get(\"tags\")):\n                yield nested_entry\n\n    def _preprocess_templates(self, templates):\n        output = []\n        for template in templates:\n            tags = template.get('tags', [])\n            new_templates = []\n            try:\n                regex = re.compile(template.get('template'))\n                new_templates = list(\n                    map(lambda x: {'template': x, 'tags': tags}, filter(regex.search, self._env.list_templates())))\n            except Exception as e:\n                log.warning(f'Exception during preprocess {template}, {e}, passing it as is')\n\n            if len(new_templates) == 0:\n                output.append(template)\n            else:\n                output += new_templates\n        return output\n\n    def generate_by_context(self, context):\n        if context is None:\n            raise RuntimeError('Can\\'t generate templates from None context')\n\n        templates = self._preprocess_templates(context.get('templates', []))\n        if len(templates) == 0:\n            templates = context.get('kubectl', [])\n            if len(templates) == 0:\n                return\n\n        output = []\n        for template in self._iterate_entries(templates):\n            try:\n                path = self._generate_file(template, settings.TEMP_DIR, context)\n                log.info('File \"{}\" successfully generated'.format(path))\n                output.append(path)\n            except TemplateNotFound as e:\n                raise TemplateRenderingError(\n                    \"Processing {}: template {} hasn't been found\".format(template['template'], e.name))\n            except (UndefinedError, TemplateSyntaxError) as e:\n                raise TemplateRenderingError('Unable to render {}, due to: {}'.format(template, e))",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 201,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "k8s_handle/templating.py",
  "content": "        return output\n\n    def _generate_file(self, item, directory, context):\n        try:\n            log.info('Trying to generate file from template \"{}\" in \"{}\"'.format(item['template'], directory))\n            template = self._env.get_template(item['template'])\n        except TemplateNotFound as e:\n            log.info('Templates path: {}, available templates: {}'.format(self._templates_dir,\n                                                                          self._env.list_templates()))\n            raise e\n        except KeyError:\n            raise RuntimeError('Templates section doesn\\'t have any template items')\n\n        new_name = item['template'].replace('.j2', '')\n        path = os.path.join(directory, new_name)\n\n        try:\n            if not os.path.exists(os.path.dirname(path)):\n                os.makedirs(os.path.dirname(path))\n\n            with open(path, 'w+') as f:\n                f.write(template.render(context))\n\n        except TemplateRenderingError:\n            raise\n        except (FileNotFoundError, PermissionError) as e:\n            raise RuntimeError(e)\n\n        return path\n\n    @staticmethod\n    def _get_template_tags(template):\n        if 'tags' not in template:\n            return set()\n\n        tags = template['tags']\n\n        if isinstance(tags, list):\n            return set([i for i, _ in itertools.groupby(tags)])\n\n        if isinstance(tags, str):\n            return set(tags.split(','))\n\n        raise TypeError('Unable to parse tags of \"{}\" template: unexpected type {}'.format(template, type(tags)))\n\n    @staticmethod\n    def _evaluate_tags(tags, only_tags, skip_tags):\n        if only_tags is None and skip_tags is None:\n            return True\n\n        return tags.isdisjoint(skip_tags or []) and not tags.isdisjoint(only_tags or tags)",
  "start_line": 151,
  "end_line": 201,
  "total_lines": 201,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "k8s_handle/settings.py",
  "content": "import os\n\nLOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO')\nLOG_FORMAT = '%(asctime)s %(levelname)s:%(name)s:%(message)s'\nLOG_DATE_FORMAT = '%Y-%m-%d %H:%M:%S'\n\nCONFIG_FILE = os.environ.get('CONFIG_FILE', 'config.yaml')\n\nCOMMON_SECTION_NAME = 'common'\nTEMPLATES_DIR = os.environ.get('TEMPLATES_DIR', 'templates')\n\nK8S_CONFIG_DIR = os.environ.get('K8S_CONFIG_DIR', '{}/.kube/'.format(os.path.expanduser('~')))\n\nK8S_NAMESPACE = None\n\nTEMP_DIR = os.environ.get('TEMP_DIR', '/tmp/k8s-handle')\n\nCHECK_STATUS_TRIES = 360\nCHECK_STATUS_TIMEOUT = 5\n\nCHECK_CONTAINERS_IN_POD_TRIES = 360\nCHECK_CONTAINERS_IN_POD_TIMEOUT = 5\n\nCHECK_POD_STATUS_TRIES = 360\nCHECK_POD_STATUS_TIMEOUT = 5\n\nCHECK_DAEMONSET_STATUS_TRIES = 10\nCHECK_DAEMONSET_STATUS_TIMEOUT = 5\n\nCOUNT_LOG_LINES = None\n\nGET_ENVIRON_STRICT = False",
  "start_line": 1,
  "end_line": 32,
  "total_lines": 32,
  "is_truncated": false
}

{
  "status": "ok",
  "stdout": "Case 1 (no tags, skip manual): Failed\nCase 2 (tag manual, skip manual): Passed\nCase 3 (tag other, skip manual): Passed\nCase 4 (no tags, only web): Passed\nCase 5 (tag web, only web): Passed\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "k8s_handle/templating.py",
  "content": "import base64\nimport glob\nimport itertools\nimport logging\nimport os\nimport re\nfrom hashlib import sha256\n\nimport yaml\nfrom jinja2 import Environment, FileSystemLoader, StrictUndefined\nfrom jinja2.exceptions import TemplateNotFound, TemplateSyntaxError, UndefinedError\n\nfrom k8s_handle import settings\nfrom k8s_handle.exceptions import TemplateRenderingError\n\nlog = logging.getLogger(__name__)\n\n\ndef get_template_contexts(file_path):\n    try:\n        with open(file_path) as f:\n            try:\n                contexts = yaml.safe_load_all(f.read())\n            except Exception as e:\n                raise RuntimeError('Unable to load yaml file: {}, {}'.format(file_path, e))\n\n            for context in contexts:\n                if context is None:\n                    continue  # Skip empty YAML documents\n                if 'kind' not in context or context['kind'] is None:\n                    raise RuntimeError('Field \"kind\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'metadata' not in context or context['metadata'] is None:\n                    raise RuntimeError('Field \"metadata\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'name' not in context['metadata'] or context['metadata']['name'] is None:\n                    raise RuntimeError('Field \"metadata->name\" not found (or empty) in file \"{}\"'.format(file_path))\n                if 'spec' in context:\n                    # INFO: Set replicas = 1 by default for replaces cases in Deployment and StatefulSet\n                    if 'replicas' not in context['spec'] or context['spec']['replicas'] is None:\n                        if context['kind'] in ['Deployment', 'StatefulSet']:\n                            context['spec']['replicas'] = 1\n                yield context\n    except FileNotFoundError as e:\n        raise RuntimeError(e)\n\n\ndef b64decode(string):\n    res = base64.decodebytes(string.encode())\n    return res.decode()\n\n\ndef b64encode(string):\n    res = base64.b64encode(string.encode())\n    return res.decode()\n\n\ndef hash_sha256(string):\n    res = sha256()\n    res.update(string.encode('utf-8'))\n    return res.hexdigest()\n\n\ndef to_yaml(data, flow_style=True, width=99999):\n    return yaml.safe_dump(data, default_flow_style=flow_style, width=width)\n\n\ndef get_env(templates_dir):\n    # https://stackoverflow.com/questions/9767585/insert-static-files-literally-into-jinja-templates-without-parsing-them\n    def include_file(path):\n        path = os.path.join(templates_dir, '../', path)\n        output = []\n        for file_path in glob.glob(path):\n            with open(file_path, 'r') as f:\n                output.append(f.read())\n        return '\\n'.join(output)\n\n    env = Environment(\n        undefined=StrictUndefined,\n        loader=FileSystemLoader([templates_dir]))\n\n    env.filters['b64decode'] = b64decode\n    env.filters['b64encode'] = b64encode\n    env.filters['hash_sha256'] = hash_sha256\n    env.filters['to_yaml'] = to_yaml\n    env.globals['include_file'] = include_file\n\n    log.debug('Available templates in path {}: {}'.format(templates_dir, env.list_templates()))\n    return env\n\n\nclass Renderer:\n    def __init__(self, templates_dir, tags=None, tags_skip=None):\n        self._templates_dir = templates_dir\n        self._tags = tags\n        self._tags_skip = tags_skip\n        self._env = get_env(self._templates_dir)\n\n    def _iterate_entries(self, entries, tags=None):\n        if tags is None:\n            tags = set()\n\n        for entry in entries:\n            entry[\"tags\"] = self._get_template_tags(entry).union(tags)\n\n            if \"group\" not in entry.keys():\n                if not self._evaluate_tags(entry.get(\"tags\"), self._tags, self._tags_skip):\n                    continue\n                yield entry\n\n            for nested_entry in self._iterate_entries(entry.get(\"group\", []), entry.get(\"tags\")):\n                yield nested_entry\n\n    def _preprocess_templates(self, templates):\n        output = []\n        for template in templates:\n            tags = template.get('tags', [])\n            new_templates = []\n            try:\n                regex = re.compile(template.get('template'))\n                new_templates = list(\n                    map(lambda x: {'template': x, 'tags': tags}, filter(regex.search, self._env.list_templates())))\n            except Exception as e:\n                log.warning(f'Exception during preprocess {template}, {e}, passing it as is')\n\n            if len(new_templates) == 0:\n                output.append(template)\n            else:\n                output += new_templates\n        return output\n\n    def generate_by_context(self, context):\n        if context is None:\n            raise RuntimeError('Can\\'t generate templates from None context')\n\n        templates = self._preprocess_templates(context.get('templates', []))\n        if len(templates) == 0:\n            templates = context.get('kubectl', [])\n            if len(templates) == 0:\n                return\n\n        output = []\n        for template in self._iterate_entries(templates):\n            try:\n                path = self._generate_file(template, settings.TEMP_DIR, context)\n                log.info('File \"{}\" successfully generated'.format(path))\n                output.append(path)\n            except TemplateNotFound as e:\n                raise TemplateRenderingError(\n                    \"Processing {}: template {} hasn't been found\".format(template['template'], e.name))\n            except (UndefinedError, TemplateSyntaxError) as e:\n                raise TemplateRenderingError('Unable to render {}, due to: {}'.format(template, e))",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 201,
  "is_truncated": true
}

{
  "status": "ok",
  "filepath": "k8s_handle/templating.py",
  "content": "        return output\n\n    def _generate_file(self, item, directory, context):\n        try:\n            log.info('Trying to generate file from template \"{}\" in \"{}\"'.format(item['template'], directory))\n            template = self._env.get_template(item['template'])\n        except TemplateNotFound as e:\n            log.info('Templates path: {}, available templates: {}'.format(self._templates_dir,\n                                                                          self._env.list_templates()))\n            raise e\n        except KeyError:\n            raise RuntimeError('Templates section doesn\\'t have any template items')\n\n        new_name = item['template'].replace('.j2', '')\n        path = os.path.join(directory, new_name)\n\n        try:\n            if not os.path.exists(os.path.dirname(path)):\n                os.makedirs(os.path.dirname(path))\n\n            with open(path, 'w+') as f:\n                f.write(template.render(context))\n\n        except TemplateRenderingError:\n            raise\n        except (FileNotFoundError, PermissionError) as e:\n            raise RuntimeError(e)\n\n        return path\n\n    @staticmethod\n    def _get_template_tags(template):\n        if 'tags' not in template:\n            return set()\n\n        tags = template['tags']\n\n        if isinstance(tags, list):\n            return set([i for i, _ in itertools.groupby(tags)])\n\n        if isinstance(tags, str):\n            return set(tags.split(','))\n\n        raise TypeError('Unable to parse tags of \"{}\" template: unexpected type {}'.format(template, type(tags)))\n\n    @staticmethod\n    def _evaluate_tags(tags, only_tags, skip_tags):\n        if only_tags is None and skip_tags is None:\n            return True\n\n        return tags.isdisjoint(skip_tags or []) and not tags.isdisjoint(only_tags or tags)",
  "start_line": 151,
  "end_line": 201,
  "total_lines": 201,
  "is_truncated": false
}