Skip to content

Commit

Permalink
Testing
Browse files Browse the repository at this point in the history
  • Loading branch information
@josephine-wolf-oberholtzer
josephine-wolf-oberholtzer committed Dec 23, 2024
1 parent fd21503 commit b0eb9eb
Show file tree
Hide file tree
Showing 2 changed files with 109 additions and 27 deletions.
4 changes: 2 additions & 2 deletions supriya/contexts/realtime.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -605,7 +605,7 @@ def boot(self, *, options: Optional[Options] = None, **kwargs) -> "Server":
self._lifecycle_thread.start()
if not (self._boot_future.result()):
if (self._shutdown_future.result()) == ServerShutdownEvent.PROCESS_PANIC:
raise ServerCannotBoot
raise ServerCannotBoot(self._process_protocol.error_text)
return self

def connect(self, *, options: Optional[Options] = None, **kwargs) -> "Server":
Expand Down Expand Up @@ -1158,7 +1158,7 @@ async def boot(
self._lifecycle_task = loop.create_task(self._lifecycle(owned=True))
if not (await self._boot_future):
if (await self._shutdown_future) == ServerShutdownEvent.PROCESS_PANIC:
raise ServerCannotBoot
raise ServerCannotBoot(self._process_protocol.error_text)
return self

async def connect(
Expand Down
132 changes: 107 additions & 25 deletions tests/mixers/test_Track.py
View file
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import asyncio
from typing import Dict, List, Optional, Tuple, Union, cast
from typing import Dict, List, Optional, Tuple, Union

import pytest

Expand Down Expand Up @@ -1332,11 +1332,10 @@ async def test_Track_set_input(

@pytest.mark.parametrize("online", [False, True])
@pytest.mark.parametrize(
"target, muted, expected_commands, expected_state",
"actions, expected_commands, expected_state",
[
(
"mixers[0].tracks[0]",
True,
[("mixers[0].tracks[0]", [True])],
[OscMessage("/c_set", 5, 0.0)],
[
("m[0].t[0]", (0.0, 0.0), False),
Expand All @@ -1349,8 +1348,7 @@ async def test_Track_set_input(
],
),
(
"mixers[0].tracks[0].tracks[0]",
True,
[("mixers[0].tracks[0].tracks[0]", [True])],
[OscMessage("/c_set", 11, 0.0)],
[
("m[0].t[0]", (0.0, 0.0), True),
Expand All @@ -1366,12 +1364,11 @@ async def test_Track_set_input(
)
@pytest.mark.asyncio
async def test_Track_set_muted(
muted: bool,
actions: List[Tuple[str, List[bool]]],
complex_session: Tuple[Session, str],
expected_commands: List[Union[OscBundle, OscMessage]],
expected_state: List[Tuple[str, Tuple[float, ...], bool]],
online: bool,
target: str,
) -> None:
# Pre-conditions
session, _ = complex_session
Expand All @@ -1381,11 +1378,12 @@ async def test_Track_set_muted(
await session.mixers[0].tracks[1].sends[0].delete()
await asyncio.sleep(0.25)
initial_tree = await debug_tree(session)
target_ = session[target]
assert isinstance(target_, Track)
# Operation
with capture(session["mixers[0]"].context) as commands:
await target_.set_muted(muted)
for target, args in actions:
target_ = session[target]
assert isinstance(target_, Track)
await target_.set_muted(*args)
# Post-conditions
if not online:
return
Expand All @@ -1398,10 +1396,11 @@ async def test_Track_set_muted(
assert [
(
track.short_address,
tuple(round(x, 6) for x in cast(Track, track).output_levels),
tuple(round(x, 6) for x in track.output_levels),
track.is_active,
)
for track in session._walk(Track)
if isinstance(track, Track)
] == expected_state


Expand Down Expand Up @@ -1673,42 +1672,125 @@ async def test_Track_set_output(
assert commands == expected_commands


@pytest.mark.xfail
@pytest.mark.parametrize("online", [False, True])
@pytest.mark.parametrize(
"expected_commands, expected_diff",
"actions, expected_commands, expected_state",
[
([], ""),
(
[
("mixers[0].tracks[0]", [True]),
],
[
OscMessage("/c_set", 5, 1.0),
OscMessage("/c_set", 11, 1.0),
OscMessage("/c_set", 17, 1.0),
OscMessage("/c_set", 23, 1.0),
OscMessage("/c_set", 29, 0.0),
OscMessage("/c_set", 35, 0.0),
],
[
("m[0].t[0]", (1.0, 1.0), True),
("m[0].t[0].t[0]", (1.0, 1.0), True),
("m[0].t[0].t[0].t[0]", (1.0, 1.0), True),
("m[0].t[0].t[1]", (0.0, 0.0), True),
("m[0].t[1]", (0.0, 0.0), False),
("m[0].t[2]", (0.0, 0.0), False),
("m[1].t[0]", (0.0, 0.0), True),
],
),
(
[
("mixers[0].tracks[0].tracks[0]", [True]),
],
[
OscMessage("/c_set", 5, 0.0),
OscMessage("/c_set", 11, 1.0),
OscMessage("/c_set", 17, 1.0),
OscMessage("/c_set", 23, 0.0),
OscMessage("/c_set", 29, 0.0),
OscMessage("/c_set", 35, 0.0),
],
[
("m[0].t[0]", (0.0, 0.0), False),
("m[0].t[0].t[0]", (1.0, 1.0), True),
("m[0].t[0].t[0].t[0]", (1.0, 1.0), True),
("m[0].t[0].t[1]", (0.0, 0.0), False),
("m[0].t[1]", (0.0, 0.0), False),
("m[0].t[2]", (0.0, 0.0), False),
("m[1].t[0]", (0.0, 0.0), True),
],
),
(
[
("mixers[0].tracks[0]", [True]),
("mixers[0].tracks[1]", [True, False]),
],
[
OscMessage("/c_set", 5, 1.0),
OscMessage("/c_set", 11, 1.0),
OscMessage("/c_set", 17, 1.0),
OscMessage("/c_set", 23, 1.0),
OscMessage("/c_set", 29, 0.0),
OscMessage("/c_set", 35, 0.0),
OscMessage("/c_set", 5, 1.0),
OscMessage("/c_set", 11, 1.0),
OscMessage("/c_set", 17, 1.0),
OscMessage("/c_set", 23, 1.0),
OscMessage("/c_set", 29, 1.0),
OscMessage("/c_set", 35, 0.0),
],
[
("m[0].t[0]", (1.0, 1.0), True),
("m[0].t[0].t[0]", (1.0, 1.0), True),
("m[0].t[0].t[0].t[0]", (1.0, 1.0), True),
("m[0].t[0].t[1]", (0.0, 0.0), True),
("m[0].t[1]", (1.0, 1.0), True),
("m[0].t[2]", (0.0, 0.0), False),
("m[1].t[0]", (0.0, 0.0), True),
],
),
],
)
@pytest.mark.asyncio
async def test_Track_set_soloed(
actions: List[Tuple[str, List[bool]]],
complex_session: Tuple[Session, str],
expected_commands: List[Union[OscBundle, OscMessage]],
expected_diff: str,
expected_state: List[Tuple[str, Tuple[float, ...], bool]],
online: bool,
target: str,
) -> None:
# Pre-conditions
session, initial_tree = complex_session
if online:
await session.boot()
target_ = session[target]
assert isinstance(target_, Track)
await session.mixers[0].tracks[0].tracks[0].tracks[0].add_device()
await session.mixers[0].tracks[1].sends[0].delete()
await asyncio.sleep(0.25)
initial_tree = await debug_tree(session)
# Operation
with capture(session["mixers[0]"].context) as commands:
await target_.set_soloed()
for target, args in actions:
target_ = session[target]
assert isinstance(target_, Track)
await target_.set_soloed(*args)
# Post-conditions
if not online:
raise Exception
return
await assert_diff(
session,
expected_diff,
expected_initial_tree="""
""",
"",
expected_initial_tree=initial_tree,
)
assert commands == expected_commands
raise Exception
assert [
(
track.short_address,
tuple(round(x, 6) for x in track.output_levels),
track.is_active,
)
for track in session._walk(Track)
if isinstance(track, Track)
] == expected_state


@pytest.mark.xfail
Expand Down

0 comments on commit b0eb9eb

Please sign in to comment.