improve coverage

Technologicat · Technologicat · commit 5ebfcfafc5f3 · 2020-08-27T00:34:35.000+03:00
diff --git a/unpythonic/dynassign.py b/unpythonic/dynassign.py
@@ -55,7 +55,10 @@ def __init__(self):
         super().__init__(self)
         self._refresh()
         _getobservers()[id(self)] = self
-    def __del__(self):
+    # TODO: __del__ most certainly runs during test_dynassign (as can be
+    # evidenced by placing a debug print inside it), but coverage fails
+    # to report it as covered.
+    def __del__(self):  # pragma: no cover
         # No idea how, but our REPL server can trigger a KeyError here
         # if the user views `help()`, which causes the client to get stuck.
         # Then pressing `q` in the server console to quit the help, and then
diff --git a/unpythonic/test/test_dynassign.py b/unpythonic/test/test_dynassign.py
@@ -1,18 +1,31 @@
 # -*- coding: utf-8 -*-
 
 from ..syntax import macros, test, test_raises  # noqa: F401
-from .fixtures import session, testset
+from .fixtures import session, testset, returns_normally
 
 import threading
 from queue import Queue
+import gc
 
 from ..dynassign import dyn, make_dynvar
+from ..misc import slurp
 
 def runtests():
+    # various parts of unpythonic use dynvars, so get what's there before we insert anything for testing
+    implicits = [k for k in dyn]
+    def noimplicits(kvs):
+        return tuple(sorted((k, v) for k, v in kvs if k not in implicits))
+    def noimplicits_keys(keys):
+        return tuple(sorted(k for k in keys if k not in implicits))
+
+    # some test data
+    D = {"a": 1, "b": 2}
+    D2 = {"c": 3, "d": 4}
+
     def f():
         test[dyn.a == 2]  # no a in lexical scope
 
-    def runtest():
+    def basictests():
         with testset("basic usage"):
             with dyn.let(a=2, b="foo"):
                 test[dyn.a == 2]
@@ -46,14 +59,9 @@ def threadtest(q):
             t2.join()
             err = comm.get()
             test[err is not None]
-    runtest()
+    basictests()
 
     with testset("syntactic sugar"):
-        # various parts of unpythonic use dynvars, so get what's there before we insert anything for testing
-        implicits = [k for k in dyn]
-        def noimplicits(kvs):
-            return tuple(sorted((k, v) for k, v in kvs if k not in implicits))
-        D = {"a": 1, "b": 2}
         with dyn.let(**D):
             # membership test
             test["a" in dyn]
@@ -66,7 +74,6 @@ def noimplicits(kvs):
         test[noimplicits(dyn.items()) == ()]
 
     with testset("update existing bindings"):
-        D2 = {"c": 3, "d": 4}
         with dyn.let(**D):
             with dyn.let(**D2):
                 test[noimplicits(dyn.items()) == (("a", 1), ("b", 2), ("c", 3), ("d", 4))]
@@ -76,7 +83,12 @@ def noimplicits(kvs):
                 test[noimplicits(dyn.items()) == (("a", 42), ("b", 2), ("c", 23), ("d", 4))]
                 with test_raises(AttributeError, "should not be able to update unbound dynamic variable"):
                     dyn.e = 5
-            test[noimplicits(dyn.items()) == (("a", 42), ("b", 2))]
+
+                # subscript notation also works for updating
+                with test:
+                    dyn["a"] = 9001
+                test[dyn.a == 9001]
+            test[noimplicits(dyn.items()) == (("a", 9001), ("b", 2))]
         test[noimplicits(dyn.items()) == ()]
 
     with testset("update in presence of name shadowing"):
@@ -100,8 +112,43 @@ def noimplicits(kvs):
             test[noimplicits(dyn.items()) == (("a", 10), ("b", 20))]
         test[noimplicits(dyn.items()) == ()]
 
+    with testset("mass update with multithreading"):
+        comm = Queue()
+        def worker():
+            # test[] itself is thread-safe, but the worker threads don't have a
+            # surrounding testset to catch failures, since we don't want to print in them.
+            try:
+                local_successes = 0
+                with dyn.let(**D):
+                    with dyn.let(**D2):
+                        if noimplicits(dyn.items()) == (("a", 1), ("b", 2), ("c", 3), ("d", 4)):
+                            local_successes += 1
+                        dyn.update(a=-1, b=-2, c=-3, d=-4)
+                        if noimplicits(dyn.items()) == (("a", -1), ("b", -2), ("c", -3), ("d", -4)):
+                            local_successes += 1
+                    if noimplicits(dyn.items()) == (("a", -1), ("b", -2)):
+                        local_successes += 1
+                    dyn.update(a=10, b=20)
+                    if noimplicits(dyn.items()) == (("a", 10), ("b", 20)):
+                        local_successes += 1
+                if noimplicits(dyn.items()) == ():
+                    local_successes += 1
+                if local_successes == 5:
+                    comm.put(1)
+            except Exception:  # pragma: no cover, only happens if the test fails.
+                pass
+        n = 100
+        threads = [threading.Thread(target=worker) for _ in range(n)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        successes = sum(slurp(comm))
+        test[successes == n]
+
     with testset("make_dynvar (default values)"):
         make_dynvar(im_always_there=True)
+        test[dyn.im_always_there is True]
         with dyn.let(a=1, b=2):
             test[noimplicits(dyn.items()) == (("a", 1), ("b", 2),
                                               ("im_always_there", True))]
@@ -114,6 +161,8 @@ def noimplicits(kvs):
         with dyn.let(a=1, b=2):
             test[noimplicits(view.items()) == (("a", 1), ("b", 2),
                                                ("im_always_there", True))]
+        del view
+        gc.collect()
 
         # as does dyn.items() (it's an abbreviation for dyn.asdict().items())
         items = dyn.items()
@@ -122,6 +171,22 @@ def noimplicits(kvs):
             test[noimplicits(items) == (("a", 1), ("b", 2),
                                         ("im_always_there", True))]
 
+        # the rest of the Mapping API
+        keys = dyn.keys()  # live!
+        with dyn.let(a=1, b=2):
+            test[noimplicits_keys(keys) == ("a", "b", "im_always_there")]
+
+            test[dyn.get("a") == 1]
+            test[dyn.get("c") is None]  # default
+
+            d = dict(items)
+            test[dyn == d]
+
+        # Not much we can do with the output so let's just check these don't crash.
+        test[returns_normally(dyn.values())]
+        test[returns_normally(len(dyn))]
+
+
 if __name__ == '__main__':  # pragma: no cover
     with session(__file__):
         runtests()
