Merge branch 'aevri/no_mark_run_in_subprocess' into 'master'

tests: remove mark.in_subprocess, create_cas_usage_monitor

See merge request BuildStream/buildstream!1661

9 files changed, 11 insertions, 151 deletions

diff --git a/setup.cfg b/setup.cfg
index 53a83541c..eb61e352c 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -21,7 +21,6 @@ markers =
     datafiles: data files for tests
     integration: run test only if --integration option is specified
     remoteexecution: run test only if --remote-execution option is specified
-    in_subprocess: run test in a Python process forked from the main one
 xfail_strict=True
 
 [pycodestyle]
diff --git a/src/buildstream/_cas/cascache.py b/src/buildstream/_cas/cascache.py
index 2a2313097..eb28e2df2 100644
--- a/src/buildstream/_cas/cascache.py
+++ b/src/buildstream/_cas/cascache.py
@@ -83,6 +83,7 @@ class CASCache():
         self._casd_cas = None
         self._local_cas = None
         self._cache_usage_monitor = None
+        self._cache_usage_monitor_forbidden = False
 
         if casd:
             # Place socket in global/user temporary directory to avoid hitting
@@ -113,7 +114,7 @@ class CASCache():
                     self._casd_process = subprocess.Popen(
                         casd_args, cwd=path, stdout=logfile_fp, stderr=subprocess.STDOUT)
 
-            self._cache_usage_monitor = _CASCacheUsageMonitor.create_cas_usage_monitor(self)
+            self._cache_usage_monitor = _CASCacheUsageMonitor(self)
         else:
             self._casd_process = None
 
@@ -125,6 +126,14 @@ class CASCache():
         assert '_casd_process' in state
         state['_casd_process'] = bool(self._casd_process)
 
+        # The usage monitor is not pickle-able, but we also don't need it in
+        # child processes currently. Make sure that if this changes, we get a
+        # bug report, by setting _cache_usage_monitor_forbidden.
+        assert '_cache_usage_monitor' in state
+        assert '_cache_usage_monitor_forbidden' in state
+        state['_cache_usage_monitor'] = None
+        state['_cache_usage_monitor_forbidden'] = True
+
         return state
 
     def _init_casd(self):
@@ -1047,6 +1056,7 @@ class CASCache():
     #     (CASCacheUsage): The current status
     #
     def get_cache_usage(self):
+        assert not self._cache_usage_monitor_forbidden
         return self._cache_usage_monitor.get_cache_usage()
 
     # get_casd_process()
@@ -1099,15 +1109,6 @@ class _CASCacheUsage():
 # buildbox-casd.
 #
 class _CASCacheUsageMonitor:
-
-    #  FIXME: SemaphoreTracker crashes when triggered via spawn and used in
-    #  tests using fork.
-    @classmethod
-    def create_cas_usage_monitor(cls, cas):
-        if multiprocessing.get_start_method() == 'spawn':
-            return None
-        return cls(cas)
-
     def __init__(self, cas):
         self.cas = cas
 
