database for unit tests

apstools/plans.py

@@ -73,7 +73,7 @@ def addDeviceDataAsStream(devices, label):
 
     """
     yield from bps.create(name=label)
-    if isinstance(devices, Device):     # just in case...
+    if not isinstance(devices, list):     # just in case...
         devices = [devices]
     for d in devices:
         yield from bps.read(d)

apstools/utils.py

@@ -816,6 +816,28 @@ def json_export(headers, filename, zipfilename=None):
 def json_import(filename, zipfilename=None):
     """
     read the file exported by :meth:`~json_export()`
+
+    RETURNS
+
+    datasets : list of documents
+        list of 
+        `documents <https://blueskyproject.io/bluesky/documents.html/>`_,
+        such as returned by
+        `[list(h.documents()) for h in db]`
+
+        See:
+        https://blueskyproject.io/databroker/generated/databroker.Header.documents.html
+
+    EXAMPLE
+
+    Insert the datasets into the databroker ``db``::
+
+        def insert_docs(db, datasets):
+            for i, h in enumerate(datasets):
+                print(f"{i+1}/{len(datasets)} : {len(h)} documents")
+                for k, doc in h:
+                    db.insert(k, doc)
+
     """
     if zipfilename is None:
         with open(filename, "r") as fp:

tests/test_export_json.py

@@ -20,15 +20,28 @@
 from apstools.utils import json_export, json_import
 
 
+TEST_JSON_FILE = "data.json"
+TEST_ZIP_FILE = os.path.join(_test_path, "bluesky_data.zip")
+
+
 def get_db():
-    from databroker import Broker
-    try:
-        db = Broker.named("mongodb_config")
-    except FileNotFoundError:
-        return
+    from databroker import Broker, temp_config
+    conf = temp_config()
+    conf["metadatastore"]["config"]["timezone"] = "US/Central"
+    db = Broker.from_config(conf)
+    datasets = json_import(TEST_JSON_FILE, TEST_ZIP_FILE)
+    insert_docs(db, datasets)
     return db
 
 
+def insert_docs(db, datasets, verbose=False):
+    for i, h in enumerate(datasets):
+        if verbose:
+            print(f"{i+1}/{len(datasets)} : {len(h)} documents")
+        for k, doc in h:
+            db.insert(k, doc)
+
+
 class Test_JsonExport(unittest.TestCase):
 
     def setUp(self):
@@ -40,9 +53,6 @@ def tearDown(self):
 
     def test_export_import(self):
         db = get_db()
-        if db is None:
-            self.assertTrue(True, "skipping test: no databroker")
-            return
         headers = db(plan_name="count")
         headers = list(headers)[0:1]
 
@@ -66,9 +76,6 @@ def test_export_import(self):
 
     def test_export_import_zip(self):
         db = get_db()
-        if db is None:
-            self.assertTrue(True, "skipping test: no databroker")
-            return
         headers = db(plan_name="count")
         headers = list(headers)[0:1]
 

tests/test_plans.py

@@ -0,0 +1,151 @@
+
+"""
+simple unit tests for this package
+"""
+
+from io import StringIO
+import os
+import sys
+import unittest
+
+_test_path = os.path.dirname(__file__)
+_path = os.path.join(_test_path, '..')
+if _path not in sys.path:
+    sys.path.insert(0, _path)
+
+from apstools import plans as APS_plans
+# from apstools import utils as APS_utils
+from bluesky.simulators import summarize_plan
+import ophyd.sim
+
+class Capture_stdout(list):     # lgtm [py/missing-equals]
+    '''
+    capture all printed output (to stdout) into list
+
+    # http://stackoverflow.com/questions/16571150/how-to-capture-stdout-output-from-a-python-function-call
+    '''
+    def __enter__(self):
+        sys.stdout.flush()
+        self._stdout = sys.stdout
+        sys.stdout = self._stringio = StringIO()
+        return self
+
+    def __exit__(self, *args):
+        self.extend(self._stringio.getvalue().splitlines())
+        del self._stringio    # free up some memory
+        sys.stdout = self._stdout
+
+
+class Test_Plans(unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def test_addDeviceDataAsStream(self):
+        with Capture_stdout() as received:
+            summarize_plan(
+                APS_plans.addDeviceDataAsStream(
+                    ophyd.sim.motor1, 
+                    "test-device"))
+
+        expected = ["  Read ['motor1']"]
+        self.assertEqual(str(received), str(expected))
+
+        with Capture_stdout() as received:
+            summarize_plan(
+                APS_plans.addDeviceDataAsStream(
+                    [ophyd.sim.motor2, ophyd.sim.motor3], 
+                    "test-device-list"))
+
+        expected = [
+            "  Read ['motor2']",        # TODO: <-- Why?
+            "  Read ['motor2', 'motor3']",
+            ]
+        self.assertEqual(str(received), str(expected))
+
+    def test_run_command_file(self):
+        filename = os.path.join(_test_path, "actions.txt")
+        with Capture_stdout() as received:
+            summarize_plan(
+                APS_plans.run_command_file(filename))
+
+        # print(f"|{received}|")
+        expected = [
+            'Command file: /home/mintadmin/Documents/eclipse/apstools/tests/actions.txt',
+            '====== ============ ========================',
+            'line # action       parameters              ',
+            '====== ============ ========================',
+            '5      sample_slits 0, 0, 0.4, 1.2          ',
+            '7      preusaxstune                         ',
+            '10     FlyScan      0, 0, 0, blank          ',
+            '11     FlyScan      5, 2, 0, empty container',
+            '12     SAXS         0, 0, 0, blank          ',
+            '====== ============ ========================',
+            '',
+            'file line 5: sample_slits 0 0 0.4 1.2',
+            'no handling for line 5: sample_slits 0 0 0.4 1.2',
+            'file line 7: preusaxstune',
+            'no handling for line 7: preusaxstune',
+            'file line 10: FlyScan 0   0   0   blank',
+            'no handling for line 10: FlyScan 0   0   0   blank',
+            'file line 11: FlyScan 5   2   0   "empty container"',
+            'no handling for line 11: FlyScan 5   2   0   "empty container"',
+            'file line 12: SAXS 0 0 0 blank',
+            'no handling for line 12: SAXS 0 0 0 blank',
+            ]
+        self.assertEqual(str(received), str(expected))
+
+        filename = os.path.join(_test_path, "actions.xlsx")
+        with Capture_stdout() as received:
+            summarize_plan(
+                APS_plans.run_command_file(filename))
+
+        # print(f"|{received}|")
+        expected = [
+            'Command file: /home/mintadmin/Documents/eclipse/apstools/tests/actions.xlsx', 
+            '====== ============ =============================', 
+            'line # action       parameters                   ', 
+            '====== ============ =============================', 
+            '1      mono_shutter open                         ', 
+            '2      USAXSscan    45.07, 98.3, 0.0, Water Blank', 
+            '3      saxsExp      45.07, 98.3, 0.0, Water Blank', 
+            '4      waxwsExp     45.07, 98.3, 0.0, Water Blank', 
+            '5      USAXSscan    12, 12.0, 1.2, plastic       ', 
+            '6      USAXSscan    12, 37.0, 0.1, Al foil       ', 
+            '7      mono_shutter close                        ', 
+            '====== ============ =============================', 
+            '', 
+            "file line 1: ['mono_shutter', 'open', None, None, None]", 
+            "no handling for line 1: ['mono_shutter', 'open', None, None, None]", 
+            "file line 2: ['USAXSscan', 45.07, 98.3, 0.0, 'Water Blank']", 
+            "no handling for line 2: ['USAXSscan', 45.07, 98.3, 0.0, 'Water Blank']", 
+            "file line 3: ['saxsExp', 45.07, 98.3, 0.0, 'Water Blank']", 
+            "no handling for line 3: ['saxsExp', 45.07, 98.3, 0.0, 'Water Blank']", 
+            "file line 4: ['waxwsExp', 45.07, 98.3, 0.0, 'Water Blank']", 
+            "no handling for line 4: ['waxwsExp', 45.07, 98.3, 0.0, 'Water Blank']", 
+            "file line 5: ['USAXSscan', 12, 12.0, 1.2, 'plastic']", 
+            "no handling for line 5: ['USAXSscan', 12, 12.0, 1.2, 'plastic']", 
+            "file line 6: ['USAXSscan', 12, 37.0, 0.1, 'Al foil']", 
+            "no handling for line 6: ['USAXSscan', 12, 37.0, 0.1, 'Al foil']", 
+            "file line 7: ['mono_shutter', 'close', None, None, None]", 
+            "no handling for line 7: ['mono_shutter', 'close', None, None, None]"
+            ]
+        self.assertEqual(str(received), str(expected))
+
+
+def suite(*args, **kw):
+    test_list = [
+        Test_Plans,
+        ]
+    test_suite = unittest.TestSuite()
+    for test_case in test_list:
+        test_suite.addTest(unittest.makeSuite(test_case))
+    return test_suite
+
+
+if __name__ == "__main__":
+    runner=unittest.TextTestRunner()
+    runner.run(suite())