ref: Use pathlib in tests (#4159)

Use pathlib in tests

scripts/ci/update_lf_base_dependency.py

@@ -1,17 +1,16 @@
-import os
 import sys
 import re
+from pathlib import Path
 
 import packaging.version
 
-BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
+BASE_DIR = Path(__file__).parent.parent.parent
 
 
 def update_base_dep(pyproject_path: str, new_version: str) -> None:
     """Update the langflow-base dependency in pyproject.toml."""
-    filepath = os.path.join(BASE_DIR, pyproject_path)
-    with open(filepath, "r") as file:
-        content = file.read()
+    filepath = BASE_DIR / pyproject_path
+    content = filepath.read_text()
 
     replacement = f'langflow-base-nightly = "{new_version}"'
 
@@ -20,8 +19,7 @@ def update_base_dep(pyproject_path: str, new_version: str) -> None:
     if not pattern.search(content):
         raise Exception(f'langflow-base poetry dependency not found in "{filepath}"')
     content = pattern.sub(replacement, content)
-    with open(filepath, "w") as file:
-        file.write(content)
+    filepath.write_text(content)
 
 
 def verify_pep440(version):

scripts/ci/update_pyproject_name.py

@@ -1,15 +1,14 @@
-import os
 import sys
 import re
+from pathlib import Path
 
-BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
+BASE_DIR = Path(__file__).parent.parent.parent
 
 
 def update_pyproject_name(pyproject_path: str, new_project_name: str) -> None:
     """Update the project name in pyproject.toml."""
-    filepath = os.path.join(BASE_DIR, pyproject_path)
-    with open(filepath, "r") as file:
-        content = file.read()
+    filepath = BASE_DIR / pyproject_path
+    content = filepath.read_text()
 
     # Regex to match the version line under [tool.poetry]
     pattern = re.compile(r'(?<=^name = ")[^"]+(?=")', re.MULTILINE)
@@ -18,15 +17,13 @@ def update_pyproject_name(pyproject_path: str, new_project_name: str) -> None:
         raise Exception(f'Project name not found in "{filepath}"')
     content = pattern.sub(new_project_name, content)
 
-    with open(filepath, "w") as file:
-        file.write(content)
+    filepath.write_text(content)
 
 
 def update_uv_dep(pyproject_path: str, new_project_name: str) -> None:
     """Update the langflow-base dependency in pyproject.toml."""
-    filepath = os.path.join(BASE_DIR, pyproject_path)
-    with open(filepath, "r") as file:
-        content = file.read()
+    filepath = BASE_DIR / pyproject_path
+    content = filepath.read_text()
 
     if new_project_name == "langflow-nightly":
         pattern = re.compile(r"langflow = \{ workspace = true \}")
@@ -41,8 +38,7 @@ def update_uv_dep(pyproject_path: str, new_project_name: str) -> None:
     if not pattern.search(content):
         raise Exception(f"{replacement} uv dependency not found in {filepath}")
     content = pattern.sub(replacement, content)
-    with open(filepath, "w") as file:
-        file.write(content)
+    filepath.write_text(content)
 
 
 def main() -> None:

scripts/ci/update_pyproject_version.py

@@ -1,17 +1,16 @@
-import os
 import sys
 import re
+from pathlib import Path
 
 import packaging.version
 
-BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
+BASE_DIR = Path(__file__).parent.parent.parent
 
 
 def update_pyproject_version(pyproject_path: str, new_version: str) -> None:
     """Update the version in pyproject.toml."""
-    filepath = os.path.join(BASE_DIR, pyproject_path)
-    with open(filepath, "r") as file:
-        content = file.read()
+    filepath = BASE_DIR / pyproject_path
+    content = filepath.read_text()
 
     # Regex to match the version line under [tool.poetry]
     pattern = re.compile(r'(?<=^version = ")[^"]+(?=")', re.MULTILINE)
@@ -21,8 +20,7 @@ def update_pyproject_version(pyproject_path: str, new_version: str) -> None:
 
     content = pattern.sub(new_version, content)
 
-    with open(filepath, "w") as file:
-        file.write(content)
+    filepath.write_text(content)
 
 
 def verify_pep440(version):

scripts/ci/update_uv_dependency.py

@@ -1,18 +1,17 @@
-import os
 import sys
 import re
+from pathlib import Path
 
-BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))
+BASE_DIR = Path(__file__).parent.parent.parent
 
 
 def update_uv_dep(base_version: str) -> None:
     """Update the langflow-base dependency in pyproject.toml."""
 
