Isolate running tests and running files using processes

Author: hermansje

pythonwhat/local.py

@@ -2,12 +2,17 @@
 import os
 import random
 from pathlib import Path
+from contextlib import redirect_stdout
+
+from multiprocessing import Process, Queue
+from pythonbackend import CaptureErrors
+from pythonbackend.shell_utils import create
+from pythonbackend.tasks import TaskKillProcess, TaskCaptureFullOutput
 
 from pythonwhat.reporter import Reporter
-from contextlib import redirect_stdout
 
 
-class StubShell(object):
+class StubShell:
     def __init__(self, init_code=None):
         self.user_ns = {}
         if init_code:
@@ -17,7 +22,7 @@ def run_code(self, code):
         exec(code, self.user_ns)
 
 
-class StubProcess(object):
+class StubProcess:
     def __init__(self, init_code=None, pid=None):
         self.shell = StubShell(init_code)
         self._identity = (pid,) if pid else (random.randint(0, 1e12),)
@@ -26,6 +31,82 @@ def executeTask(self, task):
         return task(self.shell)
 
 
+class TaskCaptureOutput:
+    def __init__(self, code):
+        self.code = code
+
+    def __call__(self, shell):
+        raw_output, error = run_code(shell.run_code, self.code)
+
+        return raw_output, error
+
+
+# todo: merge with pythonbackend
+class WorkerProcess(Process):
+    instances = []
+
+    def __init__(self, pid=None):
+        Process.__init__(self)
+        self.task_queue = Queue()
+        self.result_queue = Queue()
+        self.daemon = (
+            True
+        )  # when parent process is killed, sub/childprocess get also killed
+        self.instances.append(self)
+        # used to detect single process exercise
+        self._identity = (pid,) if pid else (random.randint(0, 1e12),)
+
+    def get_shell(self):
+        return create({})
+
+    def run(self):
+        shell = self.get_shell()
+        while True:
+            output = []
+            with CaptureErrors(output):
+                next_task = self.task_queue.get()
+                answer = next_task(shell)
+            if len(output) > 0:  # means backend error happened
+                answer = output
+            output = []
+            with CaptureErrors(output):
+                self.result_queue.put_nowait(answer)
+            if len(output) > 0:  # means backend error happened
+                self.result_queue.put_nowait(output)
+            if isinstance(next_task, TaskKillProcess):
+                break  # break while loop -> we do not wait upon new task
+        return
+
+    def executeTask(self, task):
+        self.task_queue.put_nowait(task)
+        return self.result_queue.get()  # wait and fetches next item in queue
+
+    def kill(self):
+        try:
+            if self.is_alive():
+                self.executeTask(TaskKillProcess())
+                self.terminate()
+                self.join(timeout=3.0)
+                if self.is_alive():
+                    raise Exception
+            if self in self.instances:
+                self.instances.remove(self)
+        finally:
+            pass
+            # python 3.7:
+            # self.close()
+
+    @classmethod
+    def kill_all(cls):
+        for instance in list(cls.instances):
+            instance.kill()
+
+
+class SimpleProcess(WorkerProcess):
+    def get_shell(self):
+        return StubShell()
+
+
 class ChDir(object):
     """
     Step into a directory temporarily.
@@ -42,31 +123,57 @@ def __exit__(self, *args):
         os.chdir(self.old_dir)
 
 
-def run_code(process, code):
-    output = io.StringIO()
-    try:
-        with redirect_stdout(output):
-            process.shell.run_code(code)
-        raw_output = output.getvalue()
-        error = None
-    except BaseException as e:
-        raw_output = ""
-        error = str(e)
+def run_code(executor, code):
+    with io.StringIO() as output:
+        try:
+            with redirect_stdout(output):
+                executor(code)
+            raw_output = output.getvalue()
+            error = None
+        except BaseException as e:
+            raw_output = ""
+            error = str(e)
     return raw_output, error
 
 
-def run_single_process(pec, code, pid=None):
-    process = StubProcess(init_code=pec, pid=pid)
-    raw_stu_output, error = run_code(process, code)
+def run_single_process(pec, code, pid=None, mode="simple"):
+    if mode == "stub":
+        # no isolation
+        process = StubProcess(init_code=pec, pid=pid)
+        raw_stu_output, error = run_code(process.shell.run_code, code)
+
+    elif mode == "simple":
+        # no advanced functionality
+        process = SimpleProcess(pid)
+        process.start()
+        _ = process.executeTask(TaskCaptureOutput(pec))
+        raw_stu_output, error = process.executeTask(TaskCaptureOutput(code))
+
+    elif mode == "full":
+        # slow
+        process = WorkerProcess(pid)
+        process.start()
+        _ = process.executeTask(
+            TaskCaptureFullOutput((pec,), "<PEC>", None, silent=True)
+        )
+        output, raw_output = process.executeTask(
+            TaskCaptureFullOutput((code,), "script.py", None, silent=True)
+        )
+        raw_stu_output = raw_output["output_stream"]
+        error = raw_output["error"]
+
+    else:
+        raise ValueError("Invalid mode")
+
     return process, raw_stu_output, error
 
 
-def run_exercise(pec, sol_code, stu_code, pid=None, sol_wd=None, stu_wd=None):
+def run_exercise(pec, sol_code, stu_code, sol_wd=None, stu_wd=None, **kwargs):
     with ChDir(sol_wd or os.getcwd()):
-        sol_process, _, _ = run_single_process(pec, sol_code, pid=pid)
+        sol_process, _, _ = run_single_process(pec, sol_code, **kwargs)
 
     with ChDir(stu_wd or os.getcwd()):
-        stu_process, raw_stu_output, error = run_single_process(pec, stu_code, pid=pid)
+        stu_process, raw_stu_output, error = run_single_process(pec, stu_code, **kwargs)
 
     return sol_process, stu_process, raw_stu_output, error
 
@@ -135,7 +242,6 @@ def run(state, relative_working_dir="", solution_dir="solution"):
         pec="",
         sol_code=state.solution_code or "",
         stu_code=state.student_code,
-        pid=None,
         sol_wd=sol_wd,
         stu_wd=stu_wd,
     )

pythonwhat/test_exercise.py

@@ -104,9 +104,9 @@ def prep_context():
     return tree, cntxt
 
 
-def setup_state(stu_code="", sol_code="", pec="", pid=None):
-    sol_process, stu_process, raw_stu_output, _ = run_exercise(
-        pec, sol_code, stu_code, pid=pid
+def setup_state(stu_code="", sol_code="", pec="", **kwargs):
+    sol_process, stu_process, raw_stu_output, error = run_exercise(
+        pec, sol_code, stu_code, **kwargs
     )
 
     state = State(
@@ -116,7 +116,7 @@ def setup_state(stu_code="", sol_code="", pec="", pid=None):
         student_process=stu_process,
         solution_process=sol_process,
         raw_student_output=raw_stu_output,
-        reporter=Reporter(),
+        reporter=Reporter(errors=[error] if error else []),
     )
 
     State.root_state = state

tests/conftest.py

@@ -0,0 +1,8 @@
+import pytest
+from pythonwhat.local import WorkerProcess
+
+
+@pytest.fixture(scope="function", autouse=True)
+def kill_processes():
+    yield
+    WorkerProcess.kill_all()

tests/helper.py

@@ -4,7 +4,7 @@
 from collections import defaultdict
 from functools import wraps
 
-from pythonwhat.local import StubProcess, run_exercise, ChDir
+from pythonwhat.local import StubProcess, run_exercise, ChDir, WorkerProcess
 from contextlib import contextmanager
 from protowhat.Test import TestFail as TF
 from pythonwhat.test_exercise import test_exercise
@@ -59,8 +59,8 @@ def run(data, run_code=True):
                 )
             else:
                 raw_stu_output = ""
-                stu_process = StubProcess()
-                sol_process = StubProcess()
+                stu_process = StubProcess()  # WorkerProcess()
+                sol_process = StubProcess()  # WorkerProcess()
                 error = None
 
             res = test_exercise(

tests/run.py

@@ -0,0 +1,7 @@
+import os
+import sys
+import pytest
+
+os.environ["OBJC_DISABLE_INITIALIZE_FORK_SAFETY"] = "YES"
+sys.setrecursionlimit(5000)
+pytest.main()

tests/test_check_function.py

@@ -1,5 +1,6 @@
 import pytest
 import tests.helper as helper
+from inspect import getsource
 from pythonwhat.test_exercise import setup_state
 from protowhat.Test import TestFail as TF
 from protowhat.Feedback import InstructorError
@@ -84,18 +85,19 @@ def test_bind_args():
     from inspect import signature
     from pythonwhat.checks.check_function import bind_args
 
-    pec = "def my_fun(a, b, *args, **kwargs): pass"
+    def my_fun(a, b, *args, **kwargs): pass
+    pec = getsource(my_fun).strip()
     s = setup_state(pec=pec, stu_code="my_fun(1, 2, 3, 4, c = 5)")
     args = s._state.ast_dispatcher.find("function_calls", s._state.student_ast)[
         "my_fun"
     ][0]["args"]
-    sig = signature(s._state.student_process.shell.user_ns["my_fun"])
-    binded_args = bind_args(sig, args)
-    assert binded_args["a"]["node"].n == 1
-    assert binded_args["b"]["node"].n == 2
-    assert binded_args["args"][0]["node"].n == 3
-    assert binded_args["args"][1]["node"].n == 4
-    assert binded_args["kwargs"]["c"]["node"].n == 5
+    sig = signature(my_fun)
+    bound_args = bind_args(sig, args)
+    assert bound_args["a"]["node"].n == 1
+    assert bound_args["b"]["node"].n == 2
+    assert bound_args["args"][0]["node"].n == 3
+    assert bound_args["args"][1]["node"].n == 4
+    assert bound_args["kwargs"]["c"]["node"].n == 5
 
 
 @pytest.mark.parametrize("argspec", [["args", 0], ["args", 1], ["kwargs", "c"]])

tests/test_local.py

@@ -0,0 +1,91 @@
+import pytest
+
+from protowhat.Test import TestFail as TF
+from pythonwhat.test_exercise import setup_state
+from tests.helper import verify_sct
+
+
+modify_sys = (
+    """
+import sys
+sys.modules["foo"] = "bar"
+""",
+    """
+import sys
+bar = sys.modules.get("foo")
+    """,
+)
+
+asset = "https://s3.amazonaws.com/assets.datacamp.com/production/course_998/datasets/L-L1_LOSC_4_V1-1126259446-32.hdf5"
+
+
+@pytest.mark.parametrize("sol_code, stu_code", [modify_sys])
+def test_running_code_isolation_stub(sol_code, stu_code):
+    chain = setup_state(stu_code, sol_code, pec="", mode="stub")
+
+    chain.has_equal_value(name="bar", override="bar")
+
+    # clean up
+    import sys
+
+    del sys.modules["foo"]
+
+
+@pytest.mark.parametrize("sol_code, stu_code", [modify_sys])
+def test_running_code_isolation_run(sol_code, stu_code):
+    # test that setup_state is isolated
+    chain = setup_state(sol_code, stu_code, pec="")
+    chain._state.solution_code = sol_code
+    chain._state.student_code = stu_code
+
+    with verify_sct(False):
+        # test that run is isolated
+        chain.run().has_equal_value(name="bar", override="bar")
+
+
+@pytest.mark.parametrize(
+    "sol_code, stu_code",
+    [
+        (
+            """from urllib.request import urlretrieve;
+urlretrieve('{}', 'LIGO_data.hdf5')""".format(
+                asset
+            ),
+        )
+        * 2,
+    ],
+)
+def test_urlretrieve_in_process(sol_code, stu_code):
+    # on mac an env var needs to be set:
+    # OBJC_DISABLE_INITIALIZE_FORK_SAFETY=YES
+    chain = setup_state("", "", pec="")
+
+    chain._state.solution_code = sol_code
+    chain._state.student_code = stu_code
+
+    with verify_sct(True):
+        chain.run()
+
+
+@pytest.mark.parametrize(
+    "sol_code, stu_code",
+    [
+        (
+            """import urllib.request; import shutil
+with urllib.request.urlopen({}) as response, open('LIGO_data.hdf5', 'wb') as out_file:
+    shutil.copyfileobj(response, out_file)
+        """.format(
+                asset
+            ),
+        )
+        * 2,
+    ],
+)
+def test_urlopen_in_process(sol_code, stu_code):
+    chain = setup_state("", "", pec="")
+
+    chain._state.solution_code = sol_code
+    chain._state.student_code = stu_code
+
+    with verify_sct(True):
+        chain.run()