diff --git a/src/buildstream/testing/_forked.py b/src/buildstream/testing/_forked.py
deleted file mode 100644
index 164906b0c..000000000
--- a/src/buildstream/testing/_forked.py
+++ /dev/null
@@ -1,100 +0,0 @@
-# This code was based on pytest-forked, commit 6098c1, found here:
-# <https://github.com/pytest-dev/pytest-forked>
-# Its copyright notice is included below.
-#
-# Permission is hereby granted, free of charge, to any person obtaining a copy
-# of this software and associated documentation files (the "Software"), to deal
-# in the Software without restriction, including without limitation the rights
-# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-# copies of the Software, and to permit persons to whom the Software is
-# furnished to do so, subject to the following conditions:
-#
-# The above copyright notice and this permission notice shall be included in all
-# copies or substantial portions of the Software.
-#
-# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-# SOFTWARE.
-
-import os
-import marshal
-
-import py
-import pytest
-# XXX Using pytest private internals here
-from _pytest import runner
-
-from buildstream import utils
-
-EXITSTATUS_TESTEXIT = 4
-
-
-# copied from xdist remote
-def serialize_report(rep):
-    d = rep.__dict__.copy()
-    if hasattr(rep.longrepr, 'toterminal'):
-        d['longrepr'] = str(rep.longrepr)
-    else:
-        d['longrepr'] = rep.longrepr
-    for name in d:
-        if isinstance(d[name], py.path.local):  # pylint: disable=no-member
-            d[name] = str(d[name])
-        elif name == "result":
-            d[name] = None  # for now
-    return d
-
-
-def forked_run_report(item):
-    def runforked():
-        # This process is now the main BuildStream process
-        # for the duration of this test.
-        utils._MAIN_PID = os.getpid()
-
-        try:
-            reports = runner.runtestprotocol(item, log=False)
-        except KeyboardInterrupt:
-            os._exit(EXITSTATUS_TESTEXIT)
-        return marshal.dumps([serialize_report(x) for x in reports])
-
-    ff = py.process.ForkedFunc(runforked)  # pylint: disable=no-member
-    result = ff.waitfinish()
-    if result.retval is not None:
-        report_dumps = marshal.loads(result.retval)
-        return [runner.TestReport(**x) for x in report_dumps]
-    else:
-        if result.exitstatus == EXITSTATUS_TESTEXIT:
-            pytest.exit("forked test item %s raised Exit" % (item,))
-        return [report_process_crash(item, result)]
-
-
-def report_process_crash(item, result):
-    try:
-        from _pytest.compat import getfslineno
-    except ImportError:
-        # pytest<4.2
-        path, lineno = item._getfslineno()
-    else:
-        path, lineno = getfslineno(item)
-    info = ("%s:%s: running the test CRASHED with signal %d" %
-            (path, lineno, result.signal))
-
-    # We need to create a CallInfo instance that is pre-initialised to contain
-    # info about an exception. We do this by using a function which does
-    # 0/0. Also, the API varies between pytest versions.
-    has_from_call = getattr(runner.CallInfo, "from_call", None) is not None
-    if has_from_call:  # pytest >= 4.1
-        call = runner.CallInfo.from_call(lambda: 0 / 0, "???")
-    else:
-        call = runner.CallInfo(lambda: 0 / 0, "???")
-    call.excinfo = info
-
-    rep = runner.pytest_runtest_makereport(item, call)
-    if result.out:
-        rep.sections.append(("captured stdout", result.out))
-    if result.err:
-        rep.sections.append(("captured stderr", result.err))
-    return rep
diff --git a/tests/artifactcache/artifactservice.py b/tests/artifactcache/artifactservice.py
index 00d14b45d..51f128c32 100644
--- a/tests/artifactcache/artifactservice.py
+++ b/tests/artifactcache/artifactservice.py
@@ -32,7 +32,6 @@ from buildstream import utils
 from tests.testutils.artifactshare import create_artifact_share
 
 
-@pytest.mark.in_subprocess
 def test_artifact_get_not_found(tmpdir):
     sharedir = os.path.join(str(tmpdir), "share")
     with create_artifact_share(sharedir) as share:
@@ -54,7 +53,6 @@ def test_artifact_get_not_found(tmpdir):
 
 
 # Successfully getting the artifact
-@pytest.mark.in_subprocess
 @pytest.mark.parametrize("files", ["present", "absent", "invalid"])
 def test_update_artifact(tmpdir, files):
     sharedir = os.path.join(str(tmpdir), "share")
diff --git a/tests/artifactcache/pull.py b/tests/artifactcache/pull.py
index c32ed94a0..a42a01bf7 100644
--- a/tests/artifactcache/pull.py
+++ b/tests/artifactcache/pull.py
@@ -33,7 +33,6 @@ def tree_maker(cas, tree, directory):
         tree_maker(cas, tree, child_directory)
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.datafiles(DATA_DIR)
 def test_pull(cli, tmpdir, datafiles):
     project_dir = str(datafiles)
@@ -105,7 +104,6 @@ def test_pull(cli, tmpdir, datafiles):
             assert cli.artifact.is_cached(cache_dir, element, element_key)
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.datafiles(DATA_DIR)
 def test_pull_tree(cli, tmpdir, datafiles):
     project_dir = str(datafiles)
diff --git a/tests/artifactcache/push.py b/tests/artifactcache/push.py
index 8b00d0fb7..62c443d61 100644
--- a/tests/artifactcache/push.py
+++ b/tests/artifactcache/push.py
@@ -53,7 +53,6 @@ def _push(cli, cache_dir, project_dir, config_file, target):
     return element_key
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.datafiles(DATA_DIR)
 def test_push(cli, tmpdir, datafiles):
     project_dir = str(datafiles)
@@ -87,7 +86,6 @@ def test_push(cli, tmpdir, datafiles):
         assert share.get_artifact(cli.get_artifact_name(project_dir, 'test', 'target.bst', cache_key=element_key))
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.datafiles(DATA_DIR)
 def test_push_split(cli, tmpdir, datafiles):
     project_dir = str(datafiles)
@@ -131,7 +129,6 @@ def test_push_split(cli, tmpdir, datafiles):
         assert storage.get_cas_files(proto) is not None
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.datafiles(DATA_DIR)
 def test_push_message(tmpdir, datafiles):
     project_dir = str(datafiles)
diff --git a/tests/conftest.py b/tests/conftest.py
index 4aa7be7fb..4bd226afb 100755
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -25,7 +25,6 @@ import pytest
 
 from buildstream.testing import register_repo_kind, sourcetests_collection_hook
 from buildstream.testing._fixtures import reset_global_node_state, thread_check  # pylint: disable=unused-import
-from buildstream.testing._forked import forked_run_report
 from buildstream.testing.integration import integration_cache  # pylint: disable=unused-import
 
 
