wehub-resource-sync
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
import platform
|
|
import tempfile
|
|
from typing import TYPE_CHECKING
|
|
|
|
import rerun as rr
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
|
|
def test_init_twice() -> None:
|
|
"""Regression test for #9948: creating the same recording twice caused hangs in the past (should instead warn)."""
|
|
# Always set strict mode to false in case it leaked from another test
|
|
rr.set_strict_mode(False)
|
|
|
|
# Using default recording id
|
|
rr.init("rerun_example_test_app_id")
|
|
recording_id = rr.get_recording_id()
|
|
|
|
rr.init("rerun_example_test_app_id")
|
|
assert recording_id == rr.get_recording_id()
|
|
|
|
# Using a custom recording id
|
|
recording_id = "test_recording_id"
|
|
rr.init("rerun_example_test_app_id", recording_id=recording_id)
|
|
rr.init("rerun_example_test_app_id", recording_id=recording_id)
|
|
assert recording_id == rr.get_recording_id()
|
|
|
|
|
|
def test_recording_stream_twice() -> None:
|
|
"""For `RecordingStream`, the default should be to have new, unique recording ids."""
|
|
|
|
with rr.RecordingStream("rerun_example_bug") as rec:
|
|
id1 = rec.get_recording_id()
|
|
|
|
with rr.RecordingStream("rerun_example_bug") as rec:
|
|
id2 = rec.get_recording_id()
|
|
|
|
assert id1 != id2
|
|
|
|
|
|
def test_isolated_streams(tmp_path: Path) -> None:
|
|
rec1_path = f"{tmp_path}/rec1.rrd"
|
|
rec2_path = f"{tmp_path}/rec2.rrd"
|
|
|
|
rec1 = rr.RecordingStream("rerun_example_multi_stream", recording_id="rec1")
|
|
rec1.log("/data1", rr.TextLog("Data1"))
|
|
rec1.save(rec1_path)
|
|
|
|
rec2 = rr.RecordingStream("rerun_example_multi_stream", recording_id="rec2")
|
|
rec2.log("/data2", rr.TextLog("Data2"))
|
|
rec2.save(rec2_path)
|
|
|
|
server = rr.server.Server(datasets={"test_dataset": tmp_path})
|
|
ds = server.client().get_dataset("test_dataset")
|
|
assert ds.filter_segments("rec1").filter_contents("/data1").reader(index="log_time").count() == 1
|
|
assert ds.filter_segments("rec2").filter_contents("/data2").reader(index="log_time").count() == 1
|
|
assert (
|
|
ds.filter_segments(["rec1", "rec2"]).filter_contents(["/data1", "/data2"]).reader(index="log_time").count() == 2
|
|
)
|
|
|
|
|
|
def test_cleanup_reinit() -> None:
|
|
system = platform.system()
|
|
if system == "Linux":
|
|
import os
|
|
|
|
def is_file_open(file_path: str) -> bool:
|
|
file_path = os.path.realpath(file_path)
|
|
|
|
fd_dir = "/proc/self/fd"
|
|
if not os.path.isdir(fd_dir):
|
|
return False # Only works on platforms with /proc
|
|
|
|
try:
|
|
for fd in os.listdir(fd_dir):
|
|
try:
|
|
target = os.readlink(os.path.join(fd_dir, fd))
|
|
if os.path.samefile(file_path, target):
|
|
return True
|
|
except (FileNotFoundError, PermissionError, OSError):
|
|
continue
|
|
except Exception:
|
|
pass
|
|
|
|
return False
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
rec_path = f"{tmpdir}/my_rec.rrd"
|
|
rr.init("rerun_example_reinit")
|
|
rr.save(rec_path)
|
|
rr.log("/data1", rr.TextLog("Data1"))
|
|
|
|
assert is_file_open(rec_path), "Recording file should be open after saving"
|
|
|
|
rr.init("rerun_example_reinit")
|
|
|
|
assert not is_file_open(rec_path), "Recording file should be closed after calling rr.init"
|