Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
628 changes: 26 additions & 602 deletions acestep/ui/gradio/events/__init__.py

Large diffs are not rendered by default.

18 changes: 18 additions & 0 deletions acestep/ui/gradio/events/wiring/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,11 +12,22 @@
build_mode_ui_outputs,
)
from .generation_metadata_wiring import register_generation_metadata_handlers
from .generation_metadata_file_wiring import register_generation_metadata_file_handlers
from .generation_batch_navigation_wiring import register_generation_batch_navigation_handlers
from .generation_mode_wiring import register_generation_mode_handlers
from .generation_run_wiring import register_generation_run_handlers
from .results_aux_wiring import register_results_aux_handlers
from .results_display_wiring import (
register_results_restore_and_lrc_handlers,
register_results_save_button_handlers,
)
from .generation_service_wiring import register_generation_service_handlers
from .training_dataset_builder_wiring import register_training_dataset_builder_handlers
from .training_dataset_preprocess_wiring import (
register_training_dataset_load_handler,
register_training_preprocess_handler,
)
from .training_run_wiring import register_training_run_handlers

__all__ = [
"GenerationWiringContext",
Expand All @@ -25,9 +36,16 @@
"build_auto_checkbox_outputs",
"build_mode_ui_outputs",
"register_generation_batch_navigation_handlers",
"register_generation_metadata_file_handlers",
"register_generation_metadata_handlers",
"register_generation_mode_handlers",
"register_generation_run_handlers",
"register_results_aux_handlers",
"register_results_restore_and_lrc_handlers",
"register_results_save_button_handlers",
"register_generation_service_handlers",
"register_training_dataset_builder_handlers",
"register_training_dataset_load_handler",
"register_training_preprocess_handler",
"register_training_run_handlers",
]
18 changes: 18 additions & 0 deletions acestep/ui/gradio/events/wiring/ast_test_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
"""Shared AST parsing helpers for wiring contract tests."""

import ast
from pathlib import Path


def load_module_ast(module_path: Path) -> ast.Module:
"""Return the parsed AST module for the provided source path."""

return ast.parse(module_path.read_text(encoding="utf-8"))


def subscript_key(node: ast.Subscript) -> str | None:
"""Return constant key value from a simple subscript expression."""

if isinstance(node.slice, ast.Constant) and isinstance(node.slice.value, str):
return node.slice.value
return None
Original file line number Diff line number Diff line change
@@ -1,102 +1,71 @@
"""Regression tests for event wiring decomposition contracts.

These tests validate source-level delegation in
``acestep.ui.gradio.events.__init__`` without importing Gradio dependencies.
"""
"""Generation-focused decomposition contract tests."""

import ast
from pathlib import Path
import unittest


_EVENTS_INIT_PATH = Path(__file__).resolve().parents[1] / "__init__.py"
_MODE_WIRING_PATH = Path(__file__).resolve().with_name("generation_mode_wiring.py")
_RUN_WIRING_PATH = Path(__file__).resolve().with_name("generation_run_wiring.py")
_BATCH_NAV_WIRING_PATH = Path(__file__).resolve().with_name(
"generation_batch_navigation_wiring.py"
)


def _load_setup_event_handlers_node() -> ast.FunctionDef:
"""Return the AST node for ``setup_event_handlers``."""

source = _EVENTS_INIT_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "setup_event_handlers":
return node
raise AssertionError("setup_event_handlers not found")


def _load_generation_mode_wiring_node() -> ast.FunctionDef:
"""Return the AST node for ``register_generation_mode_handlers``."""

source = _MODE_WIRING_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "register_generation_mode_handlers":
return node
raise AssertionError("register_generation_mode_handlers not found")


def _load_generation_run_wiring_node() -> ast.FunctionDef:
"""Return the AST node for ``register_generation_run_handlers``."""

source = _RUN_WIRING_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "register_generation_run_handlers":
return node
raise AssertionError("register_generation_run_handlers not found")


def _load_generation_batch_navigation_wiring_node() -> ast.FunctionDef:
"""Return the AST node for ``register_generation_batch_navigation_handlers``."""

source = _BATCH_NAV_WIRING_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "register_generation_batch_navigation_handlers":
return node
raise AssertionError("register_generation_batch_navigation_handlers not found")


def _call_name(node: ast.AST) -> str | None:
"""Extract a simple function name from a call node target."""

if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
return node.attr
return None


