#!/usr/bin/env python3 import os import yaml import html import random import string from clickhouse_driver import Client client = Client(host = 'localhost', port = 9000) settings = {'default_table_engine': 'Memory', 'union_default_mode': 'DISTINCT', 'calculate_text_stack_trace': 0} database_name = 'sqltest_' + ''.join(random.choice(string.ascii_lowercase) for _ in range(10)) client.execute(f"DROP DATABASE IF EXISTS {database_name}", settings = settings) client.execute(f"CREATE DATABASE {database_name}", settings = settings) client = Client(host = 'localhost', port = 9000, database = database_name) summary = {'success': 0, 'total': 0, 'results': {}} log_file = open("test.log", "w") report_html_file = open("report.html", "w") with open('features.yml', 'r') as file: yaml_content = yaml.safe_load(file) for category in yaml_content: log_file.write(category.capitalize() + " features:\n") summary['results'][category] = {'success': 0, 'total': 0, 'results': {}} for test in yaml_content[category]: log_file.write(test + ": " + yaml_content[category][test] + "\n") summary['results'][category]['results'][test] = {'success': 0, 'total': 0, 'description': yaml_content[category][test]} test_path = test[0] + "/" + test + ".tests.yml" if os.path.exists(test_path): with open(test_path, 'r') as test_file: test_yaml_content = yaml.load_all(test_file, Loader = yaml.FullLoader) for test_case in test_yaml_content: queries = test_case['sql']; if not isinstance(queries, list): queries = [queries] for query in queries: # Example: E011-01 test_group = '' if '-' in test: test_group = test.split("-", 1)[0] summary['results'][category]['results'][test_group]['total'] += 1 summary['results'][category]['results'][test]['total'] += 1 summary['results'][category]['total'] += 1 summary['total'] += 1 log_file.write(query + "\n") try: result = client.execute(query, settings = settings) log_file.write(result + "\n") if test_group: summary['results'][category]['results'][test_group]['success'] += 1 summary['results'][category]['results'][test]['success'] += 1 summary['results'][category]['success'] += 1 summary['success'] += 1 except Exception as e: log_file.write(f"Error occurred: {str(e)}\n") client.execute(f"DROP DATABASE {database_name}", settings = settings) def enable_color(ratio): if ratio == 0: return "" elif ratio < 0.5: return "" elif ratio < 1: return "" else: return "" reset_color = "" def print_ratio(indent, name, success, total, description): report_html_file.write("{}{}: {}{} / {} ({:.1%}){}{}
\n".format( ' ' * indent, name, enable_color(success / total), success, total, success / total, reset_color, f" - " + html.escape(description) if description else '')) report_html_file.write("
\n")

print_ratio(0, 'Total', summary['success'], summary['total'], '')

for category in summary['results']:
    cat_summary = summary['results'][category]

    if cat_summary['total'] == 0:
        continue

    print_ratio(2, category, cat_summary['success'], cat_summary['total'], '')

    for test in summary['results'][category]['results']:
        test_summary = summary['results'][category]['results'][test]

        if test_summary['total'] == 0:
            continue

        print_ratio(4, test, test_summary['success'], test_summary['total'], test_summary['description'])

report_html_file.write("
\n")