Samuel Onoja
7a7df0106c
Invalid configuration currently prevents login. Ferese now starts with built-in defaults and shows a notification when the shell is ready, leaving the file untouched. Loading and validation belong to the config module, while the theme module resolves appearance. Validation remains strict, and failed live reloads keep the active settings. The native Morph regression checks the animated input region and idle frame scheduling while its backing surface stays allocated. Validation: workspace and Python tests, formatting, Clippy, release build, and isolated Wayland startup/reload and panel-motion tests. Native CI includes the startup regression. Related: #28.
410 lines
27 KiB
Python
410 lines
27 KiB
Python
"""Exercise real shell frame-driven motion on a private nested desktop and bus.
|
|
|
|
FERESE_TEST_SHELL_MOTION=1 python3 scripts/tests/test_shell_motion_isolated.py
|
|
Set FERESE_TEST_BAR_LAYOUT=floating to also test island masks and live switching.
|
|
Uses release binaries by default. Requires a Wayland host, dbus-daemon and gdbus.
|
|
Protocol traces contain only this test's shell traffic, never the host desktop.
|
|
"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from session_socket import ipc_environment, ipc_socket
|
|
import re
|
|
import signal
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
|
|
def panel_config(preset):
|
|
style = preset
|
|
return f'''panel main {{
|
|
style "{style}"
|
|
start {{ group navigation {{ island-padding 12; item overview kind=overview; item spaces kind=workspaces; }}; }}
|
|
center {{ group title {{ island-padding 12; item title kind=focused-window; }}; }}
|
|
end {{ group status {{ island-padding 12; item media kind=media; item quick kind=quick-settings; item network kind=network; item audio kind=audio; item recording kind=recording; item notifications kind=notifications; item battery kind=battery; item clock kind=clock; item display kind=display-mode; }}; }}
|
|
}}
|
|
'''
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("FERESE_TEST_SHELL_MOTION") == "1", "opt-in nested shell test")
|
|
class ShellMotionTest(unittest.TestCase):
|
|
def test_notifications_and_modals_follow_frames_and_finish_closing(self):
|
|
repo = Path(__file__).resolve().parents[2]
|
|
binary = repo / os.environ.get("FERESE_TEST_BINARY", "target/release/ferese")
|
|
shell = repo / os.environ.get("FERESE_TEST_SHELL", "target/release/ferese-shell")
|
|
ctl = repo / os.environ.get("FERESE_TEST_CTL", "target/release/feresectl")
|
|
children = []
|
|
with tempfile.TemporaryDirectory(prefix="ferese-shell-motion-") as temporary:
|
|
root = Path(temporary)
|
|
runtime = root / "runtime"
|
|
runtime.mkdir(mode=0o700)
|
|
config = root / "config/ferese"
|
|
config.mkdir(parents=True)
|
|
bar_layout = os.environ.get("FERESE_TEST_BAR_LAYOUT", "flat")
|
|
self.assertIn(bar_layout, ("flat", "floating"))
|
|
config_file = config / "config.kdl"
|
|
config_file.write_text(
|
|
'animations { speed 0.5; }\n'
|
|
'status { keybinding-guide #false; }\n' + panel_config(bar_layout))
|
|
display = Path(os.environ["WAYLAND_DISPLAY"])
|
|
if not display.is_absolute():
|
|
display = Path(os.environ["XDG_RUNTIME_DIR"]) / display
|
|
env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), WAYLAND_DISPLAY=str(display),
|
|
XDG_CONFIG_HOME=str(root / "config"), XDG_STATE_HOME=str(root / "state"),
|
|
XDG_CACHE_HOME=str(root / "cache"), RUST_LOG="ferese=info,ferese::nested_input=debug")
|
|
env.pop("WAYLAND_SOCKET", None)
|
|
env.pop("FERESE_SHELL_CONTROL_SOCKET", None)
|
|
log_path = root / "session.log"
|
|
# No service directories: readiness checks must not auto-activate
|
|
# the host's notification daemon on this private bus.
|
|
bus_config = root / "bus.conf"
|
|
bus_config.write_text(
|
|
'<busconfig><type>session</type><listen>unix:tmpdir=/tmp</listen>'
|
|
'<policy context="default"><allow send_destination="*"/>'
|
|
'<allow receive_sender="*"/><allow own="*"/></policy></busconfig>')
|
|
|
|
def launch(command, **kwargs):
|
|
process = subprocess.Popen(command, env=ipc_environment(env), start_new_session=True, **kwargs)
|
|
children.append(process)
|
|
return process
|
|
|
|
def wait_for(predicate, timeout=15):
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
value = predicate()
|
|
if value:
|
|
return value
|
|
time.sleep(.025)
|
|
trace = log_path.read_text()
|
|
relevant = '\n'.join(line for line in trace.splitlines() if any(
|
|
word in line for word in ('get_layer_surface', 'ERROR', 'WARN', 'notification')))
|
|
self.fail("Timed out:\n" + relevant[-16000:] + '\n' + trace[-4000:])
|
|
|
|
def dbus(method, *args, check=True):
|
|
return subprocess.run(["gdbus", "call", "--session", "--dest", "org.freedesktop.Notifications",
|
|
"--object-path", "/org/freedesktop/Notifications", "--method",
|
|
"org.freedesktop.Notifications." + method, *args],
|
|
env=ipc_environment(env), text=True, capture_output=True, timeout=5, check=check)
|
|
|
|
def layer_surface(namespace):
|
|
matches = layer_surfaces(namespace)
|
|
# Wayland object IDs can be reused after destruction. Restrict
|
|
# assertions to this surface's lifetime in the trace.
|
|
return matches[-1] if matches else None
|
|
|
|
def layer_surfaces(namespace):
|
|
return [(match.group(1), match.start()) for match in re.finditer(
|
|
r'get_layer_surface\([^\n]*?wl_surface[@#](\d+)[^\n]*"' + namespace + '"',
|
|
log_path.read_text())]
|
|
|
|
def frames(surface):
|
|
identity, start = surface
|
|
return len(re.findall(r'wl_surface[@#]' + identity + r'\.frame\(', log_path.read_text()[start:]))
|
|
|
|
def commits(surface):
|
|
identity, start = surface
|
|
return len(re.findall(r'wl_surface[@#]' + identity + r'\.commit\(', log_path.read_text()[start:]))
|
|
|
|
def destroyed(surface):
|
|
identity, start = surface
|
|
return re.search(r'wl_surface[@#]' + identity + r'\.destroy\(', log_path.read_text()[start:]) is not None
|
|
|
|
def bar_region_count(surface):
|
|
identity, start = surface
|
|
trace = log_path.read_text()[start:]
|
|
effects = re.findall(
|
|
r'get_surface_effects\(new id ferese_surface_effects_v1[@#](\d+), wl_surface[@#]'
|
|
+ identity + r'\)', trace)
|
|
if not effects:
|
|
return None
|
|
|
|
updates = re.findall(
|
|
r'ferese_surface_effects_v1[@#]' + effects[-1] + r'\.set_panel_presentation\(array\[(\d+)\],', trace)
|
|
return int(updates[-1]) // 20 if updates else None
|
|
|
|
def bar_size(surface):
|
|
identity, start = surface
|
|
trace = log_path.read_text()[start:]
|
|
match = re.search(r'get_layer_surface\(new id zwlr_layer_surface_v1[@#](\d+), wl_surface[@#]' + identity + r',', trace)
|
|
if not match:
|
|
return None
|
|
updates = re.findall(r'zwlr_layer_surface_v1[@#]' + match.group(1) + r'\.set_size\((\d+), (\d+)\)', trace)
|
|
return tuple(map(int, updates[-1])) if updates else None
|
|
|
|
def bar_input_rectangles(surface):
|
|
identity, start = surface
|
|
trace = log_path.read_text()[start:]
|
|
updates = list(re.finditer(
|
|
r'wl_surface[@#]' + identity + r'\.set_input_region\(wl_region[@#](\d+)\)', trace))
|
|
if not updates:
|
|
return []
|
|
|
|
update = updates[-1]
|
|
region = update.group(1)
|
|
creations = list(re.finditer(r'create_region\(new id wl_region[@#]' + region + r'\)', trace[:update.start()]))
|
|
if not creations:
|
|
return []
|
|
|
|
rectangles = [tuple(map(int, values)) for values in re.findall(
|
|
r'wl_region[@#]' + region + r'\.add\((-?\d+), (-?\d+), (\d+), (\d+)\)',
|
|
trace[creations[-1].start():update.start()])]
|
|
# Wayland regions are unions. Decorated groups can include
|
|
# redundant child control rectangles without changing coverage.
|
|
def contains(outer, inner):
|
|
x, y, width, height = outer
|
|
ix, iy, iw, ih = inner
|
|
return x <= ix and y <= iy and x + width >= ix + iw and y + height >= iy + ih
|
|
return [rectangle for index, rectangle in enumerate(rectangles)
|
|
if not any(other != rectangle and contains(other, rectangle)
|
|
or other == rectangle and other_index < index
|
|
for other_index, other in enumerate(rectangles))]
|
|
|
|
def bar_input_size(surface):
|
|
rectangles = bar_input_rectangles(surface)
|
|
if not rectangles:
|
|
return None
|
|
return (max(x + width for x, y, width, height in rectangles)
|
|
- min(x for x, y, width, height in rectangles),
|
|
max(y + height for x, y, width, height in rectangles)
|
|
- min(y for x, y, width, height in rectangles))
|
|
|
|
def bar_motion_settled(surface):
|
|
updates = log_path.read_text().count('.set_panel_presentation(')
|
|
before = commits(surface)
|
|
time.sleep(.25)
|
|
return (log_path.read_text().count('.set_panel_presentation(') == updates
|
|
and commits(surface) - before <= 2)
|
|
|
|
|
|
with log_path.open("w") as log:
|
|
try:
|
|
bus = launch(["dbus-daemon", "--config-file=" + str(bus_config), "--nofork", "--print-address=1"],
|
|
stdout=subprocess.PIPE, stderr=log, text=True)
|
|
env["DBUS_SESSION_BUS_ADDRESS"] = bus.stdout.readline().strip()
|
|
compositor = launch([str(binary), "--backend", "nested", "--grant-effects", "--grant-shell-control",
|
|
"--", "env", "WAYLAND_DEBUG=client", "ICED_BACKEND=tiny-skia", str(shell)],
|
|
stdout=log, stderr=log)
|
|
wait_for(lambda: (ipc_socket(runtime)).is_socket())
|
|
wait_for(lambda: dbus("GetServerInformation", check=False).returncode == 0)
|
|
bar = wait_for(lambda: layer_surface("ferese-shell-top-bar"))
|
|
if bar_layout == "floating":
|
|
wait_for(lambda: bar_region_count(bar) == 2)
|
|
original_input = wait_for(lambda: bar_input_rectangles(bar))
|
|
original_start_width = max(x + width for x, y, width, height in original_input if x < 300) - min(x for x, y, width, height in original_input if x < 300)
|
|
time.sleep(.5)
|
|
reply = dbus("Notify", "Ferese motion test", "0", "", "Frame cadence", "Synthetic test notification", "[]", "{}", "0").stdout
|
|
notice = re.search(r'uint32 (\d+)', reply).group(1)
|
|
surface = wait_for(lambda: layer_surface("ferese-shell-notifications"))
|
|
time.sleep(1.5)
|
|
opening_frames = frames(surface)
|
|
# Require multiple animation frames, without imposing a
|
|
# hardware-dependent frame rate on software-rendered CI.
|
|
self.assertGreaterEqual(opening_frames, 4, log_path.read_text()[-8000:])
|
|
time.sleep(1)
|
|
for attempt in range(3):
|
|
trace_start = len(log_path.read_text())
|
|
before = commits(surface)
|
|
time.sleep(2)
|
|
idle_frames = commits(surface) - before
|
|
if 'ferese::nested_input:' not in log_path.read_text()[trace_start:]:
|
|
break
|
|
else:
|
|
self.fail("Host input interrupted all three idle measurement intervals")
|
|
# Clock/status updates still redraw settled surfaces. Count
|
|
# commits: Iced can request two callbacks for one redraw.
|
|
self.assertLessEqual(idle_frames, 12, "settled toast kept requesting animation frames")
|
|
material_updates = log_path.read_text().count('.set_presentation(')
|
|
dbus("CloseNotification", notice)
|
|
wait_for(lambda: destroyed(surface), timeout=5)
|
|
self.assertGreater(log_path.read_text().count('.set_presentation(') - material_updates, 1,
|
|
"notification materials did not follow their fade")
|
|
print(f"Toast: {opening_frames} opening callback requests, {idle_frames / 2:g} settled commits/s; destroyed", flush=True)
|
|
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
modal = wait_for(lambda: layer_surface("ferese-system-modal"))
|
|
wait_for(lambda: frames(modal) >= 4)
|
|
time.sleep(.3)
|
|
before_close = frames(modal)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: destroyed(modal), timeout=5)
|
|
self.assertGreater(frames(modal), before_close, "modal close did not follow frames")
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
print(f"Modal: {frames(modal)} callbacks; interrupted opening closed and destroyed", flush=True)
|
|
|
|
# A toast can resize or start closing before its previous
|
|
# software buffer has caught up with the new viewport.
|
|
for index in range(40):
|
|
reply = dbus("Notify", "Ferese motion test", "0", "", "Rapid toast",
|
|
"Short body", "[]", "{}", "0").stdout
|
|
notice = re.search(r'uint32 (\d+)', reply).group(1)
|
|
if index % 2 == 0:
|
|
dbus("Notify", "Ferese motion test", notice, "", "Updated toast",
|
|
"A longer notification body that wraps over several lines. " * 6,
|
|
"[]", "{}", "0")
|
|
dbus("CloseNotification", notice)
|
|
time.sleep(.015)
|
|
time.sleep(.5)
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertEqual(dbus("GetServerInformation", check=False).returncode, 0)
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
print("Toast: 40 rapid open/update/close cycles completed", flush=True)
|
|
|
|
for _ in range(40):
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"],
|
|
env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
time.sleep(.03)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"],
|
|
env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
time.sleep(.06)
|
|
# A final toggle may leave the owned modal closing. Let
|
|
# that motion finish before deciding whether a live modal
|
|
# needs another toggle; toggling during closing reopens it.
|
|
deadline = time.monotonic() + 5
|
|
while True:
|
|
remaining = [surface for surface in layer_surfaces("ferese-system-modal")
|
|
if not destroyed(surface)]
|
|
if not remaining or time.monotonic() >= deadline:
|
|
break
|
|
time.sleep(.05)
|
|
self.assertLessEqual(len(remaining), 1, "cancelled modals retained native surfaces")
|
|
if remaining:
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"],
|
|
env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: destroyed(remaining[0]))
|
|
print("Modal: rapid toggles released all native surfaces", flush=True)
|
|
|
|
if bar_layout == "floating":
|
|
for layout_name, expected_count in [("flat", 1), ("floating", None)]:
|
|
config_file.write_text('animations { speed 0.5; }\nstatus { keybinding-guide #false; }\n' + panel_config(layout_name))
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: bar_region_count(bar) == (expected_count or 2))
|
|
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
print("Bar: separate material regions; flat/floating live switches passed", flush=True)
|
|
|
|
for padding in [4, 0]:
|
|
source = config_file.read_text()
|
|
config_file.write_text(re.sub(
|
|
r'island-padding \d+', f'island-padding {padding}', source))
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
expected_width = original_start_width - 2 * (12 - padding)
|
|
def start_width():
|
|
rectangles = [(x, width) for x, y, width, height in bar_input_rectangles(bar) if x < 300]
|
|
return max((x + width for x, width in rectangles), default=0) - min((x for x, width in rectangles), default=0)
|
|
wait_for(lambda: start_width() == expected_width)
|
|
|
|
print("Bar: live padding reductions updated the input regions", flush=True)
|
|
if settings_binary := os.environ.get("FERESE_TEST_SETTINGS"):
|
|
child_display = next(path for path in runtime.glob("wayland-*") if path.is_socket())
|
|
settings_env = dict(ipc_environment(env), WAYLAND_DISPLAY=str(child_display), WINIT_UNIX_BACKEND="wayland", ICED_BACKEND="wgpu")
|
|
settings_env.pop("WAYLAND_SOCKET", None)
|
|
settings_env.pop("FERESE_SHELL_CONTROL_SOCKET", None)
|
|
settings = subprocess.Popen([str(repo / settings_binary), "--page", "bar"], env=settings_env,
|
|
start_new_session=True, stdout=log, stderr=log)
|
|
children.append(settings)
|
|
def windows():
|
|
response = subprocess.run([str(ctl), "--json", "windows"], env=ipc_environment(env),
|
|
text=True, capture_output=True, check=True, timeout=5)
|
|
return json.loads(response.stdout)
|
|
wait_for(lambda: any(window["mapped"] and window["app_id"] == "dev.ferese.Settings" for window in windows()))
|
|
config_file.write_text('status { keybinding-guide #false; }\n' + panel_config("floating").replace('style "floating"', 'style "adaptive"'))
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: bar_region_count(bar) == 3)
|
|
for command, count in [("toggle-maximized", 1), ("toggle-maximized", 3)]:
|
|
subprocess.run([str(ctl), command], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: bar_region_count(bar) == count)
|
|
subprocess.run([str(ctl), "toggle-fullscreen"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: not bar_input_rectangles(bar))
|
|
subprocess.run([str(ctl), "toggle-fullscreen"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: bar_input_rectangles(bar))
|
|
settings.terminate()
|
|
settings.wait(timeout=5)
|
|
wait_for(lambda: not windows())
|
|
print("Adaptive: maximization changed the real panel; fullscreen released input and restored it", flush=True)
|
|
|
|
if bar_layout == "floating":
|
|
for style, regions in [("floating", 2), ("capsules", 2), ("notch", 2), ("flat", 1), ("fused", 2), ("adaptive", 2)]:
|
|
config_file.write_text('animations { speed 0.5; }\nstatus { keybinding-guide #false; }\n' + panel_config(style))
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: bar_region_count(bar) == regions)
|
|
self.assertTrue(bar_input_rectangles(bar), style + " lost its native input")
|
|
config_file.write_text('animations { speed 0.5; }\nstatus { keybinding-guide #false; }\nnotifications { timeout-ms 1200; }\n'
|
|
+ panel_config("morph").replace('style "morph"', 'style "morph"\n morph { peek-on-hover #false; }'))
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
# The visible hub follows its measured content; its larger
|
|
# backing surface remains allocated to avoid resize flashes.
|
|
wait_for(lambda: (size := bar_input_size(bar)) and 0 < size[0] < 320
|
|
and size[1] == 34 and bar_region_count(bar) == 1)
|
|
reply = dbus("Notify", "Morph test", "0", "", "Hub notification", "Bounded native surface", "[]", "{}", "0").stdout
|
|
notice = re.search(r'uint32 (\d+)', reply).group(1)
|
|
wait_for(lambda: bar_input_size(bar) == (320, 48))
|
|
allocation = bar_size(bar)
|
|
self.assertGreaterEqual(allocation[0], 320)
|
|
self.assertGreaterEqual(allocation[1], 48)
|
|
# Removing a notice must collapse its hub immediately; an old timeout cannot reopen it.
|
|
dbus("CloseNotification", notice)
|
|
wait_for(lambda: (size := bar_input_size(bar)) and 0 < size[0] < 320 and size[1] == 34)
|
|
self.assertEqual(bar_size(bar), allocation, "closing Morph resized its retained backing surface")
|
|
# Rounded protocol geometry can stop changing while subpixel
|
|
# spring motion or the content fade still needs frames.
|
|
wait_for(lambda: bar_motion_settled(bar))
|
|
before = commits(bar)
|
|
trace_start = len(log_path.read_text())
|
|
time.sleep(.5)
|
|
# Clock/service updates can cause a short render burst; settled geometry must not animate.
|
|
self.assertNotIn(".set_panel_presentation(", log_path.read_text()[trace_start:])
|
|
self.assertLessEqual(commits(bar) - before, 4, "settled Morph keeps scheduling frames")
|
|
self.assertEqual(bar_region_count(bar), 1)
|
|
config_file.write_text('status { keybinding-guide #false; }\n' + panel_config("floating"))
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env), check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: bar_region_count(bar) == 2)
|
|
print("Styles: native masks survived every live switch; Morph grew, closed and stopped frames", flush=True)
|
|
|
|
saved = config_file.read_text()
|
|
config_file.write_text('panel main { surface "invalid-surface"; }\n')
|
|
rejected = subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env),
|
|
capture_output=True, text=True, timeout=5)
|
|
self.assertNotEqual(rejected.returncode, 0, "invalid reload was accepted")
|
|
self.assertIn("invalid_config", rejected.stderr)
|
|
self.assertFalse(destroyed(bar), "invalid reload destroyed the current panel")
|
|
self.assertEqual(dbus("GetServerInformation", check=False).returncode, 0)
|
|
config_file.write_text(saved)
|
|
subprocess.run([str(ctl), "reload-config"], env=ipc_environment(env),
|
|
check=True, capture_output=True, timeout=5)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=ipc_environment(env),
|
|
check=True, capture_output=True, timeout=5)
|
|
recovered = wait_for(lambda: (surface := layer_surface("ferese-system-modal"))
|
|
and not destroyed(surface) and surface)
|
|
subprocess.run([str(ctl), "toggle-keybinding-guide"], env=ipc_environment(env),
|
|
check=True, capture_output=True, timeout=5)
|
|
wait_for(lambda: destroyed(recovered))
|
|
self.assertIsNone(compositor.poll(), log_path.read_text()[-8000:])
|
|
self.assertNotIn("panicked at", log_path.read_text())
|
|
print("Reload: invalid config preserved the panel; repaired config reopened and released a modal", flush=True)
|
|
finally:
|
|
if destination := os.environ.get("FERESE_TEST_LOG"):
|
|
shutil.copyfile(log_path, destination)
|
|
for child in reversed(children):
|
|
if child.poll() is None:
|
|
os.killpg(child.pid, signal.SIGTERM)
|
|
try:
|
|
child.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(child.pid, signal.SIGKILL)
|
|
child.wait(timeout=5)
|
|
if child.stdout is not None:
|
|
child.stdout.close()
|
|
if directory := os.environ.get("FERESE_TEST_LOG_DIR"):
|
|
logs = Path(directory)
|
|
logs.mkdir(parents=True, exist_ok=True)
|
|
shutil.copyfile(log_path, logs / f"shell-{bar_layout}.log")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|