ref: Use blockbuster package (#4777)

* Use blockbuster package

* Upgrade blockbuster to v1.1

pyproject.toml

@@ -176,7 +176,7 @@ dev-dependencies = [
     "asgi-lifespan>=2.1.0",
     "pytest-github-actions-annotate-failures>=0.2.0",
     "pytest-codspeed>=3.0.0",
-    "forbiddenfruit>=0.1.4",
+    "blockbuster>=1.1.0,<1.2",
 ]
 
 

src/backend/tests/conftest.py

@@ -13,6 +13,7 @@
 import orjson
 import pytest
 from asgi_lifespan import LifespanManager
+from blockbuster import blockbuster_ctx
 from dotenv import load_dotenv
 from fastapi.testclient import TestClient
 from httpx import ASGITransport, AsyncClient
@@ -34,15 +35,29 @@
 from sqlmodel.pool import StaticPool
 from typer.testing import CliRunner
 
-from tests import blockbuster
 from tests.api_keys import get_openai_api_key
 
 if TYPE_CHECKING:
     from langflow.services.database.service import DatabaseService
 
 
 load_dotenv()
-blockbuster.init()
+
+
+@pytest.fixture(autouse=True)
+def blockbuster():
+    with blockbuster_ctx() as bb:
+        for func in [
+            "io.BufferedReader.read",
+            "io.BufferedWriter.write",
+            "io.TextIOWrapper.read",
+            "io.TextIOWrapper.write",
+        ]:
+            bb.functions[func].can_block_functions.append(("settings/service.py", {"initialize"}))
+        for func in bb.functions:
+            if func.startswith("sqlite3."):
+                bb.functions[func].deactivate()
+        yield bb
 
 
 def pytest_configure(config):