Parallelize test suite with pytest-xdist

.github/workflows/benchmark.yml

@@ -61,8 +61,13 @@ jobs:
           current_benchmark_results.json
         retention-days: 30
 
-    # Comment on PR with benchmark results
+    # Comment on PR with benchmark results.
+    # A pull_request run from a fork gets a read-only GITHUB_TOKEN no matter what
+    # the permissions block above says, so commenting is impossible there and the
+    # attempt fails the whole job with a 403. The report is still in the uploaded
+    # artifact either way.
     - name: Comment benchmark results on PR
+      if: github.event.pull_request.head.repo.full_name == github.repository
       uses: actions/github-script@v9
       with:
         script: |

.github/workflows/tests.yml

@@ -38,7 +38,7 @@ jobs:
         env:
           BBOT_IO_API_KEY: ${{ secrets.BBOT_IO_API_KEY }}
         run: |
-          uv run pytest -vv --reruns 2 -o timeout_func_only=true --timeout 1200 --disable-warnings --log-cli-level=INFO --cov-config=bbot/test/coverage.cfg --cov-report xml:cov.xml --cov=bbot .
+          uv run pytest -vv -n $(python bbot/test/worker_count.py) --dist loadgroup --reruns 2 -o timeout_func_only=true --timeout 1200 --disable-warnings --log-cli-level=INFO --cov-config=bbot/test/coverage.cfg --cov-report xml:cov.xml --cov=bbot .
       - name: Upload Debug Logs
         if: always()
         uses: actions/upload-artifact@v7

bbot/test/bbot_fixtures.py

@@ -15,12 +15,15 @@
 from bbot.core.config.merge import deep_merge
 from bbot.scanner import Preset
 from bbot.core.helpers.misc import mkdir, rand_string
+from bbot.test.worker import BBOT_TEST_DIR
 
 
 log = logging.getLogger("bbot.test.fixtures")
 
 
-bbot_test_dir = Path("/tmp/.bbot_test")
+# Per-xdist-worker so parallel workers don't share caches, scan output, or the
+# sessionfinish cleanup. Serial runs still get /tmp/.bbot_test.
+bbot_test_dir = BBOT_TEST_DIR
 mkdir(bbot_test_dir)
 
 

bbot/test/conftest.py

@@ -10,6 +10,13 @@
 from contextlib import suppress
 from pytest_httpserver import HTTPServer
 
+from bbot.test.worker import (
+    BBOT_TEST_DIR,
+    HTTPSERVER_ALLINTERFACES_PORT,
+    HTTPSERVER_PORT,
+    HTTPSERVER_SSL_PORT,
+)
+
 from bbot.core import CORE
 from bbot.core.helpers.misc import execute_sync_or_async
 from bbot.core.helpers.interactsh import server_list as interactsh_servers
@@ -26,12 +33,28 @@
 with open(Path(__file__).parent / "test.conf") as _f:
     test_config = yaml.safe_load(_f) or {}
 
+# Give each xdist worker its own BBOT home. test.conf carries the serial default;
+# under -n the workers would otherwise share caches, scan output and temp files,
+# and the sessionfinish cleanup below would delete a directory still in use.
+test_config["home"] = str(BBOT_TEST_DIR)
+
 os.environ["BBOT_DEBUG"] = "True"
 CORE.logger.log_level = logging.DEBUG
 
 # silence all stderr output:
 stderr_handler = CORE.logger.log_handlers["stderr"]
 stderr_handler.setLevel(logging.CRITICAL)
+# worker.py clears _BBOT_LOGGING_SETUP for xdist workers, so every process that
+# reaches here (serial or worker) owns a real QueueListener. If it is missing,
+# logging never got set up and debug.log would silently stay empty, so fail
+# loudly instead of continuing with logging quietly broken.
+if CORE.logger.listener is None:
+    raise RuntimeError(
+        "BBOT logging was not initialized in this process "
+        f"(PYTEST_XDIST_WORKER={os.environ.get('PYTEST_XDIST_WORKER', '')!r}). "
+        "debug.log would be empty and log-reading tests would fail with "
+        "confusing assertion errors."
+    )
 handlers = list(CORE.logger.listener.handlers)
 handlers.remove(stderr_handler)
 CORE.logger.listener.handlers = tuple(handlers)
@@ -96,7 +119,7 @@ def stop_server(server):
 
 @pytest.fixture
 def bbot_httpserver():
-    server = HTTPServer(host="127.0.0.1", port=8888, threaded=True)
+    server = HTTPServer(host="127.0.0.1", port=HTTPSERVER_PORT, threaded=True)
     server.start()
 
     yield server
