Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix launch warnings #421

Merged
merged 3 commits into from
May 14, 2020
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion launch_testing/launch_testing/pytest/hooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -123,6 +123,19 @@ def reportinfo(self):

class LaunchTestModule(pytest.File):

def __init__(self, parent, *, fspath):
super().__init__(parent=parent, fspath=fspath)

@classmethod
def from_parent(cls, parent, *, fspath):
"""Override from_parent for compatibility."""
# pytest.File.from_parent didn't exist before pytest 5.4
if hasattr(super(), 'from_parent'):
instance = getattr(super(), 'from_parent')(parent=parent, fspath=fspath)
else:
instance = cls(parent=parent, fspath=fspath)
return instance

def makeitem(self, *args, **kwargs):
return LaunchTestItem.from_parent(*args, **kwargs)

Expand Down Expand Up @@ -171,7 +184,7 @@ def pytest_pycollect_makemodule(path, parent):
def pytest_launch_collect_makemodule(path, parent, entrypoint):
marks = getattr(entrypoint, 'pytestmark', [])
if marks and any(m.name == 'launch_test' for m in marks):
return LaunchTestModule(path, parent)
return LaunchTestModule.from_parent(parent, fspath=path)


def pytest_addhooks(pluginmanager):
Expand Down
1 change: 1 addition & 0 deletions launch_testing/pytest.ini
Original file line number Diff line number Diff line change
Expand Up @@ -3,3 +3,4 @@
testpaths = test
# Add arguments for launch tests
addopts = --launch-args dut_arg:=test
junit_family=xunit2
46 changes: 22 additions & 24 deletions launch_testing/test/launch_testing/examples/args_launch_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -29,23 +29,21 @@
import pytest


dut_process = launch.actions.ExecuteProcess(
cmd=[
sys.executable,
os.path.join(
ament_index_python.get_package_prefix('launch_testing'),
'lib/launch_testing',
'terminating_proc',
),

# Arguments
launch.substitutions.LaunchConfiguration('dut_arg')
],
)


@pytest.mark.launch_test
def generate_test_description():
dut_process = launch.actions.ExecuteProcess(
cmd=[
sys.executable,
os.path.join(
ament_index_python.get_package_prefix('launch_testing'),
'lib/launch_testing',
'terminating_proc',
),

# Arguments
launch.substitutions.LaunchConfiguration('dut_arg')
],
)

return launch.LaunchDescription([

Expand All @@ -65,36 +63,36 @@ def generate_test_description():
launch_testing.util.KeepAliveProc(),

launch_testing.actions.ReadyToTest()
])
]), {'dut_process': dut_process}


class TestTerminatingProcessStops(unittest.TestCase):

def test_proc_terminates(self):
self.proc_info.assertWaitForShutdown(process=dut_process, timeout=10)
def test_proc_terminates(self, proc_info, dut_process):
proc_info.assertWaitForShutdown(process=dut_process, timeout=10)


@launch_testing.post_shutdown_test()
class TestProcessOutput(unittest.TestCase):

def test_ran_with_arg(self):
def test_ran_with_arg(self, dut_process):
self.assertNotIn(
'default',
dut_process.process_details['cmd'],
'Try running: launch_test test_with_args.test.py dut_arg:=arg'
)

def test_arg_printed_in_output(self):
def test_arg_printed_in_output(self, proc_output, test_args, dut_process):
launch_testing.asserts.assertInStdout(
self.proc_output,
self.test_args['dut_arg'],
proc_output,
test_args['dut_arg'],
dut_process
)

def test_default_not_printed(self):
def test_default_not_printed(self, proc_output, dut_process):
with self.assertRaises(AssertionError):
launch_testing.asserts.assertInStdout(
self.proc_output,
proc_output,
'default',
dut_process
)
Original file line number Diff line number Diff line change
Expand Up @@ -68,20 +68,20 @@ def generate_test_description():

class TestProcOutput(unittest.TestCase):

def test_process_output(self, dut):
def test_process_output(self, launch_service, proc_info, proc_output, dut):
# We can use the 'dut' argument here because it's part of the test context
# returned by `generate_test_description` It's not necessary for every
# test to use every piece of the context
self.proc_output.assertWaitFor('Loop 1', process=dut, timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 1', process=dut, timeout=10, stream='stdout')


@launch_testing.post_shutdown_test()
class TestProcessOutput(unittest.TestCase):

def test_full_output(self, dut):
def test_full_output(self, proc_output, dut):
# Same as the test_process_output test. launch_testing binds the value of
# 'dut' from the test_context to the test before it runs
with assertSequentialStdout(self.proc_output, process=dut) as cm:
with assertSequentialStdout(proc_output, process=dut) as cm:
cm.assertInStdout('Starting Up')
if os.name != 'nt':
# On Windows, process termination is always forced
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -28,58 +28,56 @@
import pytest


TEST_PROC_PATH = os.path.join(
ament_index_python.get_package_prefix('launch_testing'),
'lib/launch_testing',
'good_proc'
)

# This is necessary to get unbuffered output from the process under test
proc_env = os.environ.copy()
proc_env['PYTHONUNBUFFERED'] = '1'

dut_process = launch.actions.ExecuteProcess(
cmd=[sys.executable, TEST_PROC_PATH],
env=proc_env, output='screen'
)


@pytest.mark.launch_test
def generate_test_description():
TEST_PROC_PATH = os.path.join(
ament_index_python.get_package_prefix('launch_testing'),
'lib/launch_testing',
'good_proc'
)

# This is necessary to get unbuffered output from the process under test
proc_env = os.environ.copy()
proc_env['PYTHONUNBUFFERED'] = '1'

dut_process = launch.actions.ExecuteProcess(
cmd=[sys.executable, TEST_PROC_PATH],
env=proc_env, output='screen'
)

return launch.LaunchDescription([
dut_process,

# Start tests right away - no need to wait for anything
launch_testing.actions.ReadyToTest(),
])
]), {'dut_process': dut_process}


