Add tests for generate_policy verb (#122)

* Add tests for generate_policy verb

The verbs are missing tests. Let's get started by adding some basic
coverage to `generate_policy`. Also ensure that `generate_policy` abides
by established conventions, returning 0 upon success.

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

* Add std_msgs as a test_depend

Also use `1` instead of QoSProfile.

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

* Reorder imports

Signed-off-by: Kyle Fazzari <kyle@canonical.com>

.gitignore

@@ -1 +1,3 @@
 __pycache__
+.eggs/
+.coverage

sros2/package.xml

@@ -20,6 +20,7 @@
   <test_depend>ament_flake8</test_depend>
   <test_depend>ament_pep257</test_depend>
   <test_depend>python3-pytest</test_depend>
+  <test_depend>std_msgs</test_depend>
 
   <export>
     <build_type>ament_python</build_type>

sros2/sros2/verb/generate_policy.py

@@ -141,3 +141,4 @@ def main(self, *, args):
 
         with open(args.POLICY_FILE_PATH, 'w') as stream:
             dump_policy(policy, stream)
+        return 0

sros2/test/sros2/commands/security/verbs/test_generate_policy.py

@@ -0,0 +1,78 @@
+# Copyright 2019 Canonical Ltd
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import tempfile
+
+import pytest
+
+import rclpy
+from ros2cli import cli
+from sros2.policy import load_policy
+from std_msgs.msg import String
+
+
+def test_generate_policy():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        # Create a test-specific context so that generate_policy can still init
+        context = rclpy.Context()
+        rclpy.init(context=context)
+        node = rclpy.create_node('test_node', context=context)
+
+        try:
+            # Create a publisher and subscription
+            node.create_publisher(String, 'topic_pub', 1)
+            node.create_subscription(String, 'topic_sub', lambda msg: None, 1)
+
+            # Generate the policy for the running node
+            assert cli.main(
+                argv=['security', 'generate_policy', os.path.join(tmpdir, 'test-policy.xml')]) == 0
+        finally:
+            node.destroy_node()
+            rclpy.shutdown(context=context)
+
+        # Load the policy and pull out the allowed publications and subscriptions
+        policy = load_policy(os.path.join(tmpdir, 'test-policy.xml'))
+        profile = policy.find(path='profiles/profile[@ns="/"][@node="test_node"]')
+        assert profile is not None
+        topics_publish_allowed = profile.find(path='topics[@publish="ALLOW"]')
+        assert topics_publish_allowed is not None
+        topics_subscribe_allowed = profile.find(path='topics[@subscribe="ALLOW"]')
+        assert topics_subscribe_allowed is not None
+
+        # Verify that the allowed publications include topic_pub and not topic_sub
+        topics = topics_publish_allowed.findall('topic')
+        assert len([t for t in topics if t.text == 'topic_pub']) == 1
+        assert len([t for t in topics if t.text == 'topic_sub']) == 0
+
+        # Verify that the allowed subscriptions include topic_sub and not topic_pub
+        topics = topics_subscribe_allowed.findall('topic')
+        assert len([t for t in topics if t.text == 'topic_sub']) == 1
+        assert len([t for t in topics if t.text == 'topic_pub']) == 0
+
+
+def test_generate_policy_no_nodes(capsys):
+    with tempfile.TemporaryDirectory() as tmpdir:
+        assert cli.main(argv=[
+            'security', 'generate_policy', os.path.join(tmpdir, 'test-policy.xml')]) != 0
+        stderr = capsys.readouterr().err.strip()
+        assert stderr == 'No nodes detected in the ROS graph. No policy file was generated.'
+
+
+def test_generate_policy_no_policy_file(capsys):
+    with pytest.raises(SystemExit) as e:
+        cli.main(argv=['security', 'generate_policy'])
+        assert e.value.code != 0
+    stderr = capsys.readouterr().err.strip()
+    assert 'following arguments are required: POLICY_FILE_PATH' in stderr