@@ -115,7 +138,7 @@ def bbot_httpserver_ssl():
     keyfile = str(current_dir / "testsslkey.pem")
     certfile = str(current_dir / "testsslcert.pem")
     context.load_cert_chain(certfile, keyfile)
-    server = HTTPServer(host="127.0.0.1", port=9999, ssl_context=context, threaded=True)
+    server = HTTPServer(host="127.0.0.1", port=HTTPSERVER_SSL_PORT, ssl_context=context, threaded=True)
     server.start()
 
     yield server
@@ -222,7 +245,7 @@ async def patched_request_batch_stream(self, urls, threads=10, **kwargs):
 
 @pytest.fixture
 def bbot_httpserver_allinterfaces():
-    server = HTTPServer(host="0.0.0.0", port=5556, threaded=True)
+    server = HTTPServer(host="0.0.0.0", port=HTTPSERVER_ALLINTERFACES_PORT, threaded=True)
     server.start()
 
     yield server
@@ -334,6 +357,20 @@ def proxy_server():
     server_thread.join()
 
 
+def pytest_collection_modifyitems(config, items):
+    """Pin docker-backed tests to one xdist worker.
+
+    They start real containers with fixed names and fixed host port bindings
+    (kafka, elastic, mongo, mysql, nats, postgres, rabbitmq), so two workers
+    running them at once fight over both. They already mark themselves with
+    ``skip_distro_tests``, so reuse that as the signal.
+    """
+    for item in items:
+        cls = getattr(item, "cls", None)
+        if cls is not None and getattr(cls, "skip_distro_tests", False):
+            item.add_marker(pytest.mark.xdist_group("docker"))
+
+
 def pytest_terminal_summary(terminalreporter, exitstatus, config):  # pragma: no cover
     RED = "\033[1;31m"
     GREEN = "\033[1;32m"
@@ -460,8 +497,10 @@ def pytest_sessionfinish(session, exitstatus):
             if child.is_alive():
                 child.kill()
 
-    # Wipe out BBOT home dir
-    shutil.rmtree("/tmp/.bbot_test", ignore_errors=True)
+    # Wipe out BBOT home dir. Scoped to this worker: under xdist the first
+    # worker to finish would otherwise delete the directory out from under
+    # every worker still running.
+    shutil.rmtree(BBOT_TEST_DIR, ignore_errors=True)
 
     # Ensure stdout/stderr are blocking before pytest writes summaries
     try:

bbot/test/test_step_1/test_bbot_fastapi.py

@@ -7,14 +7,15 @@
 from urllib.request import urlopen, Request
 from urllib.error import URLError
 from urllib.parse import urlencode
+from bbot.test.worker import FASTAPI_PORT, FASTAPI_URL, HTTPSERVER_URL
 
 cwd = Path(__file__).parent.parent.parent
 
 
 def run_bbot_multiprocess(queue):
     from bbot.scanner import Scanner
 
-    scan = Scanner("http://127.0.0.1:8888", "blacklanternsecurity.com", modules=["http"])
+    scan = Scanner(HTTPSERVER_URL, "blacklanternsecurity.com", modules=["http"])
     events = [e.json() for e in scan.start()]
     queue.put(events)
 
@@ -42,7 +43,7 @@ def test_bbot_fastapi(bbot_httpserver):
         start_time = time.time()
         while True:
             try:
-                response = urlopen("http://127.0.0.1:8978/ping")
+                response = urlopen(f"{FASTAPI_URL}/ping")
                 response.read()
                 break
             except (URLError, ConnectionError):
@@ -52,8 +53,8 @@ def test_bbot_fastapi(bbot_httpserver):
                 continue
 
         # run a scan
-        params = urlencode({"targets": ["http://127.0.0.1:8888", "blacklanternsecurity.com"]}, doseq=True)
-        req = Request(f"http://127.0.0.1:8978/start?{params}")
+        params = urlencode({"targets": [HTTPSERVER_URL, "blacklanternsecurity.com"]}, doseq=True)
+        req = Request(f"{FASTAPI_URL}/start?{params}")
         response = urlopen(req, timeout=100)
         events = json.loads(response.read())
         assert len(events) >= 3
@@ -75,6 +76,8 @@ def start_fastapi_server():
         del env["BBOT_TESTING"]
     python_executable = str(sys.executable)
     process = Popen(
-        [python_executable, "-m", "uvicorn", "bbot.test.fastapi_test:app", "--port", "8978"], cwd=cwd, env=env
+        [python_executable, "-m", "uvicorn", "bbot.test.fastapi_test:app", "--port", str(FASTAPI_PORT)],
+        cwd=cwd,
+        env=env,
     )
     return process