class DecompositionContractTests(unittest.TestCase):
"""Verify delegation contracts introduced in PR2/PR3/PR4/PR5/PR6 extraction."""
try:
from .decomposition_contract_helpers import (
call_name,
load_generation_batch_navigation_wiring_node,
load_generation_metadata_file_wiring_module,
load_generation_mode_wiring_node,
load_generation_run_wiring_node,
load_results_display_wiring_module,
load_setup_event_handlers_node,
)
except ImportError: # pragma: no cover - supports direct file execution
from decomposition_contract_helpers import (
call_name,
load_generation_batch_navigation_wiring_node,
load_generation_metadata_file_wiring_module,
load_generation_mode_wiring_node,
load_generation_run_wiring_node,
load_results_display_wiring_module,
load_setup_event_handlers_node,
)


class DecompositionContractGenerationTests(unittest.TestCase):
"""Verify generation-side delegation contracts for event wiring extraction."""

def test_setup_event_handlers_uses_generation_wiring_helpers(self):
"""setup_event_handlers should delegate generation wiring registration."""

setup_node = _load_setup_event_handlers_node()
setup_node = load_setup_event_handlers_node()
call_names = []
for node in ast.walk(setup_node):
if isinstance(node, ast.Call):
name = _call_name(node.func)
name = call_name(node.func)
if name:
call_names.append(name)

self.assertIn("register_generation_service_handlers", call_names)
self.assertIn("register_generation_batch_navigation_handlers", call_names)
self.assertIn("register_generation_metadata_file_handlers", call_names)
self.assertIn("register_generation_metadata_handlers", call_names)
self.assertIn("register_generation_mode_handlers", call_names)
self.assertIn("register_generation_run_handlers", call_names)
self.assertIn("register_results_aux_handlers", call_names)
self.assertIn("register_results_save_button_handlers", call_names)
self.assertIn("register_results_restore_and_lrc_handlers", call_names)
self.assertIn("build_mode_ui_outputs", call_names)

def test_generation_metadata_file_wiring_calls_expected_handlers(self):
"""Metadata file wiring should call load-metadata and auto-uncheck handlers."""

wiring_node = load_generation_metadata_file_wiring_module()
attribute_names = []
for node in ast.walk(wiring_node):
if isinstance(node, ast.Attribute):
attribute_names.append(node.attr)

self.assertIn("load_metadata", attribute_names)
self.assertIn("uncheck_auto_for_populated_fields", attribute_names)

def test_generation_mode_wiring_uses_mode_ui_outputs_variable(self):
"""Mode wiring helper should bind generation_mode outputs to mode_ui_outputs."""

wiring_node = _load_generation_mode_wiring_node()
wiring_node = load_generation_mode_wiring_node()
found_mode_change_output_binding = False

for node in ast.walk(wiring_node):
Expand All @@ -118,12 +87,12 @@ def test_generation_mode_wiring_uses_mode_ui_outputs_variable(self):
def test_generation_run_wiring_calls_expected_results_handlers(self):
"""Run wiring should call clear, generate stream, and background pre-generation helpers."""

wiring_node = _load_generation_run_wiring_node()
wiring_node = load_generation_run_wiring_node()
call_names = []
attribute_names = []
for node in ast.walk(wiring_node):
if isinstance(node, ast.Call):
name = _call_name(node.func)
name = call_name(node.func)
if name:
call_names.append(name)
if isinstance(node, ast.Attribute):
Expand All @@ -136,12 +105,12 @@ def test_generation_run_wiring_calls_expected_results_handlers(self):
def test_batch_navigation_wiring_calls_expected_results_handlers(self):
"""Batch navigation wiring should call previous/next/background results helpers."""

wiring_node = _load_generation_batch_navigation_wiring_node()
wiring_node = load_generation_batch_navigation_wiring_node()
call_names = []
attribute_names = []
for node in ast.walk(wiring_node):
if isinstance(node, ast.Call):
name = _call_name(node.func)
name = call_name(node.func)
if name:
call_names.append(name)
if isinstance(node, ast.Attribute):
Expand All @@ -152,6 +121,18 @@ def test_batch_navigation_wiring_calls_expected_results_handlers(self):
self.assertIn("navigate_to_next_batch", attribute_names)
self.assertIn("generate_next_batch_background", call_names)

def test_results_display_wiring_calls_expected_results_handlers(self):
"""Results display wiring should call restore and LRC subtitle handlers."""

wiring_node = load_results_display_wiring_module()
attribute_names = []
for node in ast.walk(wiring_node):
if isinstance(node, ast.Attribute):
attribute_names.append(node.attr)

self.assertIn("restore_batch_parameters", attribute_names)
self.assertIn("update_audio_subtitles_from_lrc", attribute_names)