@@ -71,27 +70,6 @@ def pytest_runtest_setup(item):
 
 
 #################################################
-#              in_subprocess mark               #
-#################################################
-#
-# Various issues can occur when forking the Python process and using gRPC,
-# due to its multithreading. As BuildStream forks for parallelisation, gRPC
-# features are restricted to child processes, so tests using them must also
-# run as child processes. The in_subprocess mark handles this.
-# See <https://github.com/grpc/grpc/blob/master/doc/fork_support.md>.
-#
-@pytest.mark.tryfirst
-def pytest_runtest_protocol(item):
-    if item.get_closest_marker('in_subprocess') is not None:
-        reports = forked_run_report(item)
-        for rep in reports:
-            item.ihook.pytest_runtest_logreport(report=rep)
-        return True
-    else:
-        return None
-
-
-#################################################
 #           remote_services fixture             #
 #################################################
 #
diff --git a/tests/internals/storage.py b/tests/internals/storage.py
index d846e4b1f..a26ca4858 100644
--- a/tests/internals/storage.py
+++ b/tests/internals/storage.py
@@ -25,7 +25,6 @@ def setup_backend(backend_class, tmpdir):
             cas_cache.release_resources()
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.parametrize("backend", [
     FileBasedDirectory, CasBasedDirectory])
 @pytest.mark.datafiles(DATA_DIR)
@@ -39,7 +38,6 @@ def test_import(tmpdir, datafiles, backend):
         assert "bin/hello" in c.list_relative_paths()
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.parametrize("backend", [
     FileBasedDirectory, CasBasedDirectory])
 @pytest.mark.datafiles(DATA_DIR)
diff --git a/tests/internals/storage_vdir_import.py b/tests/internals/storage_vdir_import.py
index 8fbf4142f..808e1be9a 100644
--- a/tests/internals/storage_vdir_import.py
+++ b/tests/internals/storage_vdir_import.py
@@ -233,14 +233,12 @@ def _import_test(tmpdir, original, overlay, generator_function, verify_contents=
         cas_cache.release_resources()
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.parametrize("original", range(1, len(root_filesets) + 1))
 @pytest.mark.parametrize("overlay", range(1, len(root_filesets) + 1))
 def test_fixed_cas_import(tmpdir, original, overlay):
     _import_test(str(tmpdir), original, overlay, generate_import_roots, verify_contents=True)
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.parametrize("original", range(1, NUM_RANDOM_TESTS + 1))
 @pytest.mark.parametrize("overlay", range(1, NUM_RANDOM_TESTS + 1))
 def test_random_cas_import(tmpdir, original, overlay):
@@ -264,20 +262,17 @@ def _listing_test(tmpdir, root, generator_function):
         cas_cache.release_resources()
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.parametrize("root", range(1, NUM_RANDOM_TESTS + 1))
 def test_random_directory_listing(tmpdir, root):
     _listing_test(str(tmpdir), root, generate_random_root)
 
 
-@pytest.mark.in_subprocess
 @pytest.mark.parametrize("root", range(1, len(root_filesets) + 1))
 def test_fixed_directory_listing(tmpdir, root):
     _listing_test(str(tmpdir), root, generate_import_roots)
 
 
 # Check that the vdir is decending and readable
-@pytest.mark.in_subprocess
 def test_descend(tmpdir):
     cas_dir = os.path.join(str(tmpdir), 'cas')
     cas_cache = CASCache(cas_dir)
@@ -304,7 +299,6 @@ def test_descend(tmpdir):
 # Check symlink logic for edgecases
 # Make sure the correct erros are raised when trying
 # to decend in to files or links to files
-@pytest.mark.in_subprocess
 def test_bad_symlinks(tmpdir):
     cas_dir = os.path.join(str(tmpdir), 'cas')
     cas_cache = CASCache(cas_dir)
@@ -338,7 +332,6 @@ def test_bad_symlinks(tmpdir):
 
 # Check symlink logic for edgecases
 # Check decend accross relitive link
-@pytest.mark.in_subprocess
 def test_relative_symlink(tmpdir):
     cas_dir = os.path.join(str(tmpdir), 'cas')
     cas_cache = CASCache(cas_dir)
@@ -364,7 +357,6 @@ def test_relative_symlink(tmpdir):
 
 # Check symlink logic for edgecases
 # Check deccend accross abs link
-@pytest.mark.in_subprocess
 def test_abs_symlink(tmpdir):
     cas_dir = os.path.join(str(tmpdir), 'cas')
     cas_cache = CASCache(cas_dir)
@@ -391,7 +383,6 @@ def test_abs_symlink(tmpdir):
 
 # Check symlink logic for edgecases
 # Check symlink can not escape root
-@pytest.mark.in_subprocess
 def test_bad_sym_escape(tmpdir):
     cas_dir = os.path.join(str(tmpdir), 'cas')
     cas_cache = CASCache(cas_dir)