bbot/test/test_step_1/test_cli.py

@@ -871,6 +871,12 @@ def test_cli_presets(monkeypatch, capsys, caplog, clean_default_config):
         """
         )
 
+    # These are cleaned up at the end of the test, but an earlier failure (or an
+    # interrupted run) leaves them behind, and then these "not exists" assertions
+    # fail on every later run against the same BBOT home. Clear them up front so
+    # the test establishes its precondition instead of assuming it.
+    shutil.rmtree(output_dir, ignore_errors=True)
+
     assert not output_dir.exists()
     assert not scan_dir.exists()
     assert not output_file.exists()

bbot/test/test_step_1/test_command.py

@@ -1,5 +1,7 @@
 import time
 from ..bbot_fixtures import *
+
+from bbot.test.worker import BBOT_TEST_DIR
 from subprocess import CalledProcessError
 
 
@@ -38,15 +40,15 @@ async def test_command(bbot_scanner):
     assert result.splitlines() == [b"some", b"random", b"stdin"]
 
     # test overflow - run
-    tmpfile_path = Path("/tmp/test_bigfile")
+    tmpfile_path = BBOT_TEST_DIR / "test_bigfile"
     with open(tmpfile_path, "w") as f:
         # write 2MB
         f.write("A" * 1024 * 1024 * 2)
     result = (await scan1.helpers.run(["cat", str(tmpfile_path)], limit=1024 * 64, text=False)).stdout
     assert len(result) == 1024 * 1024 * 2
     tmpfile_path.unlink(missing_ok=True)
     # test overflow - run_live
-    tmpfile_path = Path("/tmp/test_bigfile")
+    tmpfile_path = BBOT_TEST_DIR / "test_bigfile"
     with open(tmpfile_path, "w") as f:
         # write 2MB
         f.write("A" * 10 + "\n")
@@ -119,30 +121,30 @@ async def test_command(bbot_scanner):
     # test sudo + existence of environment variables
     await scan1._prep()
     path_parts = os.environ.get("PATH", "").split(":")
-    assert "/tmp/.bbot_test/tools" in path_parts
+    assert f"{BBOT_TEST_DIR}/tools" in path_parts
     run_lines = (await scan1.helpers.run(["env"])).stdout.splitlines()
     assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_lines
     for line in run_lines:
         if line.startswith("PATH="):
             path_parts = line.split("=", 1)[-1].split(":")
-            assert "/tmp/.bbot_test/tools" in path_parts
+            assert f"{BBOT_TEST_DIR}/tools" in path_parts
     run_lines_sudo = (await scan1.helpers.run(["env"], sudo=True)).stdout.splitlines()
     assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_lines_sudo
     for line in run_lines_sudo:
         if line.startswith("PATH="):
             path_parts = line.split("=", 1)[-1].split(":")
-            assert "/tmp/.bbot_test/tools" in path_parts
+            assert f"{BBOT_TEST_DIR}/tools" in path_parts
     run_live_lines = [l async for l in scan1.helpers.run_live(["env"])]
     assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_live_lines
     for line in run_live_lines:
         if line.startswith("PATH="):
             path_parts = line.strip().split("=", 1)[-1].split(":")
-            assert "/tmp/.bbot_test/tools" in path_parts
+            assert f"{BBOT_TEST_DIR}/tools" in path_parts
     run_live_lines_sudo = [l async for l in scan1.helpers.run_live(["env"], sudo=True)]
     assert "BBOT_WEB_USER_AGENT=BBOT Test User-Agent" in run_live_lines_sudo
     for line in run_live_lines_sudo:
         if line.startswith("PATH="):
             path_parts = line.strip().split("=", 1)[-1].split(":")
-            assert "/tmp/.bbot_test/tools" in path_parts
+            assert f"{BBOT_TEST_DIR}/tools" in path_parts
 
     await scan1._cleanup()

bbot/test/test_step_1/test_excavate_url_regexes.py

@@ -11,6 +11,7 @@
 import yara
 
 from bbot.modules.internal.excavate import excavate
+from bbot.test.worker import HTTPSERVER_URL
 
 
 full_url_regex = excavate.URLExtractor.full_url_regex
@@ -30,7 +31,7 @@
 NON_IPV6_URLS = [
     "http://example.com/",
     "https://www.example.com:8080/path",
-    "http://127.0.0.1:8888/",
+    f"{HTTPSERVER_URL}/",
     "https://asdffoo.test.notreal/some/path",
 ]