Compare commits

..

No commits in common. "main" and "v0.20.1" have entirely different histories.

21 changed files with 50 additions and 134 deletions

View File

@ -1,4 +1,4 @@
# SPDX-FileCopyrightText: 2025-present onyx-and-iris <code@onyxandiris.online>
#
# SPDX-License-Identifier: MIT
__version__ = '0.21.0'
__version__ = '0.20.1'

View File

@ -103,6 +103,17 @@ def main(
show_default=5,
),
] = settings.get('timeout'),
version: Annotated[
bool,
typer.Option(
'--version',
'-v',
is_eager=True,
help='Show the CLI version and exit',
show_default=False,
callback=version_callback,
),
] = False,
style: Annotated[
str,
typer.Option(
@ -124,17 +135,6 @@ def main(
show_default=False,
),
] = settings.get('style_no_border'),
version: Annotated[
bool,
typer.Option(
'--version',
'-v',
is_eager=True,
help='Show the CLI version and exit',
show_default=False,
callback=version_callback,
),
] = False,
debug: Annotated[
bool,
typer.Option(
@ -151,9 +151,7 @@ def main(
):
"""obsws_cli is a command line interface for the OBS WebSocket API."""
ctx.ensure_object(dict)
ctx.obj['obsws'] = ctx.with_resource(
obsws.ReqClient(host=host, port=port, password=password, timeout=timeout)
)
ctx.obj['obsws'] = ctx.with_resource(obsws.ReqClient(**ctx.params))
ctx.obj['style'] = styles.request_style_obj(style, no_border)

View File

@ -24,10 +24,6 @@ def list_(
"""List all hotkeys."""
resp = ctx.obj['obsws'].get_hotkey_list()
if not resp.hotkeys:
console.out.print('No hotkeys found.')
raise typer.Exit()
table = Table(
title='Hotkeys',
padding=(0, 2),

View File

@ -22,10 +22,6 @@ def list_(ctx: typer.Context):
"""List profiles."""
resp = ctx.obj['obsws'].get_profile_list()
if not resp.profiles:
console.out.print('No profiles found.')
raise typer.Exit()
table = Table(
title='Profiles', padding=(0, 2), border_style=ctx.obj['style'].border
)

View File

@ -21,15 +21,16 @@ def main():
def list_monitors(ctx: typer.Context):
"""List available monitors."""
resp = ctx.obj['obsws'].get_monitor_list()
if not resp.monitors:
console.out.print('No monitors found.')
return
monitors = sorted(
((m['monitorIndex'], m['monitorName']) for m in resp.monitors),
key=lambda m: m[0],
)
if not monitors:
console.out.print('No monitors found.')
raise typer.Exit()
table = Table(
title='Available Monitors',
padding=(0, 2),

View File

@ -29,10 +29,6 @@ def list_(
for scene in reversed(resp.scenes)
)
if not scenes:
console.out.print('No scenes found.')
raise typer.Exit()
active_scene = ctx.obj['obsws'].get_current_program_scene().scene_name
table = Table(title='Scenes', padding=(0, 2), border_style=ctx.obj['style'].border)

View File

@ -21,10 +21,6 @@ def list_(ctx: typer.Context):
"""List all scene collections."""
resp = ctx.obj['obsws'].get_scene_collection_list()
if not resp.scene_collections:
console.out.print('No scene collections found.')
raise typer.Exit()
table = Table(
title='Scene Collections',
padding=(0, 2),

View File

@ -21,7 +21,7 @@ classifiers = [
"Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy",
]
dependencies = ["typer>=0.19.2", "obsws-python>=1.8.0", "python-dotenv>=1.1.0"]
dependencies = ["typer>=0.16.0", "obsws-python>=1.8.0", "python-dotenv>=1.1.0"]
[project.urls]
@ -42,6 +42,9 @@ dependencies = ["click-man>=0.5.1"]
cli = "obsws-cli {args:}"
man = "python man/generate.py --output=./man"
[tool.hatch.envs.lazyimports.scripts]
cli = "obsws-cli {args:}"
[tool.hatch.envs.hatch-test]
randomize = true

View File

@ -1,7 +1,6 @@
"""pytest configuration file."""
import os
import time
import obsws_python as obsws
from dotenv import find_dotenv, load_dotenv
@ -45,54 +44,9 @@ def pytest_sessionstart(session):
},
)
session.obsws.create_profile('pytest_profile')
time.sleep(0.1) # Wait for the profile to be created
session.obsws.set_profile_parameter(
'SimpleOutput',
'RecRB',
'true',
)
# hack to ensure the replay buffer is enabled
session.obsws.set_current_profile('Untitled')
session.obsws.set_current_profile('pytest_profile')
session.obsws.set_current_scene_collection('test-collection')
session.obsws.create_scene('pytest_scene')
# Ensure Desktop Audio is created.
desktop_audio_kinds = {
'windows': 'wasapi_output_capture',
'linux': 'pulse_output_capture',
'darwin': 'coreaudio_output_capture',
}
platform = os.environ.get('OBS_TESTS_PLATFORM', os.uname().sysname.lower())
try:
session.obsws.create_input(
sceneName='pytest_scene',
inputName='Desktop Audio',
inputKind=desktop_audio_kinds[platform],
inputSettings={'device_id': 'default'},
sceneItemEnabled=True,
)
except obsws.error.OBSSDKRequestError as e:
if e.code == 601:
"""input already exists, continue."""
# Ensure Mic/Aux is created.
mic_kinds = {
'windows': 'wasapi_input_capture',
'linux': 'pulse_input_capture',
'darwin': 'coreaudio_input_capture',
}
try:
session.obsws.create_input(
sceneName='pytest_scene',
inputName='Mic/Aux',
inputKind=mic_kinds[platform],
inputSettings={'device_id': 'default'},
sceneItemEnabled=True,
)
except obsws.error.OBSSDKRequestError as e:
if e.code == 601:
"""input already exists, continue."""
session.obsws.create_input(
sceneName='pytest_scene',
inputName='pytest_input',
@ -177,7 +131,7 @@ def pytest_sessionfinish(session, exitstatus):
session.obsws.remove_scene('pytest_scene')
session.obsws.set_current_scene_collection('Untitled')
session.obsws.set_current_scene_collection('default')
resp = session.obsws.get_stream_status()
if resp.output_active:
@ -195,8 +149,6 @@ def pytest_sessionfinish(session, exitstatus):
if resp.studio_mode_enabled:
session.obsws.set_studio_mode_enabled(False)
session.obsws.remove_profile('pytest_profile')
# Close the OBS WebSocket client connection
session.obsws.disconnect()

View File

@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_filter_list():

View File

@ -1,18 +1,10 @@
"""Unit tests for the group command in the OBS WebSocket CLI."""
import os
import pytest
from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
if os.environ.get('OBS_TESTS_SKIP_GROUP_TESTS'):
pytest.skip(
'Skipping group tests as per environment variable', allow_module_level=True
)
runner = CliRunner(mix_stderr=False)
def test_group_list():

View File

@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_hotkey_list():

View File

@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_input_list():
@ -13,7 +13,10 @@ def test_input_list():
assert result.exit_code == 0
assert 'Desktop Audio' in result.stdout
assert 'Mic/Aux' in result.stdout
assert all(item in result.stdout for item in ('pytest_input', 'pytest_input_2'))
assert all(
item in result.stdout
for item in ('Colour Source', 'Colour Source 2', 'Colour Source 3')
)
def test_input_list_filter_input():
@ -36,6 +39,9 @@ def test_input_list_filter_colour():
"""Test the input list command with colour filter."""
result = runner.invoke(app, ['input', 'list', '--colour'])
assert result.exit_code == 0
assert all(item in result.stdout for item in ('pytest_input', 'pytest_input_2'))
assert all(
item in result.stdout
for item in ('Colour Source', 'Colour Source 2', 'Colour Source 3')
)
assert 'Desktop Audio' not in result.stdout
assert 'Mic/Aux' not in result.stdout

View File

@ -6,7 +6,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_record_start():
@ -49,9 +49,7 @@ def test_record_toggle():
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:

View File

@ -1,20 +1,10 @@
"""Unit tests for the replaybuffer command in the OBS WebSocket CLI."""
import os
import time
import pytest
from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
if os.environ.get('OBS_TESTS_SKIP_REPLAYBUFFER_TESTS'):
pytest.skip(
'Skipping replaybuffer tests as per environment variable',
allow_module_level=True,
)
runner = CliRunner(mix_stderr=False)
def test_replaybuffer_start():
@ -24,9 +14,6 @@ def test_replaybuffer_start():
active = 'Replay buffer is active.' in resp.stdout
resp = runner.invoke(app, ['replaybuffer', 'start'])
time.sleep(0.5) # Wait for the replay buffer to start
if active:
assert resp.exit_code != 0
assert 'Replay buffer is already active.' in resp.stderr
@ -42,9 +29,6 @@ def test_replaybuffer_stop():
active = 'Replay buffer is active.' in resp.stdout
resp = runner.invoke(app, ['replaybuffer', 'stop'])
time.sleep(0.5) # Wait for the replay buffer to stop
if not active:
assert resp.exit_code != 0
assert 'Replay buffer is not active.' in resp.stderr
@ -60,11 +44,9 @@ def test_replaybuffer_toggle():
active = 'Replay buffer is active.' in resp.stdout
resp = runner.invoke(app, ['replaybuffer', 'toggle'])
assert resp.exit_code == 0
time.sleep(0.5) # Wait for the replay buffer to toggle
if active:
assert resp.exit_code == 0
assert 'Replay buffer is not active.' in resp.stdout
else:
assert resp.exit_code == 0
assert 'Replay buffer is active.' in resp.stdout

View File

@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_scene_list():

View File

@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_sceneitem_list():

View File

@ -6,7 +6,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_stream_start():
@ -23,7 +23,7 @@ def test_stream_start():
else:
assert result.exit_code == 0
assert 'Streaming started successfully.' in result.stdout
time.sleep(0.5) # Wait for the streaming to start
time.sleep(1) # Wait for the streaming to start
def test_stream_stop():
@ -37,7 +37,7 @@ def test_stream_stop():
if active:
assert result.exit_code == 0
assert 'Streaming stopped successfully.' in result.stdout
time.sleep(0.5) # Wait for the streaming to stop
time.sleep(1) # Wait for the streaming to stop
else:
assert result.exit_code != 0
assert 'Streaming is not in progress, cannot stop.' in result.stderr
@ -52,7 +52,7 @@ def test_stream_toggle():
result = runner.invoke(app, ['stream', 'toggle'])
assert result.exit_code == 0
time.sleep(0.5) # Wait for the stream to toggle
time.sleep(1) # Wait for the stream to toggle
if active:
assert 'Streaming stopped successfully.' in result.stdout

View File

@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_studio_enable():

View File

@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_text_update():

View File

@ -4,7 +4,7 @@ from typer.testing import CliRunner
from obsws_cli.app import app
runner = CliRunner()
runner = CliRunner(mix_stderr=False)
def test_version():