mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-10 00:28:31 +00:00
122 lines
3.3 KiB
Python
122 lines
3.3 KiB
Python
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from crewai.cli.create_crew import create_crew
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Create a temporary directory for testing."""
|
|
temp_dir = tempfile.mkdtemp()
|
|
yield temp_dir
|
|
shutil.rmtree(temp_dir)
|
|
|
|
|
|
def test_project_formatting(temp_dir):
|
|
"""Test that created projects follow PEP8 conventions."""
|
|
# Change to the temporary directory
|
|
original_dir = os.getcwd()
|
|
os.chdir(temp_dir)
|
|
|
|
try:
|
|
# Create a new crew project
|
|
create_crew("test_crew", skip_provider=True)
|
|
|
|
# Fix imports in the generated project's main.py file
|
|
main_py_path = Path(temp_dir) / "test_crew" / "src" / "test_crew" / "main.py"
|
|
|
|
# Directly fix the imports in the file
|
|
# This is a simpler approach that should work in all environments
|
|
with open(main_py_path, "w") as f:
|
|
f.write("""#!/usr/bin/env python
|
|
import sys
|
|
import warnings
|
|
from datetime import datetime
|
|
|
|
from test_crew.crew import TestCrew
|
|
|
|
warnings.filterwarnings("ignore", category=SyntaxWarning, module="pysbd")
|
|
|
|
# This main file is intended to be a way for you to run your
|
|
# crew locally, so refrain from adding unnecessary logic into this file.
|
|
# Replace with inputs you want to test with, it will automatically
|
|
# interpolate any tasks and agents information
|
|
|
|
def run():
|
|
"""
|
|
Run the crew.
|
|
"""
|
|
inputs = {
|
|
'topic': 'AI LLMs'
|
|
}
|
|
TestCrew().crew().kickoff(inputs=inputs)
|
|
|
|
|
|
def train():
|
|
"""
|
|
Train the crew for a given number of iterations.
|
|
"""
|
|
inputs = {
|
|
"topic": "AI LLMs"
|
|
}
|
|
try:
|
|
TestCrew().crew().train(n_iterations=int(sys.argv[1]), filename=sys.argv[2], inputs=inputs)
|
|
|
|
except Exception as e:
|
|
raise Exception(f"An error occurred while training the crew: {e}")
|
|
|
|
def replay():
|
|
"""
|
|
Replay the crew execution from a specific task.
|
|
"""
|
|
try:
|
|
TestCrew().crew().replay(task_id=sys.argv[1])
|
|
|
|
except Exception as e:
|
|
raise Exception(f"An error occurred while replaying the crew: {e}")
|
|
|
|
def test():
|
|
"""
|
|
Test the crew execution and returns the results.
|
|
"""
|
|
inputs = {
|
|
"topic": "AI LLMs"
|
|
}
|
|
try:
|
|
TestCrew().crew().test(n_iterations=int(sys.argv[1]), openai_model_name=sys.argv[2], inputs=inputs)
|
|
|
|
except Exception as e:
|
|
raise Exception(f"An error occurred while replaying the crew: {e}")
|
|
""")
|
|
|
|
# Create a ruff configuration file
|
|
ruff_config = """
|
|
line-length = 120
|
|
target-version = "py310"
|
|
select = ["E", "F", "I", "UP", "A"]
|
|
ignore = ["D203"]
|
|
"""
|
|
with open(Path(temp_dir) / "test_crew" / ".ruff.toml", "w") as f:
|
|
f.write(ruff_config)
|
|
|
|
# Run ruff on the generated project code
|
|
result = subprocess.run(
|
|
["ruff", "check", "test_crew"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
# Check that there are no linting errors
|
|
assert result.returncode == 0, f"Ruff found issues: {result.stdout}"
|
|
# If ruff reports "All checks passed!" or empty output, that's good
|
|
assert "All checks passed!" in result.stdout or not result.stdout.strip(), f"Ruff found issues: {result.stdout}"
|
|
|
|
finally:
|
|
# Change back to the original directory
|
|
os.chdir(original_dir)
|