-    pyproject_path = os.path.join(BASE_DIR, "pyproject.toml")
+    pyproject_path = BASE_DIR / "pyproject.toml"
 
     # Read the pyproject.toml file content
-    with open(pyproject_path, "r") as file:
-        content = file.read()
+    content = pyproject_path.read_text()
 
     # For the main project, update the langflow-base dependency in the UV section
     pattern = re.compile(r'(dependencies\s*=\s*\[\s*\n\s*)("langflow-base==[\d.]+")')
@@ -26,8 +25,7 @@ def update_uv_dep(base_version: str) -> None:
     content = pattern.sub(replacement, content)
 
     # Write the updated content back to the file
-    with open(pyproject_path, "w") as file:
-        file.write(content)
+    pyproject_path.write_text(content)
 
 
 def main() -> None:

src/backend/tests/conftest.py

@@ -1,5 +1,4 @@
 import json
-import os.path
 import shutil
 
 # we need to import tmpdir
@@ -198,7 +197,7 @@ def get_graph(_type="basic"):
     elif _type == "openapi":
         path = pytest.OPENAPI_EXAMPLE_PATH
 
-    with open(path) as f:
+    with path.open() as f:
         flow_graph = json.load(f)
     data_graph = flow_graph["data"]
     nodes = data_graph["nodes"]
@@ -210,7 +209,7 @@ def get_graph(_type="basic"):
 
 @pytest.fixture
 def basic_graph_data():
-    with open(pytest.BASIC_EXAMPLE_PATH) as f:
+    with pytest.BASIC_EXAMPLE_PATH.open() as f:
         return json.load(f)
 
 
@@ -231,56 +230,47 @@ def openapi_graph():
 
 @pytest.fixture
 def json_flow():
-    with open(pytest.BASIC_EXAMPLE_PATH) as f:
-        return f.read()
+    return pytest.BASIC_EXAMPLE_PATH.read_text()
 
 
 @pytest.fixture
 def grouped_chat_json_flow():
-    with open(pytest.GROUPED_CHAT_EXAMPLE_PATH) as f:
-        return f.read()
+    return pytest.GROUPED_CHAT_EXAMPLE_PATH.read_text()
 
 
 @pytest.fixture
 def one_grouped_chat_json_flow():
-    with open(pytest.ONE_GROUPED_CHAT_EXAMPLE_PATH) as f:
-        return f.read()
+    return pytest.ONE_GROUPED_CHAT_EXAMPLE_PATH.read_text()
 
 
 @pytest.fixture
 def vector_store_grouped_json_flow():
-    with open(pytest.VECTOR_STORE_GROUPED_EXAMPLE_PATH) as f:
-        return f.read()
+    return pytest.VECTOR_STORE_GROUPED_EXAMPLE_PATH.read_text()
 
 
 @pytest.fixture
 def json_flow_with_prompt_and_history():
-    with open(pytest.BASIC_CHAT_WITH_PROMPT_AND_HISTORY) as f:
-        return f.read()
+    return pytest.BASIC_CHAT_WITH_PROMPT_AND_HISTORY.read_text()
 
 
 @pytest.fixture
 def json_simple_api_test():
-    with open(pytest.SIMPLE_API_TEST) as f:
-        return f.read()
+    return pytest.SIMPLE_API_TEST.read_text()
 
 
 @pytest.fixture
 def json_vector_store():
-    with open(pytest.VECTOR_STORE_PATH) as f:
-        return f.read()
+    return pytest.VECTOR_STORE_PATH.read_text()
 
 
 @pytest.fixture
 def json_webhook_test():
