DX2913, DX2916, DX2917

Author: brianluisgomez

bandwidth/model/bxml/verbs/__init__.py

@@ -3,5 +3,8 @@
 from .phone_number import PhoneNumber
 from .record import Record
 from .sip_uri import SipUri
+from .start_stream import StartStream
+from .stop_stream import StopStream
+from .stream_param import StreamParam
 from .tag import Tag
 from .transfer import Transfer

bandwidth/model/bxml/verbs/pause_recording.py

@@ -13,4 +13,4 @@ class PauseRecording(TerminalVerb):
     def __init__(self):
         """Initialize a <PauseRecording> verb
         """
-        super().__init__(tag="PauseRecording", content=None, attributes=None)
+        super().__init__(tag="PauseRecording", content=None)

bandwidth/model/bxml/verbs/start_stream.py

@@ -0,0 +1,62 @@
+"""
+start_stream.py
+
+Bandwidth's StartStream BXML verb
+
+@copyright Bandwidth INC
+"""
+from typing import List
+
+from ..verb import Verb
+from ..verbs.stream_param import StreamParam
+
+
+class StartStream(Verb):
+
+    def __init__(
+        self, destination: str, stream_params: List[StreamParam] = [],
+        name: str=None, tracks: str=None,
+        stream_event_url: str=None,
+        stream_event_method: str=None,
+        username: str=None, password: str=None,
+    ):
+        """Initialize a <Transfer> verb
+
+        Args:
+            name (str, optional): A name to refer to this stream by. Used when sending <StopStream>. If not provided, it will default to the generated stream id as sent in the Media Stream Started webhook.
+            tracks (str, optional): The part of the call to send a stream from. inbound, outbound or both. Default is inbound.
+            destination (str, optional): A websocket URI to send the stream to. The audio from the specified tracks will be sent via websocket to this URL as base64-encoded PCMU/G711 audio. See below for more details on the websocket packet format.
+            stream_event_url (str, optional): URL to send the associated Webhook events to during this stream's lifetime. Does not accept BXML. May be a relative URL.
+            stream_event_method (str, optional): The HTTP method to use for the request to streamEventUrl. GET or POST. Default value is POST.
+            username (str, optional): The username to send in the HTTP request to streamEventUrl. If specified, the URLs must be TLS-encrypted (i.e., https).
+            password (str, optional): The password to send in the HTTP request to streamEventUrl. If specified, the URLs must be TLS-encrypted (i.e., https).
+        
+        Nested Verbs:
+            StreamParam: (optional) You may specify up to 12 <StreamParam/> elements nested within a <StartStream> tag. These elements define optional user specified parameters that will be sent to the destination URL when the stream is first started.       
+
+        """
+        self.destination = destination
+        self.stream_params = stream_params
+        self.name = name
+        self.tracks = tracks
+        self.stream_event_url = stream_event_url
+        self.stream_event_method = stream_event_method
+        self.username = username
+        self.password = password
+        super().__init__(
+            tag="StartStream",
+            content=None,
+            nested_verbs=self.stream_params
+        )
+
+    @property
+    def _attributes(self):
+        return {
+            "destination": self.destination,
+            "name": self.name,
+            "tracks": self.tracks,
+            "streamEventUrl": self.stream_event_url,
+            "streamEventMethod": self.stream_event_method,
+            "username": self.username,
+            "password": self.password,
+        }

bandwidth/model/bxml/verbs/stop_stream.py

@@ -0,0 +1,31 @@
+"""
+stop_stream.py
+
+Bandwidth's StopStream BXML verb
+
+@copyright Bandwidth INC
+"""
+from ..terminal_verb import TerminalVerb
+
+
+class StopStream(TerminalVerb):
+
+    def __init__(
+        self, name: str
+    ):
+        """Initialize a <StopStream> verb
+
+        Args:
+            name (str): The name of the stream to stop. This is either the user selected name when sending the <StartStream> verb, or the system generated name returned in the Media Stream Started webhook if <StartStream> was sent with no name attribute.
+        """
+        self.name = name
+        super().__init__(
+            tag="StopStream",
+            content=None
+        )
+
+    @property
+    def _attributes(self):
+        return {
+            "name": self.name,
+        }

