ensure tox tests run setup/teardown scripts

bump min python version to 3.11 (because of ExceptionGroup in tests)

patch bump

README.md

@@ -19,7 +19,7 @@ For an outline of past/future changes refer to: [CHANGELOG](CHANGELOG.md)
 
 ## Requirements
 
--   Python 3.10 or greater
+-   Python 3.11 or greater
 -   [Streamlabs Desktop][sl-desktop]
 -   A websocket token: Settings > Remote Control > API Token
 

pdm.lock

@@ -5,7 +5,7 @@
 groups = ["default", "dev"]
 strategy = ["inherit_metadata"]
 lock_version = "4.5.0"
-content_hash = "sha256:b032bb4d22d3d3b10233c543cd182ac8e7ec052aa9dc03a3034a967066e85db2"
+content_hash = "sha256:6f29a16938942eb7911abc4c1505647bb80b5275cb45ac72f9166b53ae6c0ef1"
 
 [[metadata.targets]]
 requires_python = ">=3.10"
@@ -238,6 +238,21 @@ files = [
     {file = "pytest-8.4.0.tar.gz", hash = "sha256:14d920b48472ea0dbf68e45b96cd1ffda4705f33307dcc86c676c1b5104838a6"},
 ]
 
+[[package]]
+name = "pytest-randomly"
+version = "3.16.0"
+requires_python = ">=3.9"
+summary = "Pytest plugin to randomly order tests and control random.seed."
+groups = ["dev"]
+dependencies = [
+    "importlib-metadata>=3.6; python_version < \"3.10\"",
+    "pytest",
+]
+files = [
+    {file = "pytest_randomly-3.16.0-py3-none-any.whl", hash = "sha256:8633d332635a1a0983d3bba19342196807f6afb17c3eef78e02c2f85dade45d6"},
+    {file = "pytest_randomly-3.16.0.tar.gz", hash = "sha256:11bf4d23a26484de7860d82f726c0629837cf4064b79157bd18ec9d41d7feb26"},
+]
+
 [[package]]
 name = "sniffio"
 version = "1.3.1"

pyproject.toml

@@ -3,7 +3,7 @@ name = "slobs-cli"
 description = "A command line application for Streamlabs Desktop"
 authors = [{ name = "onyx-and-iris", email = "code@onyxandiris.online" }]
 dependencies = ["pyslobs>=2.0.4", "asyncclick>=8.1.8"]
-requires-python = ">=3.10"
+requires-python = ">=3.11"
 readme = "README.md"
 license = { text = "MIT" }
 dynamic = ["version"]
@@ -27,9 +27,15 @@ path = "src/slobs_cli/__about__.py"
 cli.cmd = "slobs-cli {args}"
 cli.env_file = ".env"
 
+pre_test.cmd = "python tests/setup.py"
 test.cmd = "pytest {args}"
 test.env_file = ".env"
 post_test.cmd = "python tests/teardown.py"
 
 [dependency-groups]
-dev = ["tox-pdm>=0.7.2", "pytest>=8.4.0", "virtualenv-pyenv>=0.5.0"]
+dev = [
+    "tox-pdm>=0.7.2",
+    "pytest>=8.4.0",
+    "virtualenv-pyenv>=0.5.0",
+    "pytest-randomly>=3.16.0",
+]

src/slobs_cli/__about__.py

@@ -1 +1 @@
-__version__ = "0.7.6"
+__version__ = "0.7.8"

src/slobs_cli/record.py