-    with open(pytest.WEBHOOK_TEST) as f:
-        return f.read()
+    return pytest.WEBHOOK_TEST.read_text()
 
 
 @pytest.fixture
 def json_memory_chatbot_no_llm():
-    with open(pytest.MEMORY_CHATBOT_NO_LLM) as f:
-        return f.read()
+    return pytest.MEMORY_CHATBOT_NO_LLM.read_text()
 
 
 @pytest.fixture(name="client", autouse=True)
@@ -295,7 +285,7 @@ async def client_fixture(session: Session, monkeypatch, request, load_flows_dir)
         monkeypatch.setenv("LANGFLOW_AUTO_LOGIN", "false")
         if "load_flows" in request.keywords:
             shutil.copyfile(
-                pytest.BASIC_EXAMPLE_PATH, os.path.join(load_flows_dir, "c54f9130-f2fa-4a3e-b22a-3856d946351b.json")
+                pytest.BASIC_EXAMPLE_PATH, Path(load_flows_dir) / "c54f9130-f2fa-4a3e-b22a-3856d946351b.json"
             )
             monkeypatch.setenv("LANGFLOW_LOAD_FLOWS_PATH", load_flows_dir)
             monkeypatch.setenv("LANGFLOW_AUTO_LOGIN", "true")
@@ -405,14 +395,12 @@ def flow(client, json_flow: str, active_user):
 
 @pytest.fixture
 def json_chat_input():
-    with open(pytest.CHAT_INPUT) as f:
-        yield f.read()
+    return pytest.CHAT_INPUT.read_text()
 
 
 @pytest.fixture
 def json_two_outputs():
-    with open(pytest.TWO_OUTPUTS) as f:
-        yield f.read()
+    return pytest.TWO_OUTPUTS.read_text()
 
 
 @pytest.fixture

src/backend/tests/locust/locustfile.py

@@ -11,7 +11,7 @@ from rich import print
 class NameTest(FastHttpUser):
     wait_time = between(1, 5)
 
-    with open("names.txt", "r") as file:
+    with Path("names.txt").open() as file:
         names = [line.strip() for line in file.readlines()]
 
     headers: dict = {}
@@ -86,11 +86,7 @@ class NameTest(FastHttpUser):
         a_token = tokens["access_token"]
         logged_in_headers = {"Authorization": f"Bearer {a_token}"}
         print("Logged in")
-        with open(
-            Path(__file__).parent.parent / "data" / "BasicChatwithPromptandHistory.json",
-            "r",
-        ) as f:
-            json_flow = f.read()
+        json_flow = (Path(__file__).parent.parent / "data" / "BasicChatwithPromptandHistory.json").read_text()
         flow = orjson.loads(json_flow)
         data = flow["data"]
         # Create test data

src/backend/tests/unit/test_custom_component.py

@@ -1,5 +1,6 @@
 import ast
 import types
+from pathlib import Path
 from textwrap import dedent
 
 import pytest
@@ -18,9 +19,8 @@ def client():
 
 @pytest.fixture
 def code_component_with_multiple_outputs():
