WIP:added more test for StreamingContext

python/pyspark/streaming/context.py

@@ -63,6 +63,9 @@ def __init__(self, master=None, appName=None, sparkHome=None, pyFiles=None,
 
         """
 
+        if not isinstance(duration, Duration):
+            raise TypeError("Input should be pyspark.streaming.duration.Duration object")
+
         if sparkContext is None:
             # Create the Python Sparkcontext
             self._sc = SparkContext(master=master, appName=appName, sparkHome=sparkHome,

python/pyspark/streaming/tests.py

@@ -473,6 +473,7 @@ def setUp(self):
     def tearDown(self):
         # Do not call pyspark.streaming.context.StreamingContext.stop directly because
         # we do not wait to shutdown py4j client.
+        # We need change this simply calll streamingConxt.Stop
         self.ssc._jssc.stop()
         self.ssc._sc.stop()
         # Why does it long time to terminate StremaingContext and SparkContext?
@@ -484,7 +485,6 @@ def tearDownClass(cls):
         # Make sure tp shutdown the callback server
         SparkContext._gateway._shutdown_callback_server()
 
-
     def test_from_no_conf_constructor(self):
         ssc = StreamingContext(master=self.master, appName=self.appName, duration=batachDuration)
         # Alternative call master: ssc.sparkContext.master
@@ -513,12 +513,21 @@ def _addInputStream(self, s):
         # make sure numSlice is 2 due to deserializer proglem in pyspark
         s._testInputStream(test_inputs, 2)
 
+    def test_from_no_conf_plus_spark_home_plus_env(self):
+        pass
+
+    def test_from_conf_with_settings(self):
+        pass
+
+    def test_stop_only_streaming_context(self):
+        pass
 
+    def test_await_termination(self):
+        pass
 
 
 
 
 
 if __name__ == "__main__":
     unittest.main()
-    SparkContext._gateway._shutdown_callback_server()