Merge pull request #123 from davidlatwe/fix/minor-fixes

Minor fixes and adding tests for those fixes

allzpark/_rezapi.py

@@ -47,7 +47,7 @@ def find_latest(name, range_=None, paths=None):
 
 
 try:
-    from rez import project
+    from rez import __project__ as project
 except ImportError:
     # nerdvegas/rez
     project = "rez"

allzpark/control.py

@@ -844,7 +844,7 @@ def list_profiles(self, root=None):
                 profiles = root()
 
             except Exception:
-                if log.level == logging.DEBUG:
+                if log.level < logging.INFO:
                     traceback.print_exc()
 
                 self.error("Could not find profiles in %s" % root)

azure-pipelines.yml

@@ -102,6 +102,7 @@ jobs:
         displayName: "Setup Xvfb"
 
       - script: |
+          export DISPLAY=:99
           xvfb-run sudo nosetests
         displayName: "Run tests"
 

tests/test_apps.py

@@ -31,11 +31,11 @@ def test_select_app(self):
                 }
             },
         })
-        self.ctrl.reset(["foo"])
-        util.wait(self.ctrl.resetted)
+        with util.wait_signal(self.ctrl.resetted):
+            self.ctrl.reset(["foo"])
 
-        self.ctrl.select_profile("foo")
-        util.wait(self.ctrl.state_changed, "ready")
+        with util.wait_signal(self.ctrl.state_changed, "ready"):
+            self.ctrl.select_profile("foo")
 
         env = self.ctrl.state["rezEnvirons"]
 
@@ -80,11 +80,11 @@ def test_app_environ(self):
                 }
             },
         })
-        self.ctrl.reset(["foo"])
-        util.wait(self.ctrl.resetted)
+        with util.wait_signal(self.ctrl.resetted):
+            self.ctrl.reset(["foo"])
 
-        self.ctrl.select_profile("foo")
-        util.wait(self.ctrl.state_changed, "ready")
+        with util.wait_signal(self.ctrl.state_changed, "ready"):
+            self.ctrl.select_profile("foo")
 
         env = self.ctrl.state["rezEnvirons"]
 

tests/test_launch.py

@@ -0,0 +1,56 @@
+
+import sys
+from tests import util
+
+
+class TestLaunch(util.TestBase):
+
+    def test_launch_subprocess(self):
+        """Test launching subprocess command"""
+        util.memory_repository({
+            "foo": {
+                "1": {
+                    "name": "foo",
+                    "version": "1",
+                    "requires": ["~app"],
+                }
+            },
+            "app": {
+                "1": {
+                    "name": "app",
+                    "version": "1",
+                }
+            },
+        })
+        with util.wait_signal(self.ctrl.resetted):
+            self.ctrl.reset(["foo"])
+
+        with util.wait_signal(self.ctrl.state_changed, "ready"):
+            self.ctrl.select_profile("foo")
+
+        self.ctrl.select_application("app==1")
+        self.assertEqual("app==1", self.ctrl.state["appRequest"])
+
+        commands = self.ctrl.state["commands"]
+        self.assertEqual(len(commands), 0)
+
+        stdout = list()
+        stderr = list()
+        command = (
+          '%s -c "'
+          'import sys;'
+          'sys.stdout.write(\'meow\')"'
+        ) % sys.executable
+
+        with util.wait_signal(self.ctrl.state_changed, "launching"):
+            self.ctrl.launch(command=command,
+                             stdout=lambda m: stdout.append(m),
+                             stderr=lambda m: stderr.append(m))
+
+        self.assertEqual(len(commands), 1)
+
+        with util.wait_signal(commands[0].killed):
+            pass
+
+        self.assertIn("meow", "\n".join(stdout))
+        self.assertEqual("", "\n".join(stderr))

tests/test_profiles.py

@@ -1,4 +1,5 @@
 
+from unittest import mock
 from tests import util
 
 
@@ -21,9 +22,8 @@ def test_reset(self):
                 }
             }
         })
-        self.ctrl.reset(["foo", "bar"])
-
-        util.wait(self.ctrl.resetted)
+        with util.wait_signal(self.ctrl.resetted):
+            self.ctrl.reset(["foo", "bar"])
 
         # last profile will be selected by default
         self.assertEqual("bar", self.ctrl.state["profileName"])
@@ -46,12 +46,13 @@ def test_select_profile_with_out_apps(self):
                 }
             }
         })