if __name__ == "__main__":
unittest.main()
133 changes: 133 additions & 0 deletions acestep/ui/gradio/events/wiring/decomposition_contract_helpers.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,133 @@
"""Shared AST helpers for decomposition contract tests."""

import ast
from pathlib import Path


_EVENTS_INIT_PATH = Path(__file__).resolve().parents[1] / "__init__.py"
_MODE_WIRING_PATH = Path(__file__).resolve().with_name("generation_mode_wiring.py")
_METADATA_FILE_WIRING_PATH = Path(__file__).resolve().with_name(
"generation_metadata_file_wiring.py"
)
_RUN_WIRING_PATH = Path(__file__).resolve().with_name("generation_run_wiring.py")
_BATCH_NAV_WIRING_PATH = Path(__file__).resolve().with_name(
"generation_batch_navigation_wiring.py"
)
_RESULTS_DISPLAY_WIRING_PATH = Path(__file__).resolve().with_name(
"results_display_wiring.py"
)
_TRAINING_DATASET_BUILDER_WIRING_PATH = Path(__file__).resolve().with_name(
"training_dataset_builder_wiring.py"
)
_TRAINING_DATASET_PREPROCESS_WIRING_PATH = Path(__file__).resolve().with_name(
"training_dataset_preprocess_wiring.py"
)
_TRAINING_RUN_WIRING_PATH = Path(__file__).resolve().with_name("training_run_wiring.py")
_TRAINING_LOKR_WIRING_PATH = Path(__file__).resolve().with_name("training_lokr_wiring.py")


def load_setup_event_handlers_node() -> ast.FunctionDef:
"""Return the AST node for ``setup_event_handlers``."""

source = _EVENTS_INIT_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "setup_event_handlers":
return node
raise AssertionError("setup_event_handlers not found")


def load_setup_training_event_handlers_node() -> ast.FunctionDef:
"""Return the AST node for ``setup_training_event_handlers``."""

source = _EVENTS_INIT_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "setup_training_event_handlers":
return node
raise AssertionError("setup_training_event_handlers not found")


def load_generation_mode_wiring_node() -> ast.FunctionDef:
"""Return the AST node for ``register_generation_mode_handlers``."""

source = _MODE_WIRING_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "register_generation_mode_handlers":
return node
raise AssertionError("register_generation_mode_handlers not found")


def load_generation_metadata_file_wiring_module() -> ast.Module:
"""Return the parsed AST module for metadata file-load wiring."""

source = _METADATA_FILE_WIRING_PATH.read_text(encoding="utf-8")
return ast.parse(source)


def load_generation_run_wiring_node() -> ast.FunctionDef:
"""Return the AST node for ``register_generation_run_handlers``."""

source = _RUN_WIRING_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "register_generation_run_handlers":
return node
raise AssertionError("register_generation_run_handlers not found")


def load_generation_batch_navigation_wiring_node() -> ast.FunctionDef:
"""Return the AST node for ``register_generation_batch_navigation_handlers``."""

source = _BATCH_NAV_WIRING_PATH.read_text(encoding="utf-8")
module = ast.parse(source)
for node in module.body:
if isinstance(node, ast.FunctionDef) and node.name == "register_generation_batch_navigation_handlers":
return node
raise AssertionError("register_generation_batch_navigation_handlers not found")


def load_results_display_wiring_module() -> ast.Module:
"""Return the parsed AST module for results display/save wiring."""

source = _RESULTS_DISPLAY_WIRING_PATH.read_text(encoding="utf-8")
return ast.parse(source)


def load_training_run_wiring_module() -> ast.Module:
"""Return the parsed AST module for ``training_run_wiring.py``."""

source = _TRAINING_RUN_WIRING_PATH.read_text(encoding="utf-8")
return ast.parse(source)


def load_training_lokr_wiring_module() -> ast.Module:
"""Return the parsed AST module for ``training_lokr_wiring.py``."""

source = _TRAINING_LOKR_WIRING_PATH.read_text(encoding="utf-8")
return ast.parse(source)


def load_training_dataset_preprocess_wiring_module() -> ast.Module:
"""Return the parsed AST module for training dataset/preprocess wiring."""

source = _TRAINING_DATASET_PREPROCESS_WIRING_PATH.read_text(encoding="utf-8")
return ast.parse(source)


def load_training_dataset_builder_wiring_module() -> ast.Module:
"""Return the parsed AST module for training dataset-builder wiring."""

source = _TRAINING_DATASET_BUILDER_WIRING_PATH.read_text(encoding="utf-8")
return ast.parse(source)


def call_name(node: ast.AST) -> str | None:
"""Extract a simple function name from a call node target."""

if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
return node.attr
return None
Loading