Add more tests

Author: gonzalocasas

tests/ros2/test_core.py

@@ -0,0 +1,59 @@
+import pytest
+
+from roslibpy import Header, Time
+
+REF_FLOAT_SECS_TIME = 1610122759.677662
+
+
+def test_time_from_sec_based_on_time_module():
+    t = Time.from_sec(REF_FLOAT_SECS_TIME)
+    assert t.secs == 1610122759
+    assert t.nsecs == 677661895
+
+
+def test_to_nsec():
+    t = Time.from_sec(REF_FLOAT_SECS_TIME)
+    assert t.to_nsec() == 1610122759677661895
+
+
+def test_to_sec():
+    t = Time.from_sec(REF_FLOAT_SECS_TIME)
+    assert t.to_sec() == REF_FLOAT_SECS_TIME
+
+
+def test_is_zero():
+    assert Time(0, 0).is_zero()
+    assert Time(1, 0).is_zero() is False
+
+
+def test_header_ctor_supports_time():
+    header = Header(stamp=Time.from_sec(REF_FLOAT_SECS_TIME))
+    assert header["stamp"]["secs"] == 1610122759
+    assert header["stamp"]["secs"] == header["stamp"].secs
+    assert header["stamp"].to_sec() == REF_FLOAT_SECS_TIME
+
+
+def test_header_ctor_supports_dict():
+    header = Header(stamp=dict(secs=1610122759, nsecs=677661895))
+    assert header["stamp"]["secs"] == 1610122759
+    assert header["stamp"]["secs"] == header["stamp"].secs
+    assert header["stamp"].to_sec() == REF_FLOAT_SECS_TIME
+
+
+def test_time_accepts_only_ints():
+    with pytest.raises(ValueError):
+        Time(1.3, 1.0)
+    with pytest.raises(ValueError):
+        Time(100.0, 3.1)
+
+    t = Time(110.0, 0.0)
+    assert t.secs == 110
+    assert t.nsecs == 0
+
+
+def test_time_properties_are_readonly():
+    t = Time.now()
+    with pytest.raises(AttributeError):
+        t.secs = 10
+    with pytest.raises(AttributeError):
+        t.nsecs = 10

tests/ros2/test_ros.py

@@ -0,0 +1,95 @@
+from __future__ import print_function
+
+import threading
+import time
+
+from roslibpy import Ros
+
+host = "127.0.0.1"
+port = 9090
+url = "ws://%s:%d" % (host, port)
+
+
+def test_reconnect_does_not_trigger_on_client_close():
+    ros = Ros(host, port)
+    ros.run()
+
+    assert ros.is_connected, "ROS initially connected"
+    time.sleep(0.5)
+    event = threading.Event()
+    ros.on("close", lambda m: event.set())
+    ros.close()
+    event.wait(5)
+
+    assert not ros.is_connected, "Successful disconnect"
+    assert not ros.is_connecting, "Not trying to re-connect"
+
+
+def test_connection():
+    ros = Ros(host, port)
+    ros.run()
+    assert ros.is_connected
+    ros.close()
+
+
+def test_url_connection():
+    ros = Ros(url)
+    ros.run()
+    assert ros.is_connected
+    ros.close()
+
+
+def test_closing_event():
+    ros = Ros(url)
+    ros.run()
+    ctx = dict(closing_event_called=False, was_still_connected=False)
+
+    def handle_closing():
+        ctx["closing_event_called"] = True
+        ctx["was_still_connected"] = ros.is_connected
+        time.sleep(1.5)
+
+    ts_start = time.time()
+    ros.on("closing", handle_closing)
+    ros.close()
+    ts_end = time.time()
+    closing_was_handled_synchronously_before_close = ts_end - ts_start >= 1.5
+
+    assert ctx["closing_event_called"]
+    assert ctx["was_still_connected"]
+    assert closing_was_handled_synchronously_before_close
+
+
+def test_multithreaded_connect_disconnect():
+    CONNECTIONS = 30
+    clients = []
+
+    def connect(clients):
+        ros = Ros(url)
+        ros.run()
+        clients.append(ros)
+
+    # First connect all
+    threads = []
+    for _ in range(CONNECTIONS):
+        thread = threading.Thread(target=connect, args=(clients,))
+        thread.daemon = False
+        thread.start()
+        threads.append(thread)
+
+    for thread in threads:
+        thread.join()
+
+    # Assert connection status
+    for ros in clients:
+        assert ros.is_connected
+
+    # Now disconnect all
+    for ros in clients:
+        ros.close()
+
+    time.sleep(0.5)
+
+    # Assert connection status
+    for ros in clients:
+        assert not ros.is_connected

tests/ros2/test_rosapi.py

@@ -0,0 +1,47 @@
+import threading
+
+import pytest
+
+from roslibpy import Ros
+
+host = "127.0.0.1"
+port = 9090
+url = "ws://%s:%d" % (host, port)
+
+
+def test_rosapi_topics():
+    context = dict(wait=threading.Event(), result=None)
+    ros = Ros(host, port)
+    ros.run()
+
+    def callback(topic_list):
+        context["result"] = topic_list
+        context["wait"].set()
+
+    ros.get_topics(callback)
+    if not context["wait"].wait(5):
+        raise Exception
+
+    assert "/rosout" in context["result"]["topics"]
+    ros.close()
+
+
+def test_rosapi_topics_blocking():
+    ros = Ros(host, port)
+    ros.run()
+    topic_list = ros.get_topics()
+
+    print(topic_list)
+    assert "/rosout" in topic_list
+
+    ros.close()
+
+
+def test_connection_fails_when_missing_port():
+    with pytest.raises(Exception):
+        Ros(host)
+
+
+def test_connection_fails_when_schema_not_ws():
+    with pytest.raises(Exception):
+        Ros("http://%s:%d" % (host, port))