Refactor

Author: natehouk

tests/test_cli.py

@@ -7,6 +7,7 @@
 from time import time, sleep
 from rich.console import Console  # import Rich's Console to enable recording
 import re
+import io
 
 # Use the package logger consistently
 logger = get_logger(__name__)
@@ -24,59 +25,36 @@ def setup_config():
 
 runner = CliRunner()
 
-def wait_for_output_stabilize(get_output, timeout=2.0, poll_interval=0.1):
-    """
-    Wait for output from a callable to stabilize.
-    
-    Args:
-        get_output: A callable returning the current output string
-        timeout: Maximum time to wait
-        poll_interval: Time between polls
-    
-    Returns:
-        The stabilized output string.
-    """
-    start_time = time()
-    last_output = get_output()
-    while time() - start_time < timeout:
-        sleep(poll_interval)
-        current_output = get_output()
-        if current_output == last_output:
-            return current_output
-        last_output = current_output
-    logger.warning("Output did not stabilize within timeout period")
-    return last_output
+class AccumulatingStream(io.StringIO):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.accumulated = []
 
-# Automatically patch the cli's Console for testing
-@pytest.fixture(autouse=True)
-def patch_console(monkeypatch):
-    """
-    Monkeypatch the global console in audiokit.cli to use a recording console.
-    """
-    import audiokit.cli as cli  # import the module that created the console
-    # Create a Console that forces terminal behavior and records all output:
-    recording_console = Console(force_terminal=True, record=True)
-    monkeypatch.setattr(cli, "console", recording_console)
-    return recording_console
+    def write(self, s):
+        self.accumulated.append(s)
+        return super().write(s)
+
+    def get_accumulated(self):
+        return ''.join(self.accumulated)
 
 def strip_ansi(text: str) -> str:
     # Regular expression for ANSI escape sequences
     ansi_escape = re.compile(r'\x1B\[[0-?]*[ -/]*[@-~]')
     return ansi_escape.sub('', text)
 
-def test_cli_analyze(sample_audio_path, patch_console):
+def test_cli_analyze(sample_audio_path):
     """Test CLI analyze command"""
     logger.info("Testing CLI analyze command")
 
+    # Create an accumulating stream and override the runner's output stream method.
+    accum_stream = AccumulatingStream()
+    original_method = runner._get_output_stream
+    runner._get_output_stream = lambda: accum_stream
     result = runner.invoke(
         app, ["analyze", str(sample_audio_path)], catch_exceptions=False
     )
-    
-    # Wait for the recorded console output (from the patched Rich Console) to stabilize.
-    final_output = wait_for_output_stabilize(lambda: patch_console.export_text(clear=False))
-    # Fallback to using the captured output from runner.invoke if nothing is recorded.
-    if not final_output:
-        final_output = result.stdout
+    runner._get_output_stream = original_method
+    final_output = accum_stream.get_accumulated()
 
     logger.debug("CLI analyze command rendered output: {}", final_output)
     logger.debug("CLI analyze command exit code: {}", result.exit_code)
@@ -97,9 +75,13 @@ def test_cli_analyze(sample_audio_path, patch_console):
     assert "0.85" in final_output_clean         # Guitar
     assert "0.90" in final_output_clean         # Drums
 
-def test_cli_process(sample_audio_path, tmp_path, patch_console):
+def test_cli_process(sample_audio_path, tmp_path):
     """Test CLI process command"""
     logger.info("Testing CLI process command")
+    # Create an accumulating stream and override the runner's output stream method.
+    accum_stream = AccumulatingStream()
+    original_method = runner._get_output_stream
+    runner._get_output_stream = lambda: accum_stream
     result = runner.invoke(
         app, [
             "process", 
@@ -109,8 +91,8 @@ def test_cli_process(sample_audio_path, tmp_path, patch_console):
         ],
         catch_exceptions=False
     )
-    
-    final_output = wait_for_output_stabilize(lambda: patch_console.export_text(clear=False))
+    runner._get_output_stream = original_method
+    final_output = accum_stream.get_accumulated()
 
     logger.debug("CLI process command rendered output: {}", final_output)
     logger.debug("CLI process command exit code: {}", result.exit_code)