Unittest - Helper code to easily parse tx.data

test/basetest/taskd.py

@@ -8,7 +8,7 @@ import atexit
 from time import sleep
 from subprocess import Popen
 from .utils import (find_unused_port, release_port, port_used, run_cmd_wait,
-                    which)
+                    which, parse_datafile)
 from .exceptions import CommandError
 
 try:
@@ -278,4 +278,16 @@ class Taskd(object):
         else:
             return True
 
+    def client_data(self, client):
+        """Return a python list with the content of tx.data matching the given
+        task client. tx.data will be parsed to string and JSON.
+        """
+        file = os.path.join(self.datadir,
+                            client.credentials["org"],
+                            "users",
+                            client.credentials["userkey"],
+                            "tx.data")
+
+        return parse_datafile(file)
+
 # vim: ai sts=4 et sw=4

test/basetest/utils.py

@@ -9,6 +9,10 @@ from subprocess import Popen, PIPE, STDOUT
 from threading import Thread
 from Queue import Queue, Empty
 from time import sleep
+try:
+    import simplejson as json
+except ImportError:
+    import json
 from .exceptions import CommandError
 
 USED_PORTS = set()
@@ -247,4 +251,23 @@ except ImportError:
                         return name
         return None
 
+def parse_datafile(file):
+    """Parse .data files on the client and server treating files as JSON
+    """
+    data = []
+    with open(file) as fh:
+        for line in fh:
+            line = line.rstrip("\n")
+
+            # Turn [] strings into {} to be treated properly as JSON hashes
+            if line.startswith('[') and line.endswith(']'):
+                line = '{' + line[1:-1] + '}'
+
+            if line.startswith("{"):
+                data.append(json.loads(line))
+            else:
+                data.append(line)
+    return data
+
+
 # vim: ai sts=4 et sw=4