#!/usr/bin/env python3 import argparse import configparser import datetime import os import re import ydb import logging import sys from pathlib import Path from collections import defaultdict # Add the parent directory to the path to import update_mute_issues sys.path.append(os.path.dirname(os.path.abspath(__file__))) from transform_ya_junit import YaMuteCheck from update_mute_issues import ( create_and_add_issue_to_project, generate_github_issue_title_and_body, get_muted_tests_from_issues, close_unmuted_issues, ) # Configure logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') dir = os.path.dirname(__file__) config = configparser.ConfigParser() config_file_path = f"{dir}/../../config/ydb_qa_db.ini" repo_path = f"{dir}/../../../" muted_ya_path = '.github/config/muted_ya.txt' config.read(config_file_path) DATABASE_ENDPOINT = config["QA_DB"]["DATABASE_ENDPOINT"] DATABASE_PATH = config["QA_DB"]["DATABASE_PATH"] # Константы для временных окон mute-логики MUTE_DAYS = 4 UNMUTE_DAYS = 4 DELETE_DAYS = 7 def is_chunk_test(test): # Сначала смотрим на поле is_test_chunk, если оно есть if 'is_test_chunk' in test: return bool(test['is_test_chunk']) # Фоллбек на старую логику по имени return "chunk" in test.get('test_name', '').lower() def get_wildcard_unmute_candidates(aggregated_for_unmute, mute_check, is_unmute_candidate): """ Возвращает список (pattern, debug_string) для wildcard-паттернов, где все chunk'и проходят фильтр на размьют. Если хотя бы один chunk не проходит фильтр, паттерн не попадает в размьют. Debug-строка берётся от любого chunk'а, прошедшего фильтр (обычно первого). Это гарантирует, что если размьючивается паттерн, то размьючиваются все chunk'и, и наоборот. """ wildcard_groups = {} for test in aggregated_for_unmute: if mute_check and mute_check(test.get('suite_folder'), test.get('test_name')): wildcard_pattern = create_test_string(test, use_wildcards=True) if wildcard_pattern not in wildcard_groups: wildcard_groups[wildcard_pattern] = [] wildcard_groups[wildcard_pattern].append(test) result = [] for pattern, chunks in wildcard_groups.items(): passed_chunks = [chunk for chunk in chunks if is_unmute_candidate(chunk, aggregated_for_unmute)] if len(passed_chunks) == len(chunks) and chunks: debug = create_debug_string( passed_chunks[0], period_days=passed_chunks[0].get('period_days'), date_window=passed_chunks[0].get('date_window') ) result.append((pattern, debug)) return result def get_wildcard_delete_candidates(aggregated_for_delete, mute_check, is_delete_candidate): """ Возвращает список (pattern, debug_string) для wildcard-паттернов, где все chunk'и проходят фильтр на delete (удаление mute). Если хотя бы один chunk не проходит фильтр, паттерн не попадает в delete. Debug-строка берётся от любого chunk'а, прошедшего фильтр (обычно первого). Это гарантирует, что если удаляется mute по паттерну, то удаляются все chunk'и, и наоборот. """ wildcard_groups = {} for test in aggregated_for_delete: if mute_check and mute_check(test.get('suite_folder'), test.get('test_name')): wildcard_pattern = create_test_string(test, use_wildcards=True) if wildcard_pattern not in wildcard_groups: wildcard_groups[wildcard_pattern] = [] wildcard_groups[wildcard_pattern].append(test) result = [] for pattern, chunks in wildcard_groups.items(): passed_chunks = [chunk for chunk in chunks if is_delete_candidate(chunk, aggregated_for_delete)] if len(passed_chunks) == len(chunks) and chunks: debug = create_debug_string( passed_chunks[0], period_days=passed_chunks[0].get('period_days'), date_window=passed_chunks[0].get('date_window') ) result.append((pattern, debug)) return result def execute_query(branch='main', build_type='relwithdebinfo', days_window=1): # Получаем today today = datetime.date.today() start_date = today - datetime.timedelta(days=days_window-1) end_date = today query_start_time = datetime.datetime.now() logging.info(f"Executing query for branch='{branch}', build_type='{build_type}', days_window={days_window}") logging.info(f"Date range: {start_date} to {end_date}") query_string = f''' SELECT test_name, suite_folder, full_name, build_type, branch, date_window, pass_count, fail_count, mute_count, skip_count, success_rate, owner, is_muted, state, days_in_state, is_test_chunk FROM `test_results/analytics/tests_monitor` WHERE date_window >= CurrentUtcDate() - 7*Interval("P1D") AND branch = '{branch}' AND build_type = '{build_type}' ''' logging.info(f"SQL Query:\n{query_string}") try: with ydb.Driver( endpoint=DATABASE_ENDPOINT, database=DATABASE_PATH, credentials=ydb.credentials_from_env_variables(), ) as driver: driver.wait(timeout=10, fail_fast=True) logging.info("Successfully connected to YDB") query = ydb.ScanQuery(query_string, {}) table_client = ydb.TableClient(driver, ydb.TableClientSettings()) it = table_client.scan_query(query) results = [] logging.info("Starting to fetch results...") row_count = 0 while True: try: result = next(it) batch_results = result.result_set.rows results.extend(batch_results) row_count += len(batch_results) logging.debug(f"Fetched batch of {len(batch_results)} rows, total: {row_count}") except StopIteration: break query_end_time = datetime.datetime.now() query_duration = query_end_time - query_start_time logging.info(f"Query completed successfully. Total rows returned: {len(results)}") logging.info(f"Query execution time: {query_duration.total_seconds():.2f} seconds") return results except Exception as e: query_end_time = datetime.datetime.now() query_duration = query_end_time - query_start_time logging.error(f"Error executing query: {e}") logging.error(f"Query parameters: branch='{branch}', build_type='{build_type}', days_window={days_window}") logging.error(f"Date range: {start_date} to {end_date}") logging.error(f"Query execution time before error: {query_duration.total_seconds():.2f} seconds") raise def add_lines_to_file(file_path, lines_to_add): try: os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, 'w') as f: f.writelines(lines_to_add) logging.info(f"Lines added to {file_path}") except Exception as e: logging.error(f"Error adding lines to {file_path}: {e}") def aggregate_test_data(all_data, period_days): """ Универсальная функция для агрегации тестовых данных за указанный период """ today = datetime.date.today() base_date = datetime.date(1970, 1, 1) start_date = today - datetime.timedelta(days=period_days-1) start_days = (start_date - base_date).days logging.info(f"Starting aggregation for {period_days} days period...") logging.info(f"Processing {len(all_data)} test records...") # Агрегируем данные по full_name aggregated = {} processed_count = 0 total_count = len(all_data) for test in all_data: processed_count += 1 # Показываем прогресс каждые 1000 записей или каждые 10% if processed_count % 10000 == 0 or processed_count % max(1, total_count // 10) == 0: progress_percent = (processed_count / total_count) * 100 logging.info(f"Aggregation progress: {processed_count}/{total_count} ({progress_percent:.1f}%)") if test.get('date_window', 0) >= start_days: full_name = test.get('full_name') if full_name not in aggregated: aggregated[full_name] = { 'test_name': test.get('test_name'), 'suite_folder': test.get('suite_folder'), 'full_name': full_name, 'build_type': test.get('build_type'), 'branch': test.get('branch'), 'pass_count': 0, 'fail_count': 0, 'mute_count': 0, 'skip_count': 0, 'owner': test.get('owner'), 'is_muted': test.get('is_muted'), 'is_muted_date': test.get('date_window'), # Сохраняем дату для is_muted 'state': test.get('state'), 'state_history': [test.get('state')], # Начинаем историю состояний 'state_dates': [test.get('date_window')], # История дат состояний 'days_in_state': test.get('days_in_state'), 'date_window': test.get('date_window'), 'period_days': period_days, # <--- добавляем сюда 'is_test_chunk': test.get('is_test_chunk', 0), } else: # Добавляем состояние в историю, если оно отличается от последнего current_state = test.get('state') current_date = test.get('date_window') if current_state and (not aggregated[full_name]['state_history'] or aggregated[full_name]['state_history'][-1] != current_state): aggregated[full_name]['state_history'].append(current_state) aggregated[full_name]['state_dates'].append(current_date) # Обновляем is_muted если текущая дата новее if test.get('date_window', 0) > aggregated[full_name].get('is_muted_date', 0): aggregated[full_name]['is_muted'] = test.get('is_muted') aggregated[full_name]['is_muted_date'] = test.get('date_window') # Суммируем статистику aggregated[full_name]['pass_count'] += test.get('pass_count', 0) aggregated[full_name]['fail_count'] += test.get('fail_count', 0) aggregated[full_name]['mute_count'] += test.get('mute_count', 0) aggregated[full_name]['skip_count'] += test.get('skip_count', 0) # Вычисляем success_rate, создаем summary и конкатенируем state для каждого теста date_window_range = f"{start_date.strftime('%Y-%m-%d')}:{today.strftime('%Y-%m-%d')}" for test_data in aggregated.values(): test_data['date_window'] = date_window_range total_runs = test_data['pass_count'] + test_data['fail_count'] + test_data['mute_count'] + test_data['skip_count'] if total_runs > 0: test_data['success_rate'] = round((test_data['pass_count'] / total_runs) * 100, 1) else: test_data['success_rate'] = 0.0 # Создаем summary для тикета (без дублирования state, так как он уже есть в начале строки) test_data['summary'] = f"p-{test_data['pass_count']}, f-{test_data['fail_count']}, m-{test_data['mute_count']}, s-{test_data['skip_count']}, total-{total_runs}" # Конкатенируем state по дням с датами if 'state_history' in test_data and len(test_data['state_history']) > 1: # Создаем строку с состояниями и датами state_with_dates = [] for i, (state, date) in enumerate(zip(test_data['state_history'], test_data['state_dates'])): if date: # Преобразуем дату в читаемый формат date_obj = base_date + datetime.timedelta(days=date) date_str = date_obj.strftime('%m-%d') state_with_dates.append(f"{state}({date_str})") else: state_with_dates.append(state) test_data['state'] = '->'.join(state_with_dates) # Если только одно состояние, оставляем как есть logging.info(f"Aggregation completed: {len(aggregated)} unique tests aggregated from {total_count} records") return list(aggregated.values()) def calculate_success_rate(test): """Вычисляет success rate для теста""" runs = test.get('pass_count', 0) + test.get('fail_count', 0) return round((test.get('pass_count', 0) / runs * 100) if runs > 0 else 0, 1) def create_test_string(test, use_wildcards=False): testsuite = test.get('suite_folder') testcase = test.get('test_name') test_string = f"{testsuite} {testcase}" if use_wildcards: test_string = re.sub(r'\d+/(\d+)\]', r'*/*]', test_string) return test_string def sort_key_without_prefix(line): """Функция для сортировки строк, игнорирующая префиксы +++, ---, xxx""" # Убираем префиксы +++, ---, xxx для сортировки if line.startswith('+++ ') or line.startswith('--- ') or line.startswith('xxx '): return line[4:] # Убираем первые 4 символа (префикс + пробел) return line # Для строк без префикса возвращаем как есть def create_debug_string(test, success_rate=None, period_days=None, date_window=None): """Создает debug строку для теста""" if success_rate is None: success_rate = calculate_success_rate(test) testsuite = test.get('suite_folder') testcase = test.get('test_name') runs = test.get('pass_count', 0) + test.get('fail_count', 0) + test.get('mute_count', 0) debug_string = f"{testsuite} {testcase} # owner {test.get('owner', 'N/A')} success_rate {success_rate}%" # Показываем оба периода: days и диапазон дат, если есть if period_days is None: period_days = test.get('period_days') if date_window is None: date_window = test.get('date_window') if period_days: debug_string += f" (last {period_days} days)" if date_window: debug_string += f" [{date_window}]" state = test.get('state', 'N/A') if is_chunk_test(test): state = f"(chunk)" is_muted = test.get('is_muted', False) mute_state = "muted" if is_muted else "not muted" debug_string += f", p-{test.get('pass_count')}, f-{test.get('fail_count')},m-{test.get('mute_count')}, s-{test.get('skip_count')}, runs-{runs}, mute state: {mute_state}, test state {state}" return debug_string def is_flaky_test(test, aggregated_data): """Проверяет, является ли тест flaky за указанный период""" # Ищем наш тест в агрегированных данных test_data = None for agg_test in aggregated_data: if agg_test['full_name'] == test.get('full_name'): test_data = agg_test break if not test_data: return False # Обновляем данные теста для debug test['pass_count'] = test_data['pass_count'] test['fail_count'] = test_data['fail_count'] test['period_days'] = test_data.get('period_days') test['is_muted'] = test_data.get('is_muted', False) # Не считаем тест flaky, если он уже замьючен if test_data.get('is_muted', False): return False total_runs = test_data['pass_count'] + test_data['fail_count'] return (test_data['fail_count'] >= 2) or (test_data['fail_count'] >= 1 and total_runs <= 10) def is_unmute_candidate(test, aggregated_data): """Проверяет, является ли тест кандидатом на размьют за указанный период""" # Ищем наш тест в агрегированных данных test_data = None for agg_test in aggregated_data: if agg_test['full_name'] == test.get('full_name'): test_data = agg_test break if not test_data: return False # Обновляем данные теста для debug test['pass_count'] = test_data['pass_count'] test['fail_count'] = test_data['fail_count'] test['mute_count'] = test_data['mute_count'] test['period_days'] = test_data.get('period_days') test['is_muted'] = test_data.get('is_muted', False) total_runs = test_data['pass_count'] + test_data['fail_count'] + test_data['mute_count'] total_fails = test_data['fail_count'] + test_data['mute_count'] # Проверяем, что не было состояний no_runs в истории state_history = test_data.get('state', []) if 'no_runs' in state_history: return False result = total_runs > 4 and total_fails == 0 # Добавляем детальное логирование для диагностики if test_data.get('is_muted', False): # Логируем только для замьюченных тестов logging.debug(f"UNMUTE_CHECK: {test.get('full_name')} - runs:{total_runs}, fails:{total_fails}, muted:{test_data.get('is_muted')}, result:{result}") return result def is_delete_candidate(test, aggregated_data): """Проверяет, является ли тест кандидатом на удаление из mute за указанный период""" # Ищем наш тест в агрегированных данных test_data = None for agg_test in aggregated_data: if agg_test['full_name'] == test.get('full_name'): test_data = agg_test break if not test_data: return False # Обновляем данные теста для debug test['pass_count'] = test_data['pass_count'] test['fail_count'] = test_data['fail_count'] test['mute_count'] = test_data['mute_count'] test['skip_count'] = test_data['skip_count'] test['period_days'] = test_data.get('period_days') test['is_muted'] = test_data.get('is_muted', False) total_runs = test_data['pass_count'] + test_data['fail_count'] + test_data['mute_count'] + test_data['skip_count'] result = total_runs == 0 # Добавляем детальное логирование для диагностики if test_data.get('is_muted', False): # Логируем только для замьюченных тестов logging.debug(f"DELETE_CHECK: {test.get('full_name')} - runs:{total_runs}, muted:{test_data.get('is_muted')}, result:{result}") return result def create_file_set(aggregated_for_mute, filter_func, mute_check=None, use_wildcards=False, resolution=None): """Создает набор тестов для файла на основе фильтра""" result_set = set() debug_list = [] total = len(aggregated_for_mute) last_percent = -1 for idx, test in enumerate(aggregated_for_mute, 1): testsuite = test.get('suite_folder') testcase = test.get('test_name') if not testsuite or not testcase: continue # Проверяем mute_check если передан if mute_check and not mute_check(testsuite, testcase): continue # Прогресс-бар percent = int(idx / total * 100) if percent != last_percent and (percent % 5 == 0 or percent == 100): print(f"\r[create_file_set] Progress: {percent}% ({idx}/{total})", end="") last_percent = percent # Применяем фильтр if filter_func(test): test_string = create_test_string(test, use_wildcards) result_set.add(test_string) if resolution: debug_string = create_debug_string( test, period_days=test.get('period_days'), date_window=test.get('date_window') ) debug_list.append(debug_string) # Принудительно показываем 100% если не показали if last_percent != 100: print(f"\r[create_file_set] Progress: 100% ({total}/{total})", end="") print() # Перевод строки после прогресса return sorted(list(result_set)), sorted(debug_list) def write_file_set(file_path, test_set, debug_list=None, sort_without_prefixes=False): # По умолчанию игнорируем префиксы для правильной сортировки if not sort_without_prefixes: sorted_test_set = sorted(test_set) else: sorted_test_set = sorted(test_set, key=sort_key_without_prefix) # Сортируем debug_list если он есть sorted_debug_list = None if debug_list: if not sort_without_prefixes: sorted_debug_list = sorted(debug_list) else: sorted_debug_list = sorted(debug_list, key=sort_key_without_prefix) add_lines_to_file(file_path, [line + '\n' for line in sorted_test_set]) if sorted_debug_list: debug_path = file_path.replace('.txt', '_debug.txt') add_lines_to_file(debug_path, [line + '\n' for line in sorted_debug_list]) logging.info(f"Created {os.path.basename(file_path)} with {len(sorted_test_set)} tests") def apply_and_add_mutes(all_data, output_path, mute_check, aggregated_for_mute, aggregated_for_unmute, aggregated_for_delete): output_path = os.path.join(output_path, 'mute_update') logging.info(f"Creating mute files in directory: {output_path}") # Получаем уникальные тесты для обработки (используем максимальный период для получения всех тестов) # Добавляем статистику по замьюченным тестам muted_tests = [test for test in aggregated_for_mute if test.get('is_muted', False)] logging.info(f"Total muted tests found: {len(muted_tests)}") try: # 1. Кандидаты на mute def is_mute_candidate(test): return is_flaky_test(test, aggregated_for_mute) to_mute, to_mute_debug = create_file_set( aggregated_for_mute, is_mute_candidate, use_wildcards=True, resolution='to_mute' ) write_file_set(os.path.join(output_path, 'to_mute.txt'), to_mute, to_mute_debug) # 2. Кандидаты на размьют def is_unmute_candidate_wrapper(test): if is_chunk_test(test): return False # Не размьючивать chunk поштучно return is_unmute_candidate(test, aggregated_for_unmute) # Обычные тесты на размьют (по chunk'ам) to_unmute, to_unmute_debug = create_file_set( aggregated_for_unmute, is_unmute_candidate_wrapper, mute_check, resolution='to_unmute' ) # Wildcard-паттерны на размьют: # Если все chunk'и паттерна проходят фильтр, то размьючивается весь паттерн (массовый размьют). # Если хотя бы один chunk не проходит — паттерн не размьючивается. wildcard_unmute = get_wildcard_unmute_candidates(aggregated_for_unmute, mute_check, is_unmute_candidate) wildcard_unmute_patterns = [p for p, d in wildcard_unmute] wildcard_unmute_debugs = [d for p, d in wildcard_unmute] # Объединяем поштучные и wildcard-результаты to_unmute = sorted(list(set(to_unmute) | set(wildcard_unmute_patterns))) to_unmute_debug = sorted(list(set(to_unmute_debug) | set(wildcard_unmute_debugs))) write_file_set(os.path.join(output_path, 'to_unmute.txt'), to_unmute, to_unmute_debug) # 3. Кандидаты на удаление из mute (to_delete) def is_delete_candidate_wrapper(test): if is_chunk_test(test): return False # Не удалять chunk поштучно return is_delete_candidate(test, aggregated_for_delete) # Обычные тесты на delete (по chunk'ам) to_delete, to_delete_debug = create_file_set( aggregated_for_delete, is_delete_candidate_wrapper, mute_check, resolution='to_delete' ) # Wildcard-паттерны на delete: # Если все chunk'и паттерна проходят фильтр, то удаляется mute по всему паттерну (массовое удаление). # Если хотя бы один chunk не проходит — паттерн не удаляется. wildcard_delete = get_wildcard_delete_candidates(aggregated_for_delete, mute_check, is_delete_candidate) wildcard_delete_patterns = [p for p, d in wildcard_delete] wildcard_delete_debugs = [d for p, d in wildcard_delete] # Объединяем поштучные и wildcard-результаты to_delete = sorted(list(set(to_delete) | set(wildcard_delete_patterns))) to_delete_debug = sorted(list(set(to_delete_debug) | set(wildcard_delete_debugs))) write_file_set(os.path.join(output_path, 'to_delete.txt'), to_delete, to_delete_debug) # 4. muted_ya (все замьюченные сейчас) all_muted_ya, all_muted_ya_debug = create_file_set( aggregated_for_mute, lambda test: mute_check(test.get('suite_folder'), test.get('test_name')) if mute_check else True, use_wildcards=True, resolution='muted_ya' ) write_file_set(os.path.join(output_path, 'muted_ya.txt'), all_muted_ya, all_muted_ya_debug) to_mute_set = set(to_mute) to_unmute_set = set(to_unmute) to_delete_set = set(to_delete) all_muted_ya_set = set(all_muted_ya) # Создаем словари для быстрого поиска debug-строк # Универсальный словарь: ключ — строка теста (с wildcard или без), значение — debug-строка test_debug_dict = {} wildcard_to_chunks = defaultdict(list) for test in aggregated_for_mute + aggregated_for_unmute + aggregated_for_delete: test_str = create_test_string(test, use_wildcards=False) # без .strip() debug_str = create_debug_string(test) test_debug_dict[test_str] = debug_str if is_chunk_test(test): wildcard_key = create_test_string(test, use_wildcards=True) wildcard_to_chunks[wildcard_key].append(test) # Формируем debug-строку для wildcard for wildcard, chunks in wildcard_to_chunks.items(): N = len(chunks) m = sum(1 for t in chunks if t.get('is_muted')) # Собираем статистику по всем chunks для объяснения решения total_pass = sum(t.get('pass_count', 0) for t in chunks) total_fail = sum(t.get('fail_count', 0) for t in chunks) total_mute = sum(t.get('mute_count', 0) for t in chunks) total_skip = sum(t.get('skip_count', 0) for t in chunks) total_runs = total_pass + total_fail + total_mute + total_skip # Определяем причину попадания в список reason = "" if wildcard in to_mute_set: reason = f"TO_MUTE: {total_fail} fails in {total_runs} runs" elif wildcard in to_unmute_set: reason = f"TO_UNMUTE: {total_fail + total_mute} fails in {total_runs} runs" elif wildcard in to_delete_set: reason = f"TO_DELETE: {total_runs} total runs" debug_str = f"{wildcard}: {N} chunks, {m} muted ({reason})" test_debug_dict[wildcard] = debug_str # 5. muted_ya+to_mute muted_ya_plus_to_mute = list(all_muted_ya) + [t for t in to_mute if t not in all_muted_ya] muted_ya_plus_to_mute_debug = [] for test in muted_ya_plus_to_mute: debug_val = test_debug_dict.get(test, "NO DEBUG INFO") muted_ya_plus_to_mute_debug.append(debug_val) write_file_set(os.path.join(output_path, 'muted_ya+to_mute.txt'), muted_ya_plus_to_mute, muted_ya_plus_to_mute_debug) # 6. muted_ya-to_unmute muted_ya_minus_to_unmute = [t for t in all_muted_ya if t not in to_unmute] muted_ya_minus_to_unmute_debug = [] for test in muted_ya_minus_to_unmute: debug_val = test_debug_dict.get(test, "NO DEBUG INFO") muted_ya_minus_to_unmute_debug.append(debug_val) write_file_set(os.path.join(output_path, 'muted_ya-to_unmute.txt'), muted_ya_minus_to_unmute, muted_ya_minus_to_unmute_debug) # 7. muted_ya-to_delete muted_ya_minus_to_delete = [t for t in all_muted_ya if t not in to_delete] muted_ya_minus_to_delete_debug = [] for test in muted_ya_minus_to_delete: debug_val = test_debug_dict.get(test, "NO DEBUG INFO") muted_ya_minus_to_delete_debug.append(debug_val) write_file_set(os.path.join(output_path, 'muted_ya-to_delete.txt'), muted_ya_minus_to_delete, muted_ya_minus_to_delete_debug) # 8. muted_ya-to-delete-to-unmute muted_ya_minus_to_delete_to_unmute = [t for t in all_muted_ya if t not in to_delete and t not in to_unmute] muted_ya_minus_to_delete_to_unmute_debug = [] for test in muted_ya_minus_to_delete_to_unmute: debug_val = test_debug_dict.get(test, "NO DEBUG INFO") muted_ya_minus_to_delete_to_unmute_debug.append(debug_val) write_file_set(os.path.join(output_path, 'muted_ya-to-delete-to-unmute.txt'), muted_ya_minus_to_delete_to_unmute, muted_ya_minus_to_delete_to_unmute_debug) # 9. muted_ya-to-delete-to-unmute+to_mute # Вместо set используем list для полного соответствия muted_ya_minus_to_delete_to_unmute_plus_to_mute = list(muted_ya_minus_to_delete_to_unmute) + [t for t in to_mute if t not in muted_ya_minus_to_delete_to_unmute] muted_ya_minus_to_delete_to_unmute_plus_to_mute_debug = [] for test in muted_ya_minus_to_delete_to_unmute_plus_to_mute: debug_val = test_debug_dict.get(test, "NO DEBUG INFO") muted_ya_minus_to_delete_to_unmute_plus_to_mute_debug.append(debug_val) write_file_set(os.path.join(output_path, 'muted_ya-to-delete-to-unmute+to_mute.txt'), muted_ya_minus_to_delete_to_unmute_plus_to_mute, muted_ya_minus_to_delete_to_unmute_plus_to_mute_debug) # Сохраняем этот же файл как new_muted_ya.txt для совместимости с workflow write_file_set(os.path.join(output_path, 'new_muted_ya.txt'), muted_ya_minus_to_delete_to_unmute_plus_to_mute, muted_ya_minus_to_delete_to_unmute_plus_to_mute_debug) # 10. muted_ya_changes - файл с изменениями (новая логика) all_test_strings = sorted(all_muted_ya_set | to_mute_set | to_unmute_set | to_delete_set, key=sort_key_without_prefix) # Гарантируем 1:1 соответствие между .txt и debug.txt muted_ya_changes = [] muted_ya_changes_debug = [] for test_str in all_test_strings: # all_test_strings должен быть list, не set if test_str in to_mute_set and test_str not in all_muted_ya_set: prefix = "+++" elif test_str in to_unmute_set: prefix = "---" elif test_str in to_delete_set: prefix = "xxx" else: prefix = "" line = f"{prefix} {test_str}" if prefix else f"{test_str}" muted_ya_changes.append(line) debug_val = test_debug_dict.get(test_str, "NO DEBUG INFO") muted_ya_changes_debug.append(f"{prefix} {debug_val}" if prefix else debug_val) write_file_set(os.path.join(output_path, 'muted_ya_changes.txt'), muted_ya_changes, muted_ya_changes_debug, sort_without_prefixes=True) # Логирование итоговых результатов logging.info(f"To mute: {len(to_mute)}") logging.info(f"To unmute: {len(to_unmute)}") logging.info(f"To delete: {len(to_delete)}") logging.info(f"Muted_ya: {len(all_muted_ya)}") logging.info(f"muted_ya+to_mute: {len(muted_ya_plus_to_mute)}") logging.info(f"muted_ya-to_unmute: {len(muted_ya_minus_to_unmute)}") logging.info(f"muted_ya-to_delete: {len(muted_ya_minus_to_delete)}") logging.info(f"muted_ya-to_delete-to_unmute: {len(muted_ya_minus_to_delete_to_unmute)}") logging.info(f"muted_ya-to_delete-to_unmute+to_mute: {len(muted_ya_minus_to_delete_to_unmute_plus_to_mute)}") except (KeyError, TypeError) as e: logging.error(f"Error processing test data: {e}. Check your query results for valid keys.") return [] return len(to_mute) def read_tests_from_file(file_path): result = [] with open(file_path, "r") as fp: for line in fp: line = line.strip() try: testsuite, testcase = line.split(" ", maxsplit=1) result.append({'testsuite': testsuite, 'testcase': testcase, 'full_name': f"{testsuite}/{testcase}"}) except ValueError: logging.warning(f"cant parse line: {line!r}") continue return result def create_mute_issues(all_tests, file_path, close_issues=True): tests_from_file = read_tests_from_file(file_path) muted_tests_in_issues = get_muted_tests_from_issues() prepared_tests_by_suite = {} temp_tests_by_suite = {} # Create set of muted tests for faster lookup muted_tests_set = {test['full_name'] for test in tests_from_file} # Check and close issues if needed closed_issues = [] partially_unmuted_issues = [] if close_issues: closed_issues, partially_unmuted_issues = close_unmuted_issues(muted_tests_set) # First, collect all tests into temporary dictionary for test in all_tests: for test_from_file in tests_from_file: if test['full_name'] == test_from_file['full_name']: if test['full_name'] in muted_tests_in_issues: logging.info( f"test {test['full_name']} already have issue, {muted_tests_in_issues[test['full_name']][0]['url']}" ) else: key = f"{test_from_file['testsuite']}:{test['owner']}" if not temp_tests_by_suite.get(key): temp_tests_by_suite[key] = [] temp_tests_by_suite[key].append( { 'mute_string': f"{ test.get('suite_folder')} {test.get('test_name')}", 'test_name': test.get('test_name'), 'suite_folder': test.get('suite_folder'), 'full_name': test.get('full_name'), 'success_rate': test.get('success_rate'), 'days_in_state': test.get('days_in_state'), 'date_window': test.get('date_window', 'N/A'), 'owner': test.get('owner'), 'state': test.get('state'), 'summary': test.get('summary'), 'fail_count': test.get('fail_count'), 'pass_count': test.get('pass_count'), 'branch': test.get('branch'), } ) # Split groups larger than 20 tests for key, tests in temp_tests_by_suite.items(): if len(tests) <= 40: prepared_tests_by_suite[key] = tests else: # Split into groups of 40 for i in range(0, len(tests), 40): chunk = tests[i:i+40] chunk_key = f"{key}_{i//40 + 1}" # Add iterator to key starting from 1 prepared_tests_by_suite[chunk_key] = chunk results = [] for item in prepared_tests_by_suite: title, body = generate_github_issue_title_and_body(prepared_tests_by_suite[item]) owner_value = prepared_tests_by_suite[item][0]['owner'].split('/', 1)[1] if '/' in prepared_tests_by_suite[item][0]['owner'] else prepared_tests_by_suite[item][0]['owner'] result = create_and_add_issue_to_project(title, body, state='Muted', owner=owner_value) if not result: break else: results.append( { 'message': f"Created issue '{title}' for TEAM:@ydb-platform/{owner_value}, url {result['issue_url']}", 'owner': owner_value } ) # Sort results by owner results.sort(key=lambda x: x['owner']) # Group results by owner and add spacing and headers formatted_results = [] # Add closed issues section if any if closed_issues: formatted_results.append("🔒 **CLOSED ISSUES**") formatted_results.append("─────────────────────────────") formatted_results.append("") for issue in closed_issues: formatted_results.append(f"✅ **Closed** {issue['url']}") formatted_results.append(" 📝 **Unmuted tests:**") for test in issue['tests']: formatted_results.append(f" • `{test}`") formatted_results.append("") # Add partially unmuted issues section if any if partially_unmuted_issues: if closed_issues: formatted_results.append("─────────────────────────────") formatted_results.append("") formatted_results.append("🔓 **PARTIALLY UNMUTED ISSUES**") formatted_results.append("─────────────────────────────") formatted_results.append("") for issue in partially_unmuted_issues: formatted_results.append(f"⚠️ **Partially unmuted** {issue['url']}") formatted_results.append(" 📝 **Unmuted tests:**") for test in issue['unmuted_tests']: formatted_results.append(f" • `{test}`") formatted_results.append("") # Add created issues section if any if results: if closed_issues or partially_unmuted_issues: formatted_results.append("─────────────────────────────") formatted_results.append("") formatted_results.append("🆕 **CREATED ISSUES**") formatted_results.append("─────────────────────────────") formatted_results.append("") # Add created issues current_owner = None for result in results: if current_owner != result['owner']: if formatted_results and formatted_results[-1] != "": formatted_results.append('') formatted_results.append('') current_owner = result['owner'] # Add owner header with team URL formatted_results.append(f"👥 **TEAM** @ydb-platform/{current_owner}") formatted_results.append(f" https://github.com/orgs/ydb-platform/teams/{current_owner}") formatted_results.append(" ┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄┄") # Extract issue URL and title issue_url = result['message'].split('url ')[-1] title = result['message'].split("'")[1] formatted_results.append(f" 🎯 {issue_url} - `{title}`") print("\n\n") print("\n".join(formatted_results)) if 'GITHUB_OUTPUT' in os.environ: if 'GITHUB_WORKSPACE' not in os.environ: raise EnvironmentError("GITHUB_WORKSPACE environment variable is not set.") file_path = os.path.join(os.environ['GITHUB_WORKSPACE'], "created_issues.txt") print(f"Writing results to {file_path}") with open(file_path, 'w') as f: f.write("\n") f.write("\n".join(formatted_results)) f.write("\n") with open(os.environ['GITHUB_OUTPUT'], 'a') as gh_out: gh_out.write(f"created_issues_file={file_path}") print(f"Result saved to env variable GITHUB_OUTPUT by key created_issues_file") def mute_worker(args): # Simplified Connection if "CI_YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS" not in os.environ: print("Error: Env variable CI_YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS is missing, skipping") return 1 else: # Do not set up 'real' variable from gh workflows because it interfere with ydb tests # So, set up it locally os.environ["YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS"] = os.environ[ "CI_YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS" ] logging.info(f"Starting mute worker with mode: {args.mode}") logging.info(f"Branch: {args.branch}") mute_check = YaMuteCheck() mute_check.load(muted_ya_path) logging.info(f"Loaded muted_ya.txt with {len(mute_check.regexps)} test patterns") logging.info("Executing single query for 7 days window...") # Один запрос за максимальный период (7 дней) all_data = execute_query(args.branch, days_window=7) logging.info(f"Query returned {len(all_data)} test records") # Используем универсальную агрегацию для разных периодов aggregated_for_mute = aggregate_test_data(all_data, MUTE_DAYS) # MUTE_DAYS дней для mute aggregated_for_unmute = aggregate_test_data(all_data, UNMUTE_DAYS) # UNMUTE_DAYS дней для unmute aggregated_for_delete = aggregate_test_data(all_data, DELETE_DAYS) # DELETE_DAYS дней для delete logging.info(f"Aggregated data: mute={len(aggregated_for_mute)}, unmute={len(aggregated_for_unmute)}, delete={len(aggregated_for_delete)}") if args.mode == 'update_muted_ya': output_path = args.output_folder os.makedirs(output_path, exist_ok=True) logging.info(f"Creating mute files in: {output_path}") apply_and_add_mutes(all_data, output_path, mute_check, aggregated_for_mute, aggregated_for_unmute, aggregated_for_delete) elif args.mode == 'create_issues': file_path = args.file_path logging.info(f"Creating issues from file: {file_path}") # Используем уже агрегированные данные за 3 дня create_mute_issues(aggregated_for_mute, file_path, close_issues=args.close_issues) logging.info("Mute worker completed successfully") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Add tests to mutes files based on flaky_today condition") subparsers = parser.add_subparsers(dest='mode', help="Mode to perform") update_muted_ya_parser = subparsers.add_parser('update_muted_ya', help='create new muted_ya') update_muted_ya_parser.add_argument('--output_folder', default=repo_path, required=False, help='Output folder.') update_muted_ya_parser.add_argument('--branch', default='main', help='Branch to get history') create_issues_parser = subparsers.add_parser( 'create_issues', help='create issues by muted_ya like files', ) create_issues_parser.add_argument( '--file_path', default=f'{repo_path}/mute_update/to_mute.txt', required=False, help='file path' ) create_issues_parser.add_argument('--branch', default='main', help='Branch to get history') create_issues_parser.add_argument('--close_issues', action='store_true', default=True, help='Close issues when all tests are unmuted (default: True)') args = parser.parse_args() mute_worker(args)