|
import re |
|
import os |
|
import yaml |
|
from typing import Optional |
|
from smolagents import Tool |
|
from jinja2 import Environment, FileSystemLoader |
|
from pydash import py_ |
|
|
|
|
|
class GenerateHelmTest(Tool): |
|
name = "generate_helm_tests" |
|
description = "Performs a duckduckgo web search based on your query (think a Google search) then returns the top search results." |
|
inputs = {'document': {'type': 'string', 'description': 'A YAML string containing one or more Kubernetes resource definitions.'}} |
|
output_type = "string" |
|
|
|
SEP = ''' |
|
|
|
--- |
|
|
|
''' |
|
|
|
@staticmethod |
|
def get_yaml_value(d: dict[str], xpath: str, default: Optional[str] = None) -> str: |
|
try: |
|
if '.' in xpath: |
|
h = xpath.split('.')[0] |
|
t = '.'.join(xpath.split('.')[1:]) |
|
return GenerateHelmTest.get_yaml_value(d[h], t) |
|
else: |
|
return d[xpath] |
|
except Exception as ex: |
|
if default: |
|
return default |
|
else: |
|
raise ex |
|
|
|
|
|
@staticmethod |
|
def bool_to_str(value: bool) -> str: |
|
return str(value).lower() |
|
|
|
|
|
@staticmethod |
|
def get_object_by_kind(document: dict[str], kinds: tuple[str]) -> list[dict[str]]: |
|
return [*py_.filter_(document, lambda x: x['kind'] in kinds)] |
|
|
|
@staticmethod |
|
def get_envs_by_ref(container: dict[str], ref: str, name: str) -> list[dict[str]]: |
|
return [*py_.filter(container['env'], lambda x: x.get('valueFrom', {f'{ref}': {}}).get(f'{ref}', {'name': ''}).get('name') == name)] |
|
|
|
@staticmethod |
|
def get_binding_subjects(subjects: dict[str], name: str) -> list[dict[str]]: |
|
return [*py_.filter(subjects, lambda x: x['name'] == name)] |
|
|
|
def __init__(self, *args, **kwargs): |
|
self.is_initialized = False |
|
|
|
def forward(self, document: str) -> str: |
|
"""Generate Helm test scenarios from a Kubernetes YAML document. |
|
|
|
Args: |
|
document (str): A YAML string containing one or more Kubernetes resource definitions. |
|
|
|
Returns: |
|
str: A string containing the generated test scenarios, separated by a predefined |
|
separator when multiple resources are processed. |
|
""" |
|
doc = [*yaml.safe_load_all(document)] |
|
result = '' |
|
current_dir = os.path.dirname(os.path.abspath(__file__)) |
|
template_dir = os.path.join(current_dir, 'templates') |
|
|
|
for item in doc: |
|
name = item['metadata']['name'] |
|
kind = item['kind'] |
|
kindDesc = (re.sub(r'((?<=[a-z])[A-Z]|(?<!\A)[A-Z](?=[a-z]))', r' \1', kind)).lower() |
|
env = Environment(loader=FileSystemLoader(template_dir)) |
|
env.globals['gyv'] = GenerateHelmTest.get_yaml_value |
|
env.globals['bool_to_str'] = GenerateHelmTest.bool_to_str |
|
env.globals['get_object_by_kind'] = GenerateHelmTest.get_object_by_kind |
|
env.globals['get_env_by_ref'] = GenerateHelmTest.get_envs_by_ref |
|
env.globals['get_binding_subjects'] = GenerateHelmTest.get_binding_subjects |
|
template = env.get_template('scenario.jinja') |
|
output = template.render(name=name, app='app', kind=kind, kindDesc=kindDesc, item=item, document=doc) |
|
if result: |
|
result += GenerateHelmTest.SEP |
|
result += output |
|
return f'```yaml\n{result}\n```' |
|
|