@@ -19,8 +19,8 @@ async def start(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.recording_status != "offline"
+        model = await ss.get_model()
+        active = model.recording_status != "offline"
 
         if active:
             conn.close()
@@ -45,8 +45,8 @@ async def stop(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.recording_status != "offline"
+        model = await ss.get_model()
+        active = model.recording_status != "offline"
 
         if not active:
             conn.close()
@@ -71,8 +71,8 @@ async def status(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.recording_status != "offline"
+        model = await ss.get_model()
+        active = model.recording_status != "offline"
 
         if active:
             click.echo("Recording is currently active.")
@@ -95,8 +95,8 @@ async def toggle(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.recording_status != "offline"
+        model = await ss.get_model()
+        active = model.recording_status != "offline"
 
         if active:
             await ss.toggle_recording()

src/slobs_cli/replaybuffer.py

@@ -19,8 +19,8 @@ async def start(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.replay_buffer_status != "offline"
+        model = await ss.get_model()
+        active = model.replay_buffer_status != "offline"
 
         if active:
             conn.close()
@@ -44,8 +44,8 @@ async def stop(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.replay_buffer_status != "offline"
+        model = await ss.get_model()
+        active = model.replay_buffer_status != "offline"
 
         if not active:
             conn.close()
@@ -71,8 +71,8 @@ async def status(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.replay_buffer_status != "offline"
+        model = await ss.get_model()
+        active = model.replay_buffer_status != "offline"
         if active:
             click.echo("Replay buffer is currently active.")
         else:

src/slobs_cli/scene.py

@@ -77,9 +77,9 @@ async def switch(ctx: click.Context, scene_name: str, preview: bool = False):
         scenes = await ss.get_scenes()
         for scene in scenes:
             if scene.name == scene_name:
-                current_state = await ts.get_model()
+                model = await ts.get_model()
 
-                if current_state.studio_mode:
+                if model.studio_mode:
                     await ss.make_scene_active(scene.id)
                     if preview:
                         click.echo(
@@ -104,13 +104,13 @@ async def switch(ctx: click.Context, scene_name: str, preview: bool = False):
                     click.echo(
                         f"Switched to scene: {click.style(scene.name, fg='blue')} (ID: {scene.id}) in active mode."
                     )
+                conn.close()
                 break
         else:
             conn.close()
             raise click.ClickException(
                 click.style(f"Scene '{scene_name}' not found.", fg="red")
             )
-        conn.close()
 
     async with create_task_group() as tg:
         tg.start_soon(conn.background_processing)

src/slobs_cli/stream.py

@@ -19,8 +19,8 @@ async def start(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.streaming_status != "offline"
+        model = await ss.get_model()
+        active = model.streaming_status != "offline"
 
         if active:
             conn.close()
@@ -44,8 +44,8 @@ async def stop(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.streaming_status != "offline"
+        model = await ss.get_model()
+        active = model.streaming_status != "offline"
 
         if not active:
             conn.close()
@@ -69,8 +69,8 @@ async def status(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.streaming_status != "offline"
+        model = await ss.get_model()
+        active = model.streaming_status != "offline"
 
         if active:
             click.echo("Stream is currently active.")
@@ -92,8 +92,8 @@ async def toggle(ctx: click.Context):
     ss = StreamingService(conn)
 
     async def _run():
-        current_state = await ss.get_model()
-        active = current_state.streaming_status != "offline"
+        model = await ss.get_model()
+        active = model.streaming_status != "offline"
 
         await ss.toggle_streaming()
         if active:

src/slobs_cli/studiomode.py

@@ -19,8 +19,8 @@ async def enable(ctx: click.Context):
     ts = TransitionsService(conn)
 
     async def _run():
-        current_state = await ts.get_model()
-        if current_state.studio_mode:
+        model = await ts.get_model()
+        if model.studio_mode:
             conn.close()
             raise click.Abort(click.style("Studio mode is already enabled.", fg="red"))
 
@@ -42,8 +42,8 @@ async def disable(ctx: click.Context):
     ts = TransitionsService(conn)
 
     async def _run():
-        current_state = await ts.get_model()
-        if not current_state.studio_mode:
+        model = await ts.get_model()
+        if not model.studio_mode:
             conn.close()
             raise click.Abort(click.style("Studio mode is already disabled.", fg="red"))
 
@@ -65,8 +65,8 @@ async def status(ctx: click.Context):
     ts = TransitionsService(conn)
 
     async def _run():
-        current_state = await ts.get_model()
-        if current_state.studio_mode:
+        model = await ts.get_model()
+        if model.studio_mode:
             click.echo("Studio mode is currently enabled.")
         else:
             click.echo("Studio mode is currently disabled.")
@@ -86,8 +86,8 @@ async def toggle(ctx: click.Context):
     ts = TransitionsService(conn)
 
     async def _run():
-        current_state = await ts.get_model()
-        if current_state.studio_mode:
+        model = await ts.get_model()
+        if model.studio_mode:
             await ts.disable_studio_mode()
             click.echo("Studio mode disabled successfully.")
         else:
@@ -109,8 +109,8 @@ async def force_transition(ctx: click.Context):
     ts = TransitionsService(conn)
 
     async def _run():
-        current_state = await ts.get_model()
-        if not current_state.studio_mode:
+        model = await ts.get_model()
+        if not model.studio_mode:
             conn.close()
             raise click.Abort(click.style("Studio mode is not enabled.", fg="red"))
 

tests/setup.py

@@ -0,0 +1,32 @@
+import os
+
+import anyio
+from anyio import create_task_group
+from pyslobs import ConnectionConfig, ScenesService, SlobsConnection
+
+
+async def setup(conn: SlobsConnection):
+    ss = ScenesService(conn)
+    await ss.create_scene("slobs-test-scene-1")
+    await ss.create_scene("slobs-test-scene-2")
+    await ss.create_scene("slobs-test-scene-3")
+
+    conn.close()
+
+
+async def main():
+    conn = SlobsConnection(
+        ConnectionConfig(
+            domain=os.environ["SLOBS_DOMAIN"],
+            port=59650,
+            token=os.environ["SLOBS_TOKEN"],
+        )
+    )
+
+    async with create_task_group() as tg:
+        tg.start_soon(conn.background_processing)
+        tg.start_soon(setup, conn)
+
+
+if __name__ == "__main__":
+    anyio.run(main)

tests/teardown.py

@@ -2,17 +2,23 @@ import os
 
 import anyio
 from anyio import create_task_group
-from pyslobs import ConnectionConfig, SlobsConnection, StreamingService
+from pyslobs import ConnectionConfig, ScenesService, SlobsConnection, StreamingService
 
 
 async def cleanup(conn: SlobsConnection):
+    ss = ScenesService(conn)
+    scenes = await ss.get_scenes()
+    for scene in scenes:
+        if scene.name.startswith("slobs-test-scene-"):
+            await ss.remove_scene(scene.id)
+
     ss = StreamingService(conn)
-    current_state = await ss.get_model()
-    if current_state.streaming_status != "offline":
+    model = await ss.get_model()
+    if model.streaming_status != "offline":
         await ss.toggle_streaming()
-    if current_state.replay_buffer_status != "offline":
+    if model.replay_buffer_status != "offline":
         await ss.stop_replay_buffer()
-    if current_state.recording_status != "offline":
+    if model.recording_status != "offline":
         await ss.toggle_recording()
 
     conn.close()

tox.ini

@@ -1,9 +1,11 @@
 [tox]
-env_list = py{310,311,312}
+env_list = py{311,312,313}
 
 [testenv]
 passenv = *
 setenv = VIRTUALENV_DISCOVERY=pyenv
 groups = dev
 commands =
-    pytest tests
+    python tests/setup.py
+    pytest tests
+    python tests/teardown.py