mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-07-02 13:48:09 +00:00
570c7e124f
- Added `is_dmn_mode_enabled` utility to check for enterprise non-interactive mode based on the `CREWAI_DMN` environment variable. - Updated `create` function in `cli.py` to enforce required parameters when DMN mode is active, raising appropriate usage errors. - Enhanced `create_crew` and `create_json_crew` functions to skip provider prompts and handle folder existence checks in DMN mode. - Introduced non-interactive defaults for agent and task creation in DMN mode, ensuring seamless project setup without user input. - Modified `run_crew` to bypass TUI and handle runtime inputs directly when in DMN mode, improving execution flow for JSON-defined crews. - Added tests to validate DMN mode behavior, ensuring correct handling of required inputs and non-interactive defaults.
125 lines
4.2 KiB
Python
125 lines
4.2 KiB
Python
import os
|
|
import shutil
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from crewai_cli import utils
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_tree():
|
|
root_dir = tempfile.mkdtemp()
|
|
|
|
create_file(os.path.join(root_dir, "file1.txt"), "Hello, world!")
|
|
create_file(os.path.join(root_dir, "file2.txt"), "Another file")
|
|
os.mkdir(os.path.join(root_dir, "empty_dir"))
|
|
nested_dir = os.path.join(root_dir, "nested_dir")
|
|
os.mkdir(nested_dir)
|
|
create_file(os.path.join(nested_dir, "nested_file.txt"), "Nested content")
|
|
|
|
yield root_dir
|
|
|
|
shutil.rmtree(root_dir)
|
|
|
|
|
|
def create_file(path, content):
|
|
with open(path, "w") as f:
|
|
f.write(content)
|
|
|
|
|
|
def test_tree_find_and_replace_file_content(temp_tree):
|
|
utils.tree_find_and_replace(temp_tree, "world", "universe")
|
|
with open(os.path.join(temp_tree, "file1.txt"), "r") as f:
|
|
assert f.read() == "Hello, universe!"
|
|
|
|
|
|
def test_tree_find_and_replace_file_name(temp_tree):
|
|
old_path = os.path.join(temp_tree, "file2.txt")
|
|
new_path = os.path.join(temp_tree, "file2_renamed.txt")
|
|
os.rename(old_path, new_path)
|
|
utils.tree_find_and_replace(temp_tree, "renamed", "modified")
|
|
assert os.path.exists(os.path.join(temp_tree, "file2_modified.txt"))
|
|
assert not os.path.exists(new_path)
|
|
|
|
|
|
def test_tree_find_and_replace_directory_name(temp_tree):
|
|
utils.tree_find_and_replace(temp_tree, "empty", "renamed")
|
|
assert os.path.exists(os.path.join(temp_tree, "renamed_dir"))
|
|
assert not os.path.exists(os.path.join(temp_tree, "empty_dir"))
|
|
|
|
|
|
def test_tree_find_and_replace_nested_content(temp_tree):
|
|
utils.tree_find_and_replace(temp_tree, "Nested", "Updated")
|
|
with open(os.path.join(temp_tree, "nested_dir", "nested_file.txt"), "r") as f:
|
|
assert f.read() == "Updated content"
|
|
|
|
|
|
def test_tree_find_and_replace_no_matches(temp_tree):
|
|
utils.tree_find_and_replace(temp_tree, "nonexistent", "replacement")
|
|
assert set(os.listdir(temp_tree)) == {
|
|
"file1.txt",
|
|
"file2.txt",
|
|
"empty_dir",
|
|
"nested_dir",
|
|
}
|
|
|
|
|
|
def test_tree_copy_full_structure(temp_tree):
|
|
dest_dir = tempfile.mkdtemp()
|
|
try:
|
|
utils.tree_copy(temp_tree, dest_dir)
|
|
assert set(os.listdir(dest_dir)) == set(os.listdir(temp_tree))
|
|
assert os.path.isfile(os.path.join(dest_dir, "file1.txt"))
|
|
assert os.path.isfile(os.path.join(dest_dir, "file2.txt"))
|
|
assert os.path.isdir(os.path.join(dest_dir, "empty_dir"))
|
|
assert os.path.isdir(os.path.join(dest_dir, "nested_dir"))
|
|
assert os.path.isfile(os.path.join(dest_dir, "nested_dir", "nested_file.txt"))
|
|
finally:
|
|
shutil.rmtree(dest_dir)
|
|
|
|
|
|
def test_tree_copy_preserve_content(temp_tree):
|
|
dest_dir = tempfile.mkdtemp()
|
|
try:
|
|
utils.tree_copy(temp_tree, dest_dir)
|
|
with open(os.path.join(dest_dir, "file1.txt"), "r") as f:
|
|
assert f.read() == "Hello, world!"
|
|
with open(os.path.join(dest_dir, "nested_dir", "nested_file.txt"), "r") as f:
|
|
assert f.read() == "Nested content"
|
|
finally:
|
|
shutil.rmtree(dest_dir)
|
|
|
|
|
|
def test_tree_copy_to_existing_directory(temp_tree):
|
|
dest_dir = tempfile.mkdtemp()
|
|
try:
|
|
create_file(os.path.join(dest_dir, "existing_file.txt"), "I was here first")
|
|
utils.tree_copy(temp_tree, dest_dir)
|
|
assert os.path.isfile(os.path.join(dest_dir, "existing_file.txt"))
|
|
assert os.path.isfile(os.path.join(dest_dir, "file1.txt"))
|
|
finally:
|
|
shutil.rmtree(dest_dir)
|
|
|
|
|
|
@pytest.mark.parametrize("value", ["1", "true", "yes", "anything"])
|
|
def test_is_dmn_mode_enabled_for_truthy_values(monkeypatch, value):
|
|
monkeypatch.setenv("CREWAI_DMN", value)
|
|
|
|
assert utils.is_dmn_mode_enabled() is True
|
|
|
|
|
|
@pytest.mark.parametrize("value", [None, "", "0", "false", "no", "off"])
|
|
def test_is_dmn_mode_enabled_for_falsey_values(monkeypatch, value):
|
|
if value is None:
|
|
monkeypatch.delenv("CREWAI_DMN", raising=False)
|
|
else:
|
|
monkeypatch.setenv("CREWAI_DMN", value)
|
|
|
|
assert utils.is_dmn_mode_enabled() is False
|
|
|
|
|
|
# Tests for extract_available_exports, get_crews, get_flows, fetch_crews,
|
|
# is_valid_tool live in lib/crewai/tests/cli/test_utils.py — the canonical
|
|
# implementations are in crewai.utilities.project_utils.
|