update toggle commands

add toggle record/toggle stream tests

pre-release patch bump

obsws_cli/__about__.py

@@ -1,4 +1,4 @@
 # SPDX-FileCopyrightText: 2025-present onyx-and-iris <code@onyxandiris.online>
 #
 # SPDX-License-Identifier: MIT
-__version__ = "0.10.2"
+__version__ = "0.10.3a0"

obsws_cli/record.py

@@ -46,6 +46,16 @@ def stop(ctx: typer.Context):
     typer.echo('Recording stopped successfully.')
 
 
+@app.command('toggle | tg')
+def toggle(ctx: typer.Context):
+    """Toggle recording."""
+    resp = ctx.obj.toggle_record()
+    if resp.output_active:
+        type.echo('Recording started successfully.')
+    else:
+        type.echo('Recording stopped successfully.')
+
+
 @app.command('status | ss')
 def status(ctx: typer.Context):
     """Get recording status."""
@@ -59,16 +69,6 @@ def status(ctx: typer.Context):
         typer.echo('Recording is not in progress.')
 
 
-@app.command('toggle | tg')
-def toggle(ctx: typer.Context):
-    """Toggle recording."""
-    active, _ = _get_recording_status(ctx)
-    if active:
-        ctx.invoke(stop, ctx=ctx)
-    else:
-        ctx.invoke(start, ctx=ctx)
-
-
 @app.command('resume | r')
 def resume(ctx: typer.Context):
     """Resume recording."""

obsws_cli/replaybuffer.py

@@ -26,6 +26,16 @@ def stop(ctx: typer.Context):
     typer.echo('Replay buffer stopped.')
 
 
+@app.command('toggle | tg')
+def toggle(ctx: typer.Context):
+    """Toggle the replay buffer."""
+    resp = ctx.obj.toggle_replay_buffer()
+    if resp.output_active:
+        typer.echo('Replay buffer is active.')
+    else:
+        typer.echo('Replay buffer is not active.')
+
+
 @app.command('status | ss')
 def status(ctx: typer.Context):
     """Get the status of the replay buffer."""

obsws_cli/stream.py

@@ -42,6 +42,16 @@ def stop(ctx: typer.Context):
     typer.echo('Streaming stopped successfully.')
 
 
+@app.command('toggle | tg')
+def toggle(ctx: typer.Context):
+    """Toggle streaming."""
+    resp = ctx.obj.toggle_stream()
+    if resp.output_active:
+        typer.echo('Streaming started successfully.')
+    else:
+        typer.echo('Streaming stopped successfully.')
+
+
 @app.command('status | ss')
 def status(ctx: typer.Context):
     """Get streaming status."""
@@ -64,13 +74,3 @@ def status(ctx: typer.Context):
             typer.echo('Streaming is in progress.')
     else:
         typer.echo('Streaming is not in progress.')
-
-
-@app.command('toggle | tg')
-def toggle(ctx: typer.Context):
-    """Toggle streaming."""
-    active, _ = _get_streaming_status(ctx)
-    if active:
-        ctx.invoke(stop, ctx=ctx)
-    else:
-        ctx.invoke(start, ctx=ctx)

tests/conftest.py

@@ -95,6 +95,10 @@ def pytest_sessionfinish(session, exitstatus):
     if resp.output_active:
         session.obsws.stop_stream()
 
+    resp = session.obsws.get_record_status()
+    if resp.output_active:
+        session.obsws.stop_record()
+
     # Close the OBS WebSocket client connection
     session.obsws.disconnect()
 

tests/test_record.py

@@ -30,3 +30,18 @@ def test_record_start_status_stop():
     result = runner.invoke(app, ['record', 'status'])
     assert result.exit_code == 0
     assert 'Recording is not in progress.' in result.stdout
+
+
+def test_record_toggle():
+    """Test the record toggle command."""
+    result = runner.invoke(app, ['record', 'status'])
+    assert result.exit_code == 0
+    active = 'Recording is in progress.' in result.stdout
+
+    result = runner.invoke(app, ['record', 'toggle'])
+    assert result.exit_code == 0
+    time.sleep(0.5)  # Wait for the recording to toggle
+    if active:
+        assert 'Recording stopped successfully.' in result.stdout
+    else:
+        assert 'Recording started successfully.' in result.stdout

tests/test_stream.py

@@ -41,3 +41,20 @@ def test_stream_stop():
         assert 'Streaming stopped successfully.' in result.stdout
     else:
         assert 'Streaming is not in progress, cannot stop.' in result.stdout
+
+
+def test_stream_toggle():
+    """Test the stream toggle command."""
+    result = runner.invoke(app, ['stream', 'status'])
+    assert result.exit_code == 0
+    active = 'Streaming is in progress' in result.stdout
+
+    result = runner.invoke(app, ['stream', 'toggle'])
+    assert result.exit_code == 0
+
+    time.sleep(2)  # Wait for the stream to toggle
+
+    if active:
+        assert 'Streaming stopped successfully.' in result.stdout
+    else:
+        assert 'Streaming started successfully.' in result.stdout