Bump version: 4.0.0b7 → 4.0.0b8. Pytest compatibility

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 4.0.0b7
+current_version = 4.0.0b9
 files = bsb/__init__.py
 commit = True
 tag = True

.gitignore

@@ -2,6 +2,7 @@ __pycache__/
 *.gdf
 *.pyc
 *.hdf5
+*.pkl
 *.swc
 *.h5
 *.json

bsb/__init__.py

@@ -7,7 +7,7 @@
 install the `bsb` package instead.
 """
 
-__version__ = "4.0.0b7"
+__version__ = "4.0.0b9"
 
 import functools
 

bsb/reporting.py

@@ -45,6 +45,10 @@ def in_notebook():
     return True
 
 
+def in_pytest():
+    return "pytest" in sys.modules
+
+
 def set_report_file(v):
     """
     Set a file to which the scaffold package should report instead of stdout.
@@ -147,7 +151,7 @@ def _decode(payload: str):
 def setup_reporting():
     warnings.formatwarning = warning_on_one_line
     # Don't touch stdout if we're in IPython
-    if in_notebook():
+    if in_notebook() or in_pytest():
         return
     # Otherwise, tinker with stdout so that we autoflush after each write, better for MPI.
     try:

tests/test_configuration.py

@@ -1,3 +1,4 @@
+import inspect
 import json
 import sys
 import unittest
@@ -1149,14 +1150,27 @@ def _eval(statement, **vars):
         self.assertEqual(3, _eval("np.array([v - 2])[0]", v=5))
 
     def test_class(self):
+        """
+        Check that class retrieval fetches the right objects. The test is somewhat
+        complicated to make sure that this test can be run in a directory independent way.
+        """
+
         @config.root
         class Test:
             a = config.attr(type=types.class_())
             b = config.attr(type=types.class_(module_path=["test_configuration"]))
 
-        cfg = Test({"a": "test_configuration.MyTestClass", "b": "MyTestClass"})
-        self.assertEqual(MyTestClass, cfg.a)
-        self.assertEqual(MyTestClass, cfg.b)
+        import pathlib
+        import sys
+
+        sys.path.insert(0, str(pathlib.Path(__file__).parent))
+        try:
+            cfg = Test({"a": "test_configuration.MyTestClass", "b": "MyTestClass"})
+        finally:
+            sys.path.pop(0)
+
+        self.assertEqual(inspect.getsource(MyTestClass), inspect.getsource(cfg.a))
+        self.assertEqual(inspect.getsource(MyTestClass), inspect.getsource(cfg.b))
 
         with self.assertRaises(CastError):
             cfg = Test({"a": "MyTestClass"})
@@ -1273,6 +1287,10 @@ class Classmap2ChildB(Classmap2Parent):
 
 
 class MyTestClass:
+    """
+    Test class used for testing class object retrieval from a module.
+    """
+
     pass
 
 

tests/test_placement.py

@@ -19,12 +19,12 @@
 from bsb.voxels import VoxelData, VoxelSet
 
 
-def test_dud(scaffold, x, y):
+def dud_tester(scaffold, x, y):
     sleep(y)
     return x
 
 
-def test_chunk(scaffold, chunk):
+def chunk_tester(scaffold, chunk):
     return chunk
 
 
@@ -129,7 +129,7 @@ def test_create_pool(self):
     @timeout(3)
     def test_single_job(self):
         pool = JobPool(_net)
-        job = pool.queue(test_dud, (5, 0.1))
+        job = pool.queue(dud_tester, (5, 0.1))
         pool.execute()
 
     @timeout(3)
@@ -141,7 +141,7 @@ def spy(job):
             i += 1
 
         pool = JobPool(_net, listeners=[spy])
-        job = pool.queue(test_dud, (5, 0.1))
+        job = pool.queue(dud_tester, (5, 0.1))
         pool.execute()
         if not MPI.get_rank():
             self.assertEqual(1, i, "Listeners not executed.")
@@ -153,7 +153,7 @@ def test_placement_job(self):
 
     def test_chunked_job(self):
         pool = JobPool(_net)
-        job = pool.queue_chunk(test_chunk, _chunk(0, 0, 0))
+        job = pool.queue_chunk(chunk_tester, _chunk(0, 0, 0))
         pool.execute()
 
     def test_notparallel_ps_job(test):
@@ -203,16 +203,16 @@ class TestParallelScheduler(unittest.TestCase, SchedulerBaseTest):
     @timeout(3)
     def test_double_pool(self):
         pool = JobPool(_net)
-        job = pool.queue(test_dud, (5, 0.1))
+        job = pool.queue(dud_tester, (5, 0.1))
         pool.execute()
         pool = JobPool(_net)
-        job = pool.queue(test_dud, (5, 0.1))
+        job = pool.queue(dud_tester, (5, 0.1))
         pool.execute()
 
     @timeout(3)
     def test_master_loop(self):
         pool = JobPool(_net)
-        job = pool.queue(test_dud, (5, 0.1))
+        job = pool.queue(dud_tester, (5, 0.1))
         executed = False
 
         def spy_loop(p):
@@ -228,16 +228,16 @@ def spy_loop(p):
     @timeout(3)
     def test_fake_futures(self):
         pool = JobPool(_net)
-        job = pool.queue(test_dud, (5, 0.1))
+        job = pool.queue(dud_tester, (5, 0.1))
         self.assertIs(FakeFuture.done, job._future.done.__func__)
         self.assertFalse(job._future.done())
         self.assertFalse(job._future.running())
 
     @timeout(3)
     def test_dependencies(self):
         pool = JobPool(_net)
-        job = pool.queue(test_dud, (5, 0.1))
-        job2 = pool.queue(test_dud, (5, 0.1), deps=[job])
+        job = pool.queue(dud_tester, (5, 0.1))
+        job2 = pool.queue(dud_tester, (5, 0.1), deps=[job])
         result = None
 
         def spy_queue(jobs):