#!/usr/bin/env python3 """ Интерактивный запуск pytest для WESP с человекочитаемым итогом на русском. python3 tests/run_tests.py # меню python3 tests/run_tests.py sync # набор по id python3 tests/run_tests.py --list python3 tests/run_tests.py all --cov """ from __future__ import annotations import argparse import re import subprocess import sys import time from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional, Sequence, Tuple PROJECT_ROOT = Path(__file__).resolve().parent.parent PYTEST_INI = PROJECT_ROOT / "pytest.ini" # Наборы с GPIO/HX711: не входят в «all», pytest.ini их игнорирует — снимаем ignore явно. _HARDWARE_SUITE_IDS = frozenset({"gpio", "hx711"}) # id, заголовок, пути относительно PROJECT_ROOT SUITES: List[Tuple[str, str, List[str]]] = [ ( "all", "Все тесты (без GPIO/HX711)", ["tests"], ), ( "sync", "Синхронизация", [ "tests/test_sync_manager.py", "tests/test_sync_client_api.py", "tests/test_sync_dual_push_roundtrip.py", "tests/test_sync_routes.py", "tests/test_sync_request_parser.py", "tests/test_sync_error_display.py", "tests/test_sync_content_hash.py", "tests/test_recipe_sync_enqueue.py", "tests/test_sync_integration.py", "tests/test_daily_plan_sync.py", "tests/test_setup_guard_config.py", ], ), ( "admin", "Админка и auth", [ "tests/test_admin_panel_access_and_users.py", "tests/test_admin_llm.py", "tests/test_admin_llm_tools.py", "tests/test_admin_network_interfaces.py", "tests/test_admin_peripheral_monitor.py", "tests/test_auth_env_only.py", "tests/test_route_auth_guards.py", "tests/test_additional_route_guards_and_pagination.py", ], ), ( "kiosk", "Киоск и pairing", [ "tests/test_kiosk_pairing.py", "tests/test_kiosk_boot.py", "tests/test_kiosk_full_setup.py", "tests/test_kiosk_page_hints.py", "tests/test_public_device_token_guards.py", "tests/test_scales_routes.py", "tests/test_scales_reader.py", ], ), ( "legacy", "Отчёты и рецепты (legacy compat)", [ "tests/test_legacy_compat_first_batch.py", "tests/test_legacy_compat_second_batch.py", "tests/test_legacy_compat_third_batch.py", "tests/test_legacy_compat_fourth_batch.py", "tests/test_legacy_compat_fifth_batch.py", "tests/test_legacy_compat_sixth_batch.py", "tests/test_route_response_contracts.py", "tests/test_recipe_period_transfer.py", ], ), ( "mill", "Кормоцех (API, sync, UI)", [ "tests/test_feed_mill_api.py", "tests/test_feed_mill_sync.py", "tests/test_feed_mill_ui.py", "tests/test_recipe_sync_enqueue.py", ], ), ( "recipes", "Страница /recipes (контракты + API soft delete/order)", [ "tests/test_recipes_ui_contract.py", "tests/test_recipes_page_served.py", "tests/test_zootech_html_guard.py", "tests/test_feed_mill_ui.py", "tests/test_recipes_soft_delete.py", "tests/test_recipes_order.py", "tests/test_recipes_api_edges.py", "tests/test_recipes_sync_roundtrip.py", "tests/test_recipe_calculator.py", "tests/test_recipes_calculate_api.py", "tests/test_recipe_dry_matter_locked.py", ], ), ( "recipes-e2e", "Страница /recipes (Playwright E2E)", [ "tests/e2e/test_recipes_page.py", "tests/e2e/test_recipes_dispenser.py", "tests/e2e/test_recipes_period_transfer.py", "tests/e2e/test_recipes_settings.py", "tests/e2e/test_recipes_mobile.py", "tests/e2e/test_recipes_user_journeys.py", "tests/e2e/test_recipes_dual_terminal.py", ], ), ( "recipes-full", "Recipes: API + dual sync + E2E (полный прогон)", [ "tests/test_recipes_ui_contract.py", "tests/test_recipes_page_served.py", "tests/test_recipes_soft_delete.py", "tests/test_recipes_order.py", "tests/test_recipes_api_edges.py", "tests/test_recipes_sync_roundtrip.py", "tests/test_recipe_calculator.py", "tests/test_recipes_calculate_api.py", "tests/test_recipe_dry_matter_locked.py", "tests/test_sync_dual_instance.py", "tests/test_sync_dual_recipes_dispenser.py", "tests/test_sync_recipe_children_dual.py", "tests/test_sync_dual_push_roundtrip.py", "tests/test_sync_dual_concurrent_recipe.py", "tests/test_sync_dual_reports.py", "tests/test_sync_dual_offline_catchup.py", "tests/e2e/test_recipes_page.py", "tests/e2e/test_recipes_dispenser.py", "tests/e2e/test_recipes_period_transfer.py", "tests/e2e/test_recipes_settings.py", "tests/e2e/test_recipes_mobile.py", "tests/e2e/test_recipes_user_journeys.py", "tests/e2e/test_recipes_dual_terminal.py", ], ), ( "sync-dual", "Sync: два терминала (term A + term B)", [ "tests/test_sync_dual_instance.py", "tests/test_recipes_sync_roundtrip.py", "tests/test_sync_dual_recipes_dispenser.py", "tests/test_sync_recipe_children_dual.py", "tests/test_sync_dual_push_roundtrip.py", "tests/test_sync_dual_concurrent_recipe.py", "tests/test_sync_dual_reports.py", "tests/test_sync_dual_offline_catchup.py", "tests/e2e/test_daily_plan_operator_sync.py", ], ), ( "lab-p0", "Lab p0: health / скелет", ["tests/test_lab_api.py::LabApiTests::test_health"], ), ( "lab-p1", "Lab p1: миграция + ETL fixtures", [ "tests/test_migrated_schema_matches_models.py", "tests/test_lab_etl_import.py", ], ), ( "lab-p2", "Lab p2: calc engine + golden", [ "tests/test_lab_calc_engine.py", "tests/test_lab_calc_golden.py", "tests/test_lab_calc_daily_totals.py", "tests/test_lab_nutrient_mapping_regression.py", "tests/test_lab_sv_repair.py", "tests/test_recipe_calculator.py", ], ), ( "lab-p3", "Lab p3: commands + daily_plan regression", [ "tests/test_lab_commands.py", "tests/test_lab_profile_resolve.py", "tests/test_norm_catalog_import.py", "tests/test_cleanup_legacy_profiles.py", "tests/test_component_nutrients_save.py", "tests/test_ration_recalc_service.py", "tests/test_lab_math_ration_seed.py", "tests/test_lab_gfe_norms.py", "tests/test_lab_racion_norms.py", "tests/test_lab_demo_nutrients_seed.py", "tests/test_daily_plan_ingredient_weights.py", ], ), ( "lab-p4", "Lab p4: API + apply sync (execution only)", [ "tests/test_lab_api.py", "tests/test_lab_formulate.py", "tests/test_lab_formulate_score.py", "tests/test_lab_racion_norms.py", "tests/test_lab_feed_groups.py", "tests/test_lab_apply_sync_enqueue.py", "tests/test_sync_integration.py", ], ), ( "lab-p5", "Lab p5: golden path + ETL", [ "tests/test_lab_golden_path.py", "tests/test_lab_etl_import.py", "tests/test_lab_agrostar_import.py", "tests/test_lab_import_router.py", ], ), ( "lab-p6", "Lab p6: UI contract + components", [ "tests/test_lab_ui_contract.py", "tests/test_zootech_k_hub_ui_contract.py", "tests/test_component_refactor.py", ], ), ( "lab-p7", "Lab p7: regression WESP", ["tests/test_lab_regression_wesp.py"], ), ( "lab", "Модуль lab: полный набор", [ "tests/test_lab_calc_engine.py", "tests/test_lab_calc_golden.py", "tests/test_lab_nutrient_mapping_regression.py", "tests/test_lab_gfe_norms.py", "tests/test_lab_racion_norms.py", "tests/test_lab_api.py", "tests/test_lab_commands.py", "tests/test_lab_demo_nutrients_seed.py", "tests/test_lab_golden_path.py", "tests/test_lab_ui_contract.py", "tests/test_lab_etl_import.py", "tests/test_lab_apply_sync_enqueue.py", "tests/test_lab_regression_wesp.py", "tests/test_component_refactor.py", ], ), ( "zootech-full", "Zootech-страницы: API + контракты + E2E", [ "tests/test_feed_dispensers_page_served.py", "tests/test_feed_dispensers_ui_contract.py", "tests/test_components_page_served.py", "tests/test_components_ui_contract.py", "tests/test_components_api.py", "tests/test_reports_page_served.py", "tests/test_reports_ui_contract.py", "tests/test_reports_api.py", "tests/test_feed_consumption_page_served.py", "tests/test_feed_consumption_ui_contract.py", "tests/test_sklad_api.py", "tests/test_unloading_page_served.py", "tests/test_unloading_ui_contract.py", "tests/test_unloading_report_flow.py", "tests/test_zootech_html_guard.py", "tests/test_notifications_api.py", "tests/test_feed_quality_rules.py", "tests/test_feed_quality_evaluator.py", "tests/test_feed_quality_api.py", "tests/test_feed_quality_sync.py", "tests/test_feed_quality_settings_store.py", "tests/test_feed_quality_settings_api.py", "tests/test_feed_quality_notify.py", "tests/test_feed_quality_migration.py", "tests/test_analytics_services.py", "tests/test_analytics_api.py", "tests/test_analytics_ui_contract.py", "tests/test_analytics_stock_forecast.py", "tests/test_daily_plan_builder.py", "tests/test_daily_plan_api.py", "tests/test_daily_trip_skip_service.py", "tests/test_daily_trip_skip_api.py", "tests/test_daily_plan_sync.py", "tests/test_daily_plan_part_skip_service.py", "tests/test_daily_plan_part_skip_api.py", "tests/test_zootech_k_hub_ui_contract.py", "tests/test_lab_calc_golden.py", "tests/test_lab_api.py", "tests/e2e/test_feed_dispensers_page.py", "tests/e2e/test_components_page.py", "tests/e2e/test_reports_page.py", "tests/e2e/test_feed_consumption_page.py", "tests/e2e/test_unloading_page.py", "tests/e2e/test_daily_plan_operator_sync.py", ], ), ( "update", "OTA / update", [ "tests/test_update_flow.py", "tests/test_update_verify.py", "tests/test_update_notifier_api.py", "tests/test_update_health_api.py", "tests/test_update_state_store.py", "tests/test_post_update_script.py", "tests/test_auto_update_db.py", "tests/test_auto_update_runtime.py", "tests/test_verify_release_deps.py", "tests/test_materialize_wheelhouse_sdists.py", ], ), ( "install", "Установка / setup wizard", [ "tests/test_setup_wizard.py", "tests/test_fresh_install.py", "tests/test_factory_reset.py", "tests/test_migrated_schema_matches_models.py", ], ), ( "network", "Сеть / Pi / kiosk boot", [ "tests/test_network_settings_api.py", "tests/test_pi_platform_setup.py", "tests/test_pi_boot_config.py", "tests/test_plymouth_theme.py", "tests/test_system_restart_service.py", "tests/test_startup_background.py", "tests/test_startup_gate.py", "tests/test_traceroute_analyze.py", ], ), ( "smoke", "Быстрый smoke", [ "tests/test_setup_guard_config.py", "tests/test_sync_integration.py", "tests/test_auth_env_only.py", "tests/test_sync_manager.py", "tests/test_sync_dual_instance.py::SyncDualInstanceTests::test_recipe_create_reaches_both_terminals", "tests/test_sync_dual_instance.py::SyncDualInstanceTests::test_confirm_waits_for_both_terminals", ], ), ( "misc", "Прочее (UI, exports, install deps)", [ "tests/test_feed_accounting_exports.py", "tests/test_sklad_metrics.py", "tests/test_static_offline_assets.py", "tests/test_zootech_html_guard.py", "tests/test_html_head_injects.py", "tests/test_install_deps_offline.py", ], ), ( "gpio", "Оборудование: GPIO (только явный запуск, не в «all»)", ["tests/test_gpio_controller.py"], ), ( "hx711", "Оборудование: HX711 (только явный запуск, не в «all»)", ["tests/test_hx711_wrapper.py"], ), ( "calculate", "Расчёт рецепта: API + алгоритмы", [ "tests/test_recipes_calculate_api.py", "tests/test_recipe_calculator.py", "tests/test_recipe_calculator_unloading.py", "tests/test_recipe_calculate_pipeline.py", "tests/test_recipe_dry_matter_locked.py", ], ), ] _SUMMARY_RE = re.compile( r"=+\s*(\d+)\s+failed.*?(\d+)\s+passed" r"|(\d+)\s+passed.*?(\d+)\s+failed" r"|(\d+)\s+passed" r"|(\d+)\s+failed", re.IGNORECASE | re.DOTALL, ) _FAILED_LINE = re.compile(r"^FAILED\s+(.+?)\s*$", re.MULTILINE) _ERROR_LINE = re.compile(r"^E\s+(.+?)\s*$", re.MULTILINE) @dataclass class FailureInfo: nodeid: str test_name: str file_name: str reason: str = "" @dataclass class RunReport: title: str duration_sec: float passed: int = 0 failed: int = 0 skipped: int = 0 errors: int = 0 exit_code: int = 0 failures: List[FailureInfo] = field(default_factory=list) raw_tail: str = "" def _suite_by_id(suite_id: str) -> Optional[Tuple[str, str, List[str]]]: key = (suite_id or "").strip().lower() for sid, title, paths in SUITES: if sid == key: return sid, title, paths return None def _list_suites() -> None: print("Доступные наборы тестов WESP:\n") for sid, title, paths in SUITES: print(f" {sid:10} {title} ({len(paths)} файл(ов))") print("\nПример: python3 tests/run_tests.py sync") def _interactive_menu() -> Optional[str]: print("\nWESP — выбор тестов\n") items = [(sid, title) for sid, title, _ in SUITES] for i, (_, title) in enumerate(items, start=1): print(f" {i}) {title}") print(" 0) Выход") try: choice = input("\nНомер: ").strip() except (EOFError, KeyboardInterrupt): print() return None if choice in ("0", "q", "quit", "exit"): return None if choice.isdigit(): idx = int(choice) if 1 <= idx <= len(items): return items[idx - 1][0] # по id if _suite_by_id(choice): return choice.strip().lower() print("Неизвестный выбор.") return _interactive_menu() def _parse_summary(combined: str) -> Tuple[int, int, int]: passed = failed = skipped = 0 for line in combined.splitlines(): line = line.strip() if "passed" in line and " in " in line: m = re.search(r"(\d+)\s+passed", line) if m: passed = int(m.group(1)) m = re.search(r"(\d+)\s+failed", line) if m: failed = int(m.group(1)) m = re.search(r"(\d+)\s+skipped", line) if m: skipped = int(m.group(1)) break return passed, failed, skipped def _parse_failures(combined: str) -> List[FailureInfo]: failures: List[FailureInfo] = [] lines = combined.splitlines() i = 0 while i < len(lines): m = _FAILED_LINE.match(lines[i].strip()) if m: nodeid = m.group(1).strip() reason = "" j = i + 1 while j < len(lines) and j < i + 12: em = _ERROR_LINE.match(lines[j].strip()) if em: reason = em.group(1).strip()[:200] break j += 1 parts = nodeid.split("::") file_name = parts[0].replace("tests/", "") if parts else nodeid test_name = parts[-1] if len(parts) > 1 else nodeid failures.append( FailureInfo( nodeid=nodeid, test_name=test_name, file_name=file_name, reason=reason, ) ) i += 1 return failures def run_pytest( suite_id: str, *, with_cov: bool = False, verbose: bool = False, ) -> RunReport: found = _suite_by_id(suite_id) if not found: raise SystemExit(f"Неизвестный набор: {suite_id!r}") _, title, paths = found cmd = [sys.executable, "-m", "pytest"] if PYTEST_INI.is_file(): cmd.extend(["-c", str(PYTEST_INI)]) if suite_id in _HARDWARE_SUITE_IDS: cmd.extend(["--override-ini", "addopts="]) cmd.extend(paths) if suite_id == "all": cmd.extend(["-m", "not e2e"]) if verbose: cmd.append("-v") else: cmd.append("--tb=short") if with_cov: cmd.extend( [ "--cov=app", "--cov=sync_client", "--cov-report=term-missing:skip-covered", ] ) started = time.monotonic() proc = subprocess.run( cmd, cwd=str(PROJECT_ROOT), capture_output=True, text=True, ) duration = time.monotonic() - started combined = (proc.stdout or "") + "\n" + (proc.stderr or "") passed, failed, skipped = _parse_summary(combined) failures = _parse_failures(combined) if proc.returncode != 0 else [] tail = "" if proc.returncode != 0 and combined.strip(): tail_lines = combined.strip().splitlines()[-30:] tail = "\n".join(tail_lines) return RunReport( title=title, duration_sec=duration, passed=passed, failed=failed, skipped=skipped, exit_code=proc.returncode, failures=failures, raw_tail=tail, ) def _print_report(report: RunReport) -> None: total = report.passed + report.failed + report.skipped ok = report.exit_code == 0 bar = "═" * 40 print(f"\n{bar}") print(f" {report.title} — итог") print(bar) print(f" Время: {report.duration_sec:.1f} с") if total: print(f" Всего: {total}") print(f" Успешно: {report.passed}") print(f" Провалено: {report.failed}") if report.skipped: print(f" Пропущено: {report.skipped}") print() print(f" Статус: {'ПРОЙДЕНО' if ok else 'НЕ ПРОЙДЕНО'}") if report.failures: print("\n Проваленные тесты:") print(" " + "─" * 38) for i, f in enumerate(report.failures[:25], start=1): print(f" {i}. {f.file_name}") print(f" {f.test_name}") if f.reason: print(f" Причина: {f.reason}") if len(report.failures) > 25: print(f" … и ещё {len(report.failures) - 25}") if report.raw_tail and not ok: print("\n Последние строки лога:") print(" " + "─" * 38) for line in report.raw_tail.splitlines(): print(f" {line}") print(f"{bar}\n") def main(argv: Optional[Sequence[str]] = None) -> int: parser = argparse.ArgumentParser(description="Запуск тестов WESP с RU-отчётом") parser.add_argument( "suite", nargs="?", help="id набора: all, sync, admin, kiosk, …", ) parser.add_argument("--list", action="store_true", help="Список наборов") parser.add_argument("--cov", action="store_true", help="pytest-cov (если установлен)") parser.add_argument("-v", "--verbose", action="store_true", help="pytest -v") args = parser.parse_args(argv) if args.list: _list_suites() return 0 suite_id = args.suite if not suite_id: suite_id = _interactive_menu() if not suite_id: return 0 try: report = run_pytest(suite_id, with_cov=args.cov, verbose=args.verbose) except FileNotFoundError: print("Не найден python или pytest.", file=sys.stderr) return 2 _print_report(report) return report.exit_code if __name__ == "__main__": raise SystemExit(main())