-        self.ctrl.reset(["foo", "bar"])
-        util.wait(self.ctrl.resetted)
+        with util.wait_signal(self.ctrl.resetted):
+            self.ctrl.reset(["foo", "bar"])
+
+        with util.wait_signal(self.ctrl.state_changed, "noapps"):
+            self.ctrl.select_profile("foo")
+            # wait enter 'noapps' state
 
-        self.ctrl.select_profile("foo")
-        # enter 'noapps' state
-        util.wait(self.ctrl.state_changed, "noapps")
         self.assertEqual("foo", self.ctrl.state["profileName"])
 
     def test_profile_list_apps(self):
@@ -88,15 +89,51 @@ def test_profile_list_apps(self):
                 }
             },
         })
-        self.ctrl.reset(["foo"])
-        util.wait(self.ctrl.resetted)
+        with util.wait_signal(self.ctrl.resetted):
+            self.ctrl.reset(["foo"])
+
+        with util.wait_signal(self.ctrl.state_changed, "ready"):
+            self.ctrl.select_profile("foo")
 
-        self.ctrl.select_profile("foo")
-        util.wait(self.ctrl.state_changed, "ready")
         self.assertEqual(
             [
                 "app_A==1.0.0",
                 "app_B==1.0.0",
             ],
             list(self.ctrl.state["rezApps"].keys())
         )
+
+    def test_profile_listing_without_root_err(self):
+        """Listing profile without root will raise AssertionError"""
+        self.assertRaises(AssertionError, self.ctrl.reset)
+        self.assertRaises(AssertionError, self.ctrl.list_profiles)
+
+    def test_profile_listing_callable_root_err(self):
+        """Listing profile with bad callable will prompt error message"""
+        import traceback
+        import logging
+        from allzpark import control
+
+        traceback.print_exc = mock.MagicMock(name="traceback.print_exc")
+        self.ctrl.error = mock.MagicMock(name="Controller.error")
+
+        def bad_root():
+            raise Exception("This should be caught.")
+        self.ctrl.list_profiles(bad_root)
+
+        # ctrl.error must be called in all cases
+        self.ctrl.error.assert_called_once()
+        # traceback.print_exc should be called if logging level is set
+        # lower than INFO, e.g. DEBUG or NOTSET
+        if control.log.level < logging.INFO:
+            traceback.print_exc.assert_called_once()
+
+    def test_profile_listing_invalid_type_root_err(self):
+        """Listing profile with invalid input type will raise TypeError"""
+        self.assertRaises(TypeError, self.ctrl.list_profiles, {"foo"})
+
+    def test_profile_listing_filter_out_empty_names(self):
+        """Listing profile with empty names will be filtered"""
+        expected = ["foo", "bar"]
+        profiles = self.ctrl.list_profiles(expected + [None, ""])
+        self.assertEqual(profiles, expected)

tests/util.py

@@ -1,6 +1,7 @@
 
 import os
 import unittest
+import contextlib
 
 
 MEMORY_LOCATION = "memory@any"
@@ -34,30 +35,47 @@ def memory_repository(packages):
     repository.data = packages
 
 
-def wait(signal=None, on_value=None, timeout=1000):
+def wait(timeout=1000):
     from allzpark.vendor.Qt import QtCore
 
     loop = QtCore.QEventLoop()
     timer = QtCore.QTimer()
-    state = {"timeout": False}
 
-    if on_value:
+    timer.timeout.connect(loop.quit)
+    timer.start(timeout)
+    loop.exec_()
+
+
+@contextlib.contextmanager
+def wait_signal(signal, on_value=None, timeout=1000):
+    from allzpark.vendor.Qt import QtCore
+
+    loop = QtCore.QEventLoop()
+    timer = QtCore.QTimer()
+    state = {"received": False}
+
+    if on_value is None:
+        def trigger(*args):
+            state["received"] = True
+            timer.stop()
+            loop.quit()
+    else:
         def trigger(value):
             if value == on_value:
+                state["received"] = True
+                timer.stop()
                 loop.quit()
-                state["timeout"] = False
-    else:
-        def trigger(*args):
-            loop.quit()
-            state["timeout"] = False
 
-    if signal is not None:
-        state["timeout"] = True
-        signal.connect(trigger)
-    timer.timeout.connect(loop.quit)
+    def on_timeout():
+        loop.quit()
+        raise Exception("Signal waiting timeout.")
 
-    timer.start(timeout)
-    loop.exec_()
+    signal.connect(trigger)
+    timer.timeout.connect(on_timeout)
 
-    if state["timeout"]:
-        raise Exception("Signal waiting timeout.")
+    try:
+        yield
+    finally:
+        if not state["received"]:
+            timer.start(timeout)
+            loop.exec_()