bandwidth/model/bxml/verbs/stream_param.py

@@ -0,0 +1,34 @@
+"""
+stream_param.py
+
+Bandwidth's StreamParam BXML verb
+
+@copyright Bandwidth INC
+"""
+from ..terminal_verb import TerminalVerb
+
+
+class StreamParam(TerminalVerb):
+
+    def __init__(
+        self, name: str, value: str
+    ):
+        """Initialize a <StreamParam> verb
+
+        Args:
+            name (str): The name of this parameter, up to 256 characters.
+            value (str): The value of this parameter, up to 2048 characters.
+        """
+        self.name = name
+        self.value = value
+        super().__init__(
+            tag="StreamParam",
+            content=None
+        )
+
+    @property
+    def _attributes(self):
+        return {
+            "name": self.name,
+            "value": self.value,
+        }

test/unit/bxml/test_start_stream.py

@@ -0,0 +1,51 @@
+"""
+test_start_stream.py
+
+Unit tests for the <StartStream> BXML verb
+
+@copyright Bandwidth Inc.
+"""
+import os
+import pytest
+import unittest
+
+from bandwidth.model.bxml.verbs import StartStream, StreamParam
+
+class TestStartStream(unittest.TestCase):
+    
+    def setUp(self):
+        self.stream_param1 = StreamParam(
+            name="name1",
+            value="value1"
+        )
+
+        self.stream_param2 = StreamParam(
+            name="name2",
+            value="value2"
+        )
+
+        self.start_stream = StartStream(
+            stream_params=[self.stream_param1],
+            name = "stream1",
+            tracks = "inbound",
+            destination = "testurl.com",
+            stream_event_url="eventurl.com",
+            stream_event_method= "POST",
+            username = "user",
+            password = "pass"
+        )
+    
+    def test_to_bxml(self):
+        if os.environ['PYTHON_VERSION'] == '3.7':
+            expected = '<StartStream destination="testurl.com" name="stream1" tracks="inbound" streamEventUrl="eventurl.com" streamEventMethod="POST" username="user" password="pass"><StreamParam name="name1" value="value1" /></StartStream>'
+        else:
+            expected = '<StartStream destination="testurl.com" name="stream1" tracks="inbound" streamEventUrl="eventurl.com" streamEventMethod="POST" username="user" password="pass"><StreamParam name="name1" value="value1" /></StartStream>'
+        assert(expected == self.start_stream.to_bxml())
+    
+    def test_add_verb(self):
+        if os.environ['PYTHON_VERSION'] == '3.7':
+            expected = '<StartStream destination="testurl.com" name="stream1" tracks="inbound" streamEventUrl="eventurl.com" streamEventMethod="POST" username="user" password="pass"><StreamParam name="name1" value="value1" /><StreamParam name="name2" value="value2" /></StartStream>'
+        else:
+            expected = '<StartStream destination="testurl.com" name="stream1" tracks="inbound" streamEventUrl="eventurl.com" streamEventMethod="POST" username="user" password="pass"><StreamParam name="name1" value="value1" /><StreamParam name="name2" value="value2" /></StartStream>'
+        self.start_stream.add_verb(self.stream_param2)
+        assert(expected == self.start_stream.to_bxml())

test/unit/bxml/test_stop_stream.py

@@ -0,0 +1,28 @@
+"""
+test_stop_stream.py
+
+Unit tests for the <StopStream> BXML verb
+
+@copyright Bandwidth Inc.
+"""
+import pytest
+import unittest
+
+from bandwidth.model.bxml.verb import Verb
+from bandwidth.model.bxml.verbs.stop_stream import StopStream
+
+
+class TestTag(unittest.TestCase):
+
+    def setUp(self):
+        self.stop_stream = StopStream(name="conf")
+        self.test_verb = Verb(tag="test")
+
+    def test_to_bxml(self):
+        expected = '<StopStream name="conf" />'
+        assert(expected == self.stop_stream.to_bxml())
+
+    def test_add_verb(self):
+        with pytest.raises(AttributeError):
+            self.stop_stream.add_verb(self.test_verb)
+