-    with open("src/backend/tests/data/component_multiple_outputs.py") as f:
-        code = f.read()
-        return Component(_code=code)
+    code = Path("src/backend/tests/data/component_multiple_outputs.py").read_text()
+    return Component(_code=code)
 
 
 code_default = """

src/backend/tests/unit/test_custom_component_with_client.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 import pytest
 
 from langflow.custom import Component
@@ -9,9 +11,8 @@ from langflow.services.settings.feature_flags import FEATURE_FLAGS
 
 @pytest.fixture
 def code_component_with_multiple_outputs():
-    with open("src/backend/tests/data/component_multiple_outputs.py") as f:
-        code = f.read()
-        return Component(_code=code)
+    code = Path("src/backend/tests/data/component_multiple_outputs.py").read_text()
+    return Component(_code=code)
 
 
 @pytest.fixture

src/backend/tests/unit/test_data_components.py

@@ -1,4 +1,3 @@
-import os
 import tempfile
 from pathlib import Path
 from unittest.mock import Mock, patch, ANY
@@ -128,7 +127,7 @@ def test_directory_component_build_with_multithreading(
 ):
     # Arrange
     directory_component = data.DirectoryComponent()
-    path = os.path.dirname(os.path.abspath(__file__))
+    path = Path(__file__).resolve().parent
     depth = 1
     max_concurrency = 2
     load_hidden = False
@@ -136,16 +135,15 @@ def test_directory_component_build_with_multithreading(
     silent_errors = False
     use_multithreading = True
 
-    mock_resolve_path.return_value = path
-    mock_retrieve_file_paths.return_value = [
-        os.path.join(path, file) for file in os.listdir(path) if file.endswith(".py")
-    ]
+    mock_resolve_path.return_value = str(path)
+
+    mock_retrieve_file_paths.return_value = [str(p) for p in path.iterdir() if p.suffix == ".py"]
     mock_parallel_load_data.return_value = [Mock()]
 
     # Act
     directory_component.set_attributes(
         {
-            "path": path,
+            "path": str(path),
             "depth": depth,
             "max_concurrency": max_concurrency,
             "load_hidden": load_hidden,
@@ -157,9 +155,9 @@ def test_directory_component_build_with_multithreading(
     directory_component.load_directory()
 
     # Assert
-    mock_resolve_path.assert_called_once_with(path)
+    mock_resolve_path.assert_called_once_with(str(path))
     mock_retrieve_file_paths.assert_called_once_with(
-        path, load_hidden=load_hidden, recursive=recursive, depth=depth, types=ANY
+        str(path), load_hidden=load_hidden, recursive=recursive, depth=depth, types=ANY
     )
     mock_parallel_load_data.assert_called_once_with(
         mock_retrieve_file_paths.return_value, silent_errors=silent_errors, max_concurrency=max_concurrency
@@ -170,11 +168,9 @@ def test_directory_without_mocks():
     directory_component = data.DirectoryComponent()
 
     with tempfile.TemporaryDirectory() as temp_dir:
-        with open(temp_dir + "/test.txt", "w") as f:
-            f.write("test")
+        (Path(temp_dir) / "test.txt").write_text("test")
         # also add a json file
-        with open(temp_dir + "/test.json", "w") as f:
-            f.write('{"test": "test"}')
+        (Path(temp_dir) / "test.json").write_text('{"test": "test"}')
 
         directory_component.set_attributes({"path": str(temp_dir), "use_multithreading": False})
         results = directory_component.load_directory()

src/backend/tests/unit/test_files.py

@@ -1,4 +1,3 @@
-import os
 import re
 import shutil
 import tempfile
@@ -39,7 +38,7 @@ async def files_client_fixture(session: Session, monkeypatch, request, load_flow
         monkeypatch.setenv("LANGFLOW_AUTO_LOGIN", "false")
         if "load_flows" in request.keywords:
             shutil.copyfile(
-                pytest.BASIC_EXAMPLE_PATH, os.path.join(load_flows_dir, "c54f9130-f2fa-4a3e-b22a-3856d946351b.json")
+                pytest.BASIC_EXAMPLE_PATH, Path(load_flows_dir) / "c54f9130-f2fa-4a3e-b22a-3856d946351b.json"
             )
             monkeypatch.setenv("LANGFLOW_LOAD_FLOWS_PATH", load_flows_dir)
             monkeypatch.setenv("LANGFLOW_AUTO_LOGIN", "true")

src/backend/tests/unit/test_helper_components.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 from langflow.components import helpers
 from langflow.custom.utils import build_custom_component_template
 from langflow.schema import Data
@@ -40,7 +42,7 @@ def client():
 def test_uuid_generator_component():
     # Arrange
     uuid_generator_component = helpers.IDGeneratorComponent()
-    uuid_generator_component._code = open(helpers.IDGenerator.__file__).read()
+    uuid_generator_component._code = Path(helpers.IDGenerator.__file__).read_text()
 
     frontend_node, _ = build_custom_component_template(uuid_generator_component)