# These tests will run concurrently with the dut process. After all these tests are done,
# the launch system will shut down the processes that it started up
class TestGoodProcess(unittest.TestCase):

def test_count_to_four(self):
def test_count_to_four(self, proc_output):
# This will match stdout from any process. In this example there is only one process
# running
self.proc_output.assertWaitFor('Loop 1', timeout=10, stream='stdout')
self.proc_output.assertWaitFor('Loop 2', timeout=10, stream='stdout')
self.proc_output.assertWaitFor('Loop 3', timeout=10, stream='stdout')
self.proc_output.assertWaitFor('Loop 4', timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 1', timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 2', timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 3', timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 4', timeout=10, stream='stdout')


@launch_testing.post_shutdown_test()
class TestProcessOutput(unittest.TestCase):

def test_exit_code(self):
def test_exit_code(self, proc_info):
# Check that all processes in the launch (in this case, there's just one) exit
# with code 0
launch_testing.asserts.assertExitCodes(self.proc_info)
launch_testing.asserts.assertExitCodes(proc_info)

def test_full_output(self):
def test_full_output(self, proc_output, dut_process):
# Using the SequentialStdout context manager asserts that the following stdout
# happened in the same order that it's checked
with assertSequentialStdout(self.proc_output, dut_process) as cm:
with assertSequentialStdout(proc_output, dut_process) as cm:
cm.assertInStdout('Starting Up')
for n in range(4):
cm.assertInStdout('Loop {}'.format(n))
Expand All @@ -88,10 +86,10 @@ def test_full_output(self):
# and thus the last print in good_proc never makes it.
cm.assertInStdout('Shutting Down')

def test_out_of_order(self):
def test_out_of_order(self, proc_output, dut_process):
# This demonstrates that we notice out-of-order IO
with self.assertRaisesRegex(AssertionError, "'Loop 2' not found"):
with assertSequentialStdout(self.proc_output, dut_process) as cm:
with assertSequentialStdout(proc_output, dut_process) as cm:
cm.assertInStdout('Loop 1')
cm.assertInStdout('Loop 3')
cm.assertInStdout('Loop 2') # This should raise
22 changes: 11 additions & 11 deletions launch_testing/test/launch_testing/examples/ready_action_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,27 +57,27 @@ def generate_test_description():
# the launch system will shut down the processes that it started up
class TestGoodProcess(unittest.TestCase):

def test_count_to_four(self):
def test_count_to_four(self, proc_output):
# This will match stdout from any process. In this example there is only one process
# running
self.proc_output.assertWaitFor('Loop 1', timeout=10, stream='stdout')
self.proc_output.assertWaitFor('Loop 2', timeout=10, stream='stdout')
self.proc_output.assertWaitFor('Loop 3', timeout=10, stream='stdout')
self.proc_output.assertWaitFor('Loop 4', timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 1', timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 2', timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 3', timeout=10, stream='stdout')
proc_output.assertWaitFor('Loop 4', timeout=10, stream='stdout')


@launch_testing.post_shutdown_test()
class TestProcessOutput(unittest.TestCase):

def test_exit_code(self):
def test_exit_code(self, proc_info):
# Check that all processes in the launch (in this case, there's just one) exit
# with code 0
launch_testing.asserts.assertExitCodes(self.proc_info)
launch_testing.asserts.assertExitCodes(proc_info)

def test_full_output(self, dut_process):
def test_full_output(self, proc_output, dut_process):
# Using the SequentialStdout context manager asserts that the following stdout
# happened in the same order that it's checked
with assertSequentialStdout(self.proc_output, dut_process) as cm:
with assertSequentialStdout(proc_output, dut_process) as cm:
cm.assertInStdout('Starting Up')
for n in range(4):
cm.assertInStdout('Loop {}'.format(n))
Expand All @@ -86,10 +86,10 @@ def test_full_output(self, dut_process):
# and thus the last print in good_proc never makes it.
cm.assertInStdout('Shutting Down')

def test_out_of_order(self, dut_process):
def test_out_of_order(self, proc_output, dut_process):
# This demonstrates that we notice out-of-order IO
with self.assertRaisesRegex(AssertionError, "'Loop 2' not found"):
with assertSequentialStdout(self.proc_output, dut_process) as cm:
with assertSequentialStdout(proc_output, dut_process) as cm:
cm.assertInStdout('Loop 1')
cm.assertInStdout('Loop 3')
cm.assertInStdout('Loop 2') # This should raise