Basic Python unit tests.

.travis.yml

@@ -11,3 +11,7 @@ matrix:
 
 install:
   - make
+
+script:
+  - source setup-env.sh
+  - python test/test.py

lib/python/plasma.py

@@ -1,3 +1,4 @@
+import os
 import socket
 import ctypes
 
@@ -33,13 +34,14 @@ def make_plasma_id(string):
 
 class PlasmaClient(object):
   def __init__(self, socket_name):
-    self.client = ctypes.cdll.LoadLibrary("../../build/plasma_client.so")
+    plasma_client_library = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../../build/plasma_client.so")
+    self.client = ctypes.cdll.LoadLibrary(plasma_client_library)
 
     self.client.plasma_store_connect.restype = ctypes.c_int
 
     self.client.plasma_create.argtypes = [ctypes.c_int, PlasmaID, ctypes.c_int64]
     self.client.plasma_create.restype = PlasmaBuffer
-    
+
     self.client.plasma_get.argtypes = [ctypes.c_int, PlasmaID]
     self.client.plasma_get.restype = PlasmaBuffer
 
@@ -63,8 +65,6 @@ class PlasmaClient(object):
   def get(self, object_id):
     buf = self.client.plasma_get(self.sock, make_plasma_id(object_id))
     return self.buffer_from_memory(buf.data, buf.size)
-    
+
   def seal(self, object_id):
     self.client.plasma_seal(self.sock, make_plasma_id(object_id))
-
-    

setup-env.sh

@@ -0,0 +1,5 @@
+echo "Adding Plasma to PYTHONPATH" 1>&2
+
+ROOT_DIR=$(cd "$(dirname "${BASH_SOURCE:-$0}")"; pwd)
+
+export PYTHONPATH="$ROOT_DIR/lib/python/:$PYTHONPATH"

test/test.py

@@ -0,0 +1,61 @@
+import os
+import subprocess
+import sys
+import time
+import unittest
+
+import plasma
+
+class TestPlasmaAPI(unittest.TestCase):
+
+  def setUp(self):
+    # Start Plasma.
+    plasma_store_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../build/plasma_store")
+    self.p = subprocess.Popen([plasma_store_executable, "-s", "/tmp/store"])
+    time.sleep(0.1)
+    # Connect to Plasma.
+    self.plasma_client = plasma.PlasmaClient("/tmp/store")
+
+  def tearDown(self):
+    # Kill the plasma stoe process.
+    self.p.kill()
+
+  def test_create(self):
+    # Create an object string.
+    object_id = "id" + 18 * "x"
+    # Create a new buffer and write to it.
+    length = 1000
+    memory_buffer = self.plasma_client.create(object_id, length)
+    for i in range(length):
+      memory_buffer[i] = chr(i % 256)
+    # Seal the object.
+    self.plasma_client.seal(object_id)
+    # Get the object.
+    memory_buffer = self.plasma_client.get(object_id)
+    for i in range(length):
+      self.assertEqual(memory_buffer[i], chr(i % 256))
+
+  def test_illegal_functionality(self):
+    # Create an object string.
+    object_id = "id" + 18 * "x"
+    # Create a new buffer and write to it.
+    length = 1000
+    memory_buffer = self.plasma_client.create(object_id, length)
+    # Make sure we cannot access memory out of bounds.
+    self.assertRaises(Exception, lambda : memory_buffer[length])
+    # Seal the object.
+    self.plasma_client.seal(object_id)
+    # This test is commented out because it currently fails.
+    # # Make sure the object is ready only now.
+    # def illegal_assignment():
+    #   memory_buffer[0] = chr(0)
+    # self.assertRaises(Exception, illegal_assignment)
+    # Get the object.
+    memory_buffer = self.plasma_client.get(object_id)
+    # Make sure the object is read only.
+    def illegal_assignment():
+      memory_buffer[0] = chr(0)
+    self.assertRaises(Exception, illegal_assignment)
+
+if __name__ == "__main__":
+  unittest.main(verbosity=2)