added python test case(s), minor changes

src/CMakeLists.txt

@@ -231,7 +231,7 @@ find_package(Boost ${BOOST_MIN_VERSION} REQUIRED COMPONENTS
     serialization
     thread
     regex
-    ${BOOST_PYTHONLIB}
+    python
     chrono
     date_time
 )
@@ -273,10 +273,10 @@ if (PYTHONLIBS_FOUND)
 message(STATUS "Python Version ${PYTHON_VERSION_MAJOR} found - building the Python API.")
 #set (PYTHON_INCLUDE_DIRS "/Users/kai/miniconda3/include/python3.5m")
 message("PythonInclude ${PYTHON_INCLUDE_DIRS}")
-message("PythonLibs ${PYTHON_LIBDIR}")
+message("PythonLibs ${PYTHON_LIBRARIES}")
 
 include_directories(${PYTHON_INCLUDE_DIRS})
-link_directories(${PYTHON_LIBDIR})
+link_directories(${PYTHON_LIBRARIES})
 
 set(PYTHON_SOURCES
   python/PyTopology.cpp
@@ -484,7 +484,7 @@ add_library(pfabric SHARED
 )
 
 target_link_libraries(pfabric
-              ${Boost_PYTHON3_LIBRARY}
+              ${Boost_PYTHON_LIBRARY}
               ${PYTHON_LIBRARIES}
               pfabric_core
               ${BOOST_LIBRARIES}

src/python/PyTopology.cpp

@@ -72,6 +72,13 @@ PyPipe PyPipe::print() {
   return PyPipe(pipe.print(std::cout, pyFormatter));
 }
 
+PyPipe PyPipe::notify(bp::object fun) {
+  auto pipe = boost::get<TuplePipe&>(pipeImpl);
+  return PyPipe(pipe.notify([fun](auto tp, bool o) {
+    return fun(get<0>(tp), o);
+  }));
+}
+
 PyTopology::PyTopology() {
   topo = ctx.createTopology();
 }
@@ -94,7 +101,7 @@ BOOST_PYTHON_MODULE(libpfabric) {
       .def("extract", &pfabric::PyPipe::extract)
         .def("where", &pfabric::PyPipe::where)
         .def("map", &pfabric::PyPipe::map)
-        .def("print", &pfabric::PyPipe::print)
-
+        .def("pfprint", &pfabric::PyPipe::print)
+        .def("notify", &pfabric::PyPipe::notify)
     ;
 }

src/python/PyTopology.hpp

@@ -86,7 +86,7 @@ struct PyPipe {
    * Creates a map operator which applies a mapping (projection) function
    * written in Python to each tuples as the next operator on the pipe.
    *
-   * @param[in] fun  a function pointer or lambda function producinh a new tuple
+   * @param[in] fun  a function pointer or lambda function producing a new tuple
    *                from the input tuple
    * @return a new PyPipe object
    */
@@ -101,6 +101,18 @@ struct PyPipe {
    * @return a new PyPipe object
    */
   PyPipe print();
+
+  /**
+   * @brief Creates a notification operator.
+   *
+   * Creates an operator for handling general lambda functions
+   * as the next operator on the pipe.
+   *
+   * @param[in] fun  a function pointer or lambda function applied to the tuples
+   *
+   * @return a new PyPipe object
+   */
+  PyPipe notify(bp::object fun);
 };
 
 /**

src/test/CMakeLists.txt

@@ -71,4 +71,9 @@ if (BUILD_TEST_CASES)
   if(USE_MQTT)
     do_test(MQTTSourceTest)
   endif()
+
+  add_test(
+    NAME Python_Test
+    COMMAND ${PYTHON_EXECUTABLE} ${CMAKE_CURRENT_SOURCE_DIR}/PyTest.py
+  )
 endif()

src/test/PyTest.py

+import sys
+sys.path.append('../../build/')
+print(sys.path)
+
+import libpfabric
+import unittest
+import os
+
+class TestPipeFabricPython(unittest.TestCase):
+
+  #test case for extract operator
+  def test_extract(self):
+    t = libpfabric.Topology()
+
+    #write three tuples to file
+    file = open("data.csv","w")
+    file.write("1,teststring,1.5\n2,teststring,2.5\n3,teststring,3.5\n")
+    file.close()
+
+    strm = []
+    expected = ['1','teststring','1.5','2','teststring','2.5','3','teststring','3.5']
+
+    p = t.newStreamFromFile("data.csv") \
+          .extract(',') \
+          .notify(lambda tup, o: strm.extend((tup[0], tup[1], tup[2]))) \
+          .pfprint() \
+
+    t.start()
+
+    self.assertEqual(strm, expected)
+
+    os.remove("data.csv")
+
+
+  #test case for map operator
+  def test_map(self):
+    t = libpfabric.Topology()
+
+    #write three tuples to file
+    file = open("data.csv","w")
+    file.write("1,teststring,1.5\n2,teststring,2.5\n3,teststring,3.5\n")
+    file.close()
+
+    strm = []
+    expected = [(1,'teststring','1.5'),(2,'teststring','2.5'),(3,'teststring','3.5')]
+
+    p = t.newStreamFromFile("data.csv") \
+          .extract(',') \
+          .map(lambda t, o: (int(t[0]), t[1], t[2])) \
+          .notify(lambda t, o: strm.append(t)) \
+
+    t.start()
+
+    self.assertEqual(strm, expected)
+
+    os.remove("data.csv")
+
+
+  #test case for where operator
+  def test_where(self):
+    t = libpfabric.Topology()
+
+    #write three tuples to file
+    file = open("data.csv","w")
+    file.write("1,teststring,1.5\n2,teststring,2.5\n3,teststring,3.5\n")
+    file.close()
+
+    strm = []
+    expected = [(2,'teststring','2.5'),(3,'teststring','3.5')]
+
+    p = t.newStreamFromFile("data.csv") \
+          .extract(',') \
+          .map(lambda t, o: (int(t[0]), t[1], t[2])) \
+          .where(lambda x, o: x[0] > 1) \
+          .notify(lambda t, o: strm.append(t)) \
+
+    t.start()
+
+    self.assertEqual(strm, expected)
+
+    os.remove("data.csv")
+
+
+if __name__ == '__main__':
+  unittest.main()
\ No newline at end of file