FIX/ENH(TCs): Dill/Pickle various struvts

Author: ankostis

graphtik/execution.py

@@ -286,14 +286,14 @@ class _OpTask:
     This intermediate class is needed to solve pickling issue with process executor.
     """
 
-    __slots__ = ("op", "sol", "result", "solid")
+    __slots__ = ("op", "sol", "solid", "result")
     logname = __name__
 
-    def __init__(self, op, sol, solid):
+    def __init__(self, op, sol, solid, result=UNSET):
         self.op = op
         self.sol = sol
-        self.result = UNSET
         self.solid = solid
+        self.result = result
 
     def marshalled(self):
         import dill

test/conftest.py

@@ -1,4 +1,6 @@
 import pytest
+from .helpers import dilled, pickled
+
 
 # Enable pytest-sphinx fixtures
 # See https://www.sphinx-doc.org/en/master/devguide.html#unit-testing
@@ -15,3 +17,8 @@ def debug_mode():
 
     with debug_enabled(True):
         yield
+
+
+@pytest.fixture(params=[dilled, pickled])
+def ser_method(request):
+    return request.param

test/helpers.py

@@ -1,8 +1,10 @@
+import pickle
 import re
 from itertools import chain, cycle
 from pathlib import Path
 from typing import Union
 
+import dill
 import pytest
 
 ###################################################
@@ -88,3 +90,11 @@ def get_text(node):
             f"{Path(fname).absolute()}:\n  {check!r} {msg} node at path {path!r}: "
             f"{[node.text for node in nodes]}"
         )
+
+
+def dilled(i):
+    return dill.loads(dill.dumps(i))
+
+
+def pickled(i):
+    return pickle.loads(pickle.dumps(i))

test/test_base.py

@@ -14,6 +14,10 @@
 from graphtik.pipeline import Pipeline
 
 
+def test_serialize_Token(ser_method):
+    assert ser_method(base.UNSET) == base.UNSET
+
+
 @pytest.mark.parametrize("locs", [None, (), [], [0], "bad"])
 def test_jetsam_bad_locals(locs, caplog):
     caplog.set_level(logging.INFO)

test/test_graphtik.py

@@ -8,6 +8,7 @@
 import sys
 import time
 import types
+from collections import namedtuple
 from functools import partial
 from itertools import cycle
 from multiprocessing import Pool, cpu_count
@@ -22,7 +23,7 @@
 import pytest
 
 from graphtik import network
-from graphtik.base import AbortedException, IncompleteExecutionError
+from graphtik.base import AbortedException, IncompleteExecutionError, Operation
 from graphtik.config import (
     abort_run,
     debug_enabled,
@@ -34,8 +35,7 @@
     operations_reschedullled,
     tasks_marshalled,
 )
-from graphtik.execution import Solution, task_context
-from graphtik.base import Operation
+from graphtik.execution import Solution, _OpTask, task_context
 from graphtik.modifiers import dep_renamed, optional, sfx, sfxed, vararg
 from graphtik.op import NO_RESULT, NO_RESULT_BUT_SFX, operation
 from graphtik.pipeline import NULL_OP, Pipeline, compose
@@ -49,10 +49,13 @@
 _parallel = pytest.mark.parallel
 _marshal = pytest.mark.marshal
 
+_ExeParams = namedtuple("_ExeParams", "parallel, proc, marshal")
+_exe_params = _ExeParams(None, None, None)
+
 
 @pytest.fixture(
     params=[
-        # PARALLEL?, Proc/Thread?, Marshalled?
+        # PARALLEL?, Thread/Proc?, Marshalled?
         (None, None, None),
         pytest.param((1, 0, 0), marks=(_parallel, _thread)),
         pytest.param((1, 0, 1), marks=(_parallel, _thread, _marshal)),
@@ -70,7 +73,10 @@
 )
 def exemethod(request):
     """Returns (exemethod, marshal) combinations"""
+    global _exe_params
     parallel, proc_pool, marshal = request.param
+    _exe_params = _ExeParams(*request.param)
+
     nsharks = None  # number of pool swimmers....
 
     with tasks_marshalled(marshal):
@@ -118,6 +124,21 @@ def abspow(a, p):
     return c
 
 
+def test_serialize_pipeline(samplenet, ser_method):
+    def eq(pipe1, pipe2):
+        return pipe1.name == pipe2.name and pipe1.ops == pipe2.ops
+
+    assert eq(ser_method(samplenet), samplenet)
+
+
+def test_serialize_OpTask(ser_method):
+    def eq(o1, o2):
+        return all(getattr(o1, a) == getattr(o2, a) for a in _OpTask.__slots__)
+
+    ot = _OpTask(1, 2, 3, 4)
+    assert eq(ser_method(ot), ot)
+
+
 def test_solution_finalized():
     sol = Solution(MagicMock(), {})
 
@@ -285,7 +306,7 @@ def powers_in_range(a, exponent):
     assert sol == exp
 
 
-def test_task_context(exemethod):
+def test_task_context(exemethod, request):
     def check_task_context():
         assert task_context.get().op == next(iop)
 
@@ -296,10 +317,17 @@ def check_task_context():
         parallel=exemethod,
     )
     iop = iter(pipe.ops)
-    if exemethod and is_marshal_tasks():
-        with pytest.raises(AssertionError, match="^assert FunctionalOperation"):
+
+    print(_exe_params)
+    err = None
+    if _exe_params.parallel and _exe_params.marshal:
+        err = AssertionError("^assert FunctionalOperation")
+    if _exe_params.proc and _exe_params.marshal:
+        err = Exception("^Error sending result")
+    if err:
+        with pytest.raises(type(err), match=str(err)):
             pipe.compute()
-        raise pytest.xfail("Cannot marshal `task_context` :-(.")
+        raise pytest.xfail("Cannot marshal parallel processes with `task_context` :-(.")
     else:
         pipe.compute()
         with pytest.raises(StopIteration):

test/test_modifiers.py

@@ -1,4 +1,5 @@
-import dill
+# Copyright 2020, Kostis Anagnostopoulos.
+# Licensed under the terms of the Apache License, Version 2.0. See the LICENSE file associated with the project for terms.
 import pytest
 
 from graphtik.modifiers import (
@@ -16,13 +17,9 @@
 )
 
 
-def dilled(i):
-    return dill.loads(dill.dumps(i))
-
-
-def test_dill_modifier():
+def test_serialize_modifier(ser_method):
     s = sfxed("foo", "gg")
-    s == dilled(s)
+    assert repr(ser_method(s)) == repr(s)
 
 
 ## construct in lambdas, not to crash pytest while modifying core _Modifier
@@ -48,7 +45,6 @@ def test_modifs_str(mod, exp):
     got = mod()
     print(got)
     assert str(got) == exp
-    assert dilled(str(got)) == exp
 
 
 ## construct in lambdas, not to crash pytest while modifying core _Modifier
@@ -75,11 +71,11 @@ def test_modifs_str(mod, exp):
         (lambda: sfxed_varargs("f", "a", "b"), "sfxed('f'(+), 'a', 'b')"),
     ],
 )
-def test_modifs_repr(mod, exp):
+def test_modifs_repr(mod, exp, ser_method):
     got = mod()
     print(repr(got))
     assert repr(got) == exp
-    assert dilled(repr(got)) == exp
+    assert repr(ser_method(got)) == exp
 
 
 @pytest.mark.parametrize(
@@ -105,11 +101,11 @@ def test_modifs_repr(mod, exp):
         (sfxed_varargs("f", "a", "b"), "sfxed_varargs('f', 'a', 'b')"),
     ],
 )
-def test_modifs_cmd(mod, exp):
+def test_modifs_cmd(mod, exp, ser_method):
     got = mod.cmd
     print(got)
     assert str(got) == exp
-    assert dilled(str(got)) == exp
+    assert str(ser_method(got)) == exp
 
 
 def test_recreation():
@@ -189,11 +185,11 @@ def test_modifs_rename_fn(mod, exp):
         ),
     ],
 )
-def test_modifs_rename_str(mod, exp):
+def test_modifs_rename_str(mod, exp, ser_method):
     got = dep_renamed(mod(), "D")
     print(repr(got))
     assert repr(got) == exp
-    assert dilled(repr(got)) == exp
+    assert repr(ser_method(got)) == exp
 
 
 @pytest.mark.parametrize(

test/test_op.py

@@ -8,7 +8,6 @@
 from textwrap import dedent
 from types import SimpleNamespace
 
-import dill
 import pytest
 
 from graphtik import (
@@ -553,8 +552,14 @@ def _opsattrs(ops, attr, value):
         lambda: operation(lambda: None),
     ],
 )
-def test_dill_ops(op_fact):
-    dill.loads(dill.dumps(op_fact()))
+def test_serialize_op(op_fact, ser_method):
+    op = op_fact()
+    if "pickle" in str(ser_method) and "lambda" in str(getattr(op, "fn", ())):
+        with pytest.raises(AttributeError, match="^Can't pickle local object"):
+            ser_method(op) == op
+        raise pytest.xfail(reason="Pickling fails for locals i.e. lambas")
+
+    assert ser_method(op) == op
 
 
 def test_op_rename():