Merge pull request #811 from gwgeorgea/is-11-Adding-4-3-and-4-4

[IS-11] Adding Output test cases 4.3 and 4.4

nmostesting/Config.py

@@ -190,8 +190,18 @@
 # Bash shell to use for running testssl.sh
 TEST_SSL_BASH = "bash"
 
+# IP and port of IS-11 REFERENCE SENDER to test IS-11 Receivers under live streaming conditions.
+# The URL of a device API is required to access the IS-04 Node (http://{device}/x-nmos/node/v1.2),
+# IS-05 Connection (http://{device}/x-nmos/connection/v1.1) and
+# IS-11 Stream Compatibility (http://{device}/x-nmos/streamcompatibility/v1.0)
+IS11_REFERENCE_SENDER_CONNECTION_API = ""
+IS11_REFERENCE_SENDER_CONNECTION_API_PORT = ""
+IS11_REFERENCE_SENDER_NODE_API = ""
+IS11_REFERENCE_SENDER_NODE_API_PORT = ""
+
 # Stability delay for any request on IS-11
 STABLE_STATE_DELAY = 3
+
 # Definition of each API specification and its versions.
 SPECIFICATIONS = {
     "is-04": {

nmostesting/IS11Utils.py

@@ -12,36 +12,60 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from enum import Enum
+
 from . import TestHelper
 from .NMOSUtils import NMOSUtils
 
+SND_RCV_SUBSET = Enum('SndRcvSubset', ['ALL', 'WITH_I_O', 'WITHOUT_I_O'])
+
 
 class IS11Utils(NMOSUtils):
     def __init__(self, url):
         NMOSUtils.__init__(self, url)
 
     # TODO: Remove the duplication (IS05Utils)
-    def get_senders(self):
+    def get_senders(self, filter=SND_RCV_SUBSET.ALL):
         """Gets a list of the available senders on the API"""
         toReturn = []
         valid, r = TestHelper.do_request("GET", self.url + "senders/")
         if valid and r.status_code == 200:
             try:
                 for value in r.json():
-                    toReturn.append(value[:-1])
+                    if filter == SND_RCV_SUBSET.ALL:
+                        toReturn.append(value[:-1])
+                    else:
+                        valid_io, r_io = TestHelper.do_request("GET", self.url + "senders/" + value + "inputs/")
+                        if valid_io and r_io.status_code == 200:
+                            try:
+                                if len(r_io.json()) > 0 and filter == SND_RCV_SUBSET.WITH_I_O or \
+                                   len(r_io.json()) == 0 and filter == SND_RCV_SUBSET.WITHOUT_I_O:
+                                    toReturn.append(value[:-1])
+                            except ValueError:
+                                pass
             except ValueError:
                 pass
         return toReturn
 
     # TODO: Remove the duplication (IS05Utils)
-    def get_receivers(self):
+    def get_receivers(self, filter=SND_RCV_SUBSET.ALL):
         """Gets a list of the available receivers on the API"""
         toReturn = []
         valid, r = TestHelper.do_request("GET", self.url + "receivers/")
         if valid and r.status_code == 200:
             try:
                 for value in r.json():
-                    toReturn.append(value[:-1])
+                    if filter == SND_RCV_SUBSET.ALL:
+                        toReturn.append(value[:-1])
+                    else:
+                        valid_io, r_io = TestHelper.do_request("GET", self.url + "receivers/" + value + "outputs/")
+                        if valid_io and r_io.status_code == 200:
+                            try:
+                                if len(r_io.json()) > 0 and filter == SND_RCV_SUBSET.WITH_I_O or \
+                                   len(r_io.json()) == 0 and filter == SND_RCV_SUBSET.WITHOUT_I_O:
+                                    toReturn.append(value[:-1])
+                            except ValueError:
+                                pass
             except ValueError:
                 pass
         return toReturn
@@ -96,3 +120,19 @@ def checkCleanRequestJSON(self, method, dest, data=None, code=200):
                 return False, "Invalid JSON received"
         else:
             return valid, response
+
+    def get_transportfile(self, url, sender_id):
+        """Get the transport file for a given Sender"""
+        toReturn = None
+        valid, r = TestHelper.do_request("GET", url + "single/senders/" + sender_id + "/transportfile/")
+        if valid and r.status_code == 200:
+            toReturn = r.text
+        return toReturn
+
+    def get_flows(self, url, sender_id):
+        """Get the flow for a given Sender"""
+        toReturn = None
+        valid, r = TestHelper.do_request("GET", url + "flows/" + sender_id)
+        if valid and r.status_code == 200:
+            toReturn = r.json()
+        return toReturn