Commit e95847c9 authored by Kai-Uwe Sattler's avatar Kai-Uwe Sattler
Browse files

Updates on PyTest

parents b63126e2 b605c064
......@@ -73,7 +73,7 @@ add_custom_command(
if (BUILD_GOOGLE_BENCH)
download_project(PROJ benchmark
GIT_REPOSITORY https://github.com/google/benchmark.git
GIT_TAG master
GIT_TAG v1.2.0
UPDATE_DISCONNECTED 1
QUIET
)
......
......@@ -198,7 +198,8 @@ BOOST_PYTHON_MODULE(pyfabric) {
.def("join", &pfabric::PyPipe::join)
.def("queue", &pfabric::PyPipe::queue)
.def("notify", &pfabric::PyPipe::notify)
.def("print", &pfabric::PyPipe::print)
.def("pfprint", &pfabric::PyPipe::print)
.def("notify", &pfabric::PyPipe::notify)
;
bp::enum_<pfabric::AggrFuncType>("aggr")
......
......@@ -125,7 +125,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
*/
......
......@@ -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()
import sys
sys.path.append('../../build/')
print(sys.path)
import pyfabric
import unittest
import os
class TestPipeFabricPython(unittest.TestCase):
#test case for extract operator
def test_extract(self):
t = pyfabric.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 = pyfabric.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 = pyfabric.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()
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment