diff --git a/.codecov.yml b/.codecov.yml new file mode 100644 index 0000000..88c95e6 --- /dev/null +++ b/.codecov.yml @@ -0,0 +1,26 @@ +coverage: + status: + project: + default: + target: auto + threshold: 1% + informational: true + patch: + default: + target: 100% + threshold: 1% + informational: false + +comment: + layout: "reach,diff,flags,tree" + behavior: default + require_changes: false + +ignore: + - "docs/" + - "test/" + - "**/test_*.py" + - "setup.py" + - "conftest.py" + - "README.md" + - "LICENSE" diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..63d1770 --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,129 @@ +name: Run Tests + +on: + pull_request: + branches: [ "main" ] + workflow_dispatch: + # Allow manual triggering + +jobs: + test: + runs-on: ubuntu-latest + strategy: + matrix: + test-type: ["integration"] + include: + - test-type: "integration" + pytest-args: "-m 'integration'" + + services: + docker: + image: docker:dind + options: --privileged + + steps: + - name: Checkout code + uses: actions/checkout@v4 + + - name: Set up Python 3.12 + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Set up Docker Buildx + if: matrix.test-type == 'integration' + uses: docker/setup-buildx-action@v3 + + - name: Cache pip dependencies + uses: actions/cache@v4 + with: + path: ~/.cache/pip + key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }} + restore-keys: | + ${{ runner.os }}-pip- + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y build-essential + + - name: Install Python dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + pip install pytest pytest-cov pytest-mock pytest-asyncio + + - name: Install package in development mode + run: | + pip install -e . + + - name: Build Docker images for integration tests + if: matrix.test-type == 'integration' + run: | + # Build the shell server image needed for Docker tests + docker build -f src/microbots/environment/local_docker/image_builder/Dockerfile -t kavyasree261002/shell_server:latest . + + - name: Run ${{ matrix.test-type }} tests + env: + # OpenAI API Configuration + OPEN_AI_KEY: ${{ secrets.OPEN_AI_KEY }} + OPEN_AI_DEPLOYMENT_NAME: ${{ secrets.OPEN_AI_DEPLOYMENT_NAME }} + OPEN_AI_END_POINT: ${{ secrets.OPEN_AI_END_POINT }} + # Azure OpenAI API Configuration + AZURE_OPENAI_API_VERSION: ${{ secrets.AZURE_OPENAI_API_VERSION }} + AZURE_OPENAI_API_KEY: ${{ secrets.AZURE_OPENAI_API_KEY }} + AZURE_OPENAI_ENDPOINT: ${{ secrets.AZURE_OPENAI_ENDPOINT }} + run: | + python -m pytest ${{ matrix.pytest-args }} \ + --cov=src \ + --cov-report=xml \ + --cov-report=term-missing \ + --junitxml=test-results-${{ matrix.test-type }}.xml \ + -v + + - name: Upload test results + uses: actions/upload-artifact@v4 + if: always() + with: + name: test-results-${{ matrix.test-type }} + path: test-results-*.xml + + - name: Upload coverage reports + uses: actions/upload-artifact@v4 + if: always() + with: + name: coverage-${{ matrix.test-type }} + path: coverage.xml + + - name: Upload coverage to Codecov + uses: codecov/codecov-action@v4 + if: always() + with: + token: ${{ secrets.CODECOV_TOKEN }} + file: ./coverage.xml + flags: ${{ matrix.test-type }} + name: codecov-${{ matrix.test-type }} + fail_ci_if_error: false + + test-summary: + runs-on: ubuntu-latest + needs: [test] + if: always() + steps: + - name: Download all test results + uses: actions/download-artifact@v4 + with: + pattern: test-results-* + merge-multiple: true + + - name: Test Summary + if: always() + run: | + echo "## Test Results Summary" >> $GITHUB_STEP_SUMMARY + echo "| Test Type | Status |" >> $GITHUB_STEP_SUMMARY + echo "|-----------|--------|" >> $GITHUB_STEP_SUMMARY + if [ "${{ needs.test.result }}" = "success" ]; then + echo "| Integration Tests | ✅ Passed |" >> $GITHUB_STEP_SUMMARY + else + echo "| Integration Tests | ❌ Failed |" >> $GITHUB_STEP_SUMMARY + fi \ No newline at end of file diff --git a/pytest.ini b/pytest.ini index 42d331a..d0e4f1f 100644 --- a/pytest.ini +++ b/pytest.ini @@ -1,4 +1,15 @@ +[tool:pytest] +testpaths = test +python_files = test_*.py +python_functions = test_* +addopts = + -v + --tb=short + --strict-markers + [pytest] markers = + unit: Unit tests + integration: Integration tests + slow: Slow tests docker: marks tests that require a running Docker daemon and pull container images - diff --git a/src/microbots/environment/local_docker/LocalDockerEnvironment.py b/src/microbots/environment/local_docker/LocalDockerEnvironment.py index ff45a4d..65d2a25 100644 --- a/src/microbots/environment/local_docker/LocalDockerEnvironment.py +++ b/src/microbots/environment/local_docker/LocalDockerEnvironment.py @@ -130,10 +130,18 @@ def stop(self): except Exception as e: logger.error("❌ Failed to remove working directory: %s", e) + # Unused function. Keeping for reference or future use + def _escape(self, command: str) -> str: + # Escape double quotes and special characters for JSON safety + command = command.replace('"', '\\"') + command = command.replace("<", "<").replace(">", ">") + return command + def execute( self, command: str, timeout: Optional[int] = 300 ) -> CmdReturn: # TODO: Need proper return value logger.debug("➡️ Executing command in container: %s", command) + # command = self._escape(command) try: response = requests.post( f"http://localhost:{self.port}/", @@ -163,11 +171,11 @@ def execute( def copy_to_container(self, src_path: str, dest_path: str) -> bool: """ Copy a file or folder from the host machine to the Docker container. - + Args: src_path: Path to the source file/folder on the host machine dest_path: Destination path inside the container - + Returns: bool: True if copy was successful, False otherwise """ @@ -193,7 +201,7 @@ def copy_to_container(self, src_path: str, dest_path: str) -> bool: mkdir_result = self.execute(mkdir_cmd) if mkdir_result.return_code != 0: - logger.error("❌ Failed to create destination directory %s: %s", + logger.error("❌ Failed to create destination directory %s: %s", dest_dir, mkdir_result.stderr) return False else: @@ -212,7 +220,6 @@ def copy_to_container(self, src_path: str, dest_path: str) -> bool: # Execute the copy command result = subprocess.run( cmd, - shell=True, capture_output=True, text=True, timeout=300 @@ -235,11 +242,11 @@ def copy_to_container(self, src_path: str, dest_path: str) -> bool: def copy_from_container(self, src_path: str, dest_path: str) -> bool: """ Copy a file or folder from the Docker container to the host machine. - + Args: src_path: Path to the source file/folder inside the container dest_path: Destination path on the host machine - + Returns: bool: True if copy was successful, False otherwise """ @@ -271,7 +278,6 @@ def copy_from_container(self, src_path: str, dest_path: str) -> bool: # Execute the copy command result = subprocess.run( cmd, - shell=True, capture_output=True, text=True, timeout=300 diff --git a/src/microbots/environment/local_docker/image_builder/ShellCommunicator.py b/src/microbots/environment/local_docker/image_builder/ShellCommunicator.py index dbcc9e0..5894654 100644 --- a/src/microbots/environment/local_docker/image_builder/ShellCommunicator.py +++ b/src/microbots/environment/local_docker/image_builder/ShellCommunicator.py @@ -16,6 +16,7 @@ from dataclasses import dataclass logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', filename='/var/log/ShellCommunicator.log') @dataclass class CmdReturn: @@ -126,12 +127,20 @@ def _monitor_output(self, stream, output_queue: queue.Queue, stream_type: str): except Exception as e: output_queue.put((stream_type, f"Monitor error: {e}")) + # Unused function. Keeping for reference and future use def _re_escape(self, command: str) -> str: # Reverse .replace('"', '\\"') command = command.replace('\"', '"') - # command = command.replace("<", "<").replace(">", ">") + command = command.replace("<", "<").replace(">", ">") return command + # Unused function. Keeping for reference and future use + def _is_heredoc_command(self, command: str) -> bool: + """Check if command contains heredoc syntax.""" + import re + # Look for heredoc patterns like < CmdReturn: @@ -151,8 +160,8 @@ def send_command( return CmdReturn(stdout="", stderr="No active shell session", return_code=1) try: - command = self._re_escape(command) - + # command = self._re_escape(command) + if not wait_for_output: # Send the command without marker for async execution self.process.stdin.write(command + "\n") @@ -163,13 +172,14 @@ def send_command( # Generate a unique command completion marker marker = f"__COMMAND_COMPLETE_{int(time.time() * 1000000)}__" - # For bash only: Send command + marker in a single line to capture correct exit code - combined_command = f"{command}; echo '{marker}' $?" - - # Send the combined command - self.process.stdin.write(combined_command + "\n") + self.process.stdin.write(command + "\n") self.process.stdin.flush() + # Send exit code capture on a new line after user command completes + self.process.stdin.write(f"echo '{marker}' $?\n") + self.process.stdin.flush() + logger.debug("➡️ Sent command: %s", command) + logger.debug("🔖 Waiting for marker: %s", marker) # Collect output until marker is found or timeout output_lines = [] @@ -182,6 +192,7 @@ def send_command( try: # Check for output with a small timeout stream_type, line = self.output_queue.get(timeout=0.1) + logger.debug("⬅️ Received line from %s: %s", stream_type, line) # Check if this is our completion marker if marker in line: diff --git a/src/microbots/tools/tool_definitions/browser-use/browser.py b/src/microbots/tools/tool_definitions/browser-use/browser.py index e0ac562..c856311 100644 --- a/src/microbots/tools/tool_definitions/browser-use/browser.py +++ b/src/microbots/tools/tool_definitions/browser-use/browser.py @@ -28,7 +28,7 @@ async def main(args: list[str]) -> int: agent = Agent( task=what_to_browse, browser=browser, - llm=ChatAzureOpenAI(model="gpt-4.1"), + llm=ChatAzureOpenAI(model="gpt-5",temperature=1.0), # TODO: Gather it from environmental variable instead of hard coding. use_vision=False, ) history: AgentHistoryList = await agent.run() diff --git a/test/bot/browsing_bot_test.py b/test/bot/browsing_bot_test.py deleted file mode 100644 index ad445ea..0000000 --- a/test/bot/browsing_bot_test.py +++ /dev/null @@ -1,29 +0,0 @@ -import logging -import os -import sys - - -logger = logging.getLogger(__name__) -logging.basicConfig(level=logging.INFO) - -from dotenv import load_dotenv -load_dotenv() - -sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../src/"))) -from microbots.bot.BrowsingBot import BrowsingBot -from microbots.MicroBot import BotRunResult - -myBot = BrowsingBot( - model="azure-openai/mini-swe-agent-gpt5", -) - -response: BotRunResult = myBot.run( - "What is the capital of France?", - timeout_in_seconds=300, -) - -final_result = response.result -# logger.info(f"Response: {response}") -logger.debug("Status: %s\n, Error: %s\n\n\n, ***Result:***\n %s\n", response.status, response.error, response.result) - -print("Final Result: ", final_result) diff --git a/test/bot/calculator/log_analysis_test.py b/test/bot/calculator/log_analysis_test.py deleted file mode 100644 index 72e62b6..0000000 --- a/test/bot/calculator/log_analysis_test.py +++ /dev/null @@ -1,30 +0,0 @@ -import logging -import os -import sys -from pathlib import Path - -logger = logging.getLogger(__name__) -logging.basicConfig(level=logging.INFO) - -# Add src directory to path to import from local source -sys.path.insert( - 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src")) -) - -from microbots.bot.LogAnalysisBot import LogAnalysisBot -from microbots.constants import DOCKER_WORKING_DIR, LOG_FILE_DIR -from microbots.MicroBot import BotRunResult - -myBot = LogAnalysisBot( - model="azure-openai/mini-swe-agent-gpt5", - folder_to_mount=str(Path(__file__).parent / "code"), -) - -response: BotRunResult = myBot.run( - str(Path(__file__).parent / "calculator.log"), - timeout_in_seconds=300, -) - -print( - f"Status: {response.status}\n***Result:***\n{response.result}\n===\nError: {response.error}" -) diff --git a/test/bot/calculator/test_log_analysis_bot.py b/test/bot/calculator/test_log_analysis_bot.py new file mode 100644 index 0000000..1f5a0fa --- /dev/null +++ b/test/bot/calculator/test_log_analysis_bot.py @@ -0,0 +1,151 @@ +""" +Integration tests for LogAnalysisBot +""" +import pytest +import logging +import os +import sys +from pathlib import Path + +# Set up logging +logger = logging.getLogger(__name__) + +# Add src directory to path to import from local source +sys.path.insert( + 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src")) +) + +from microbots.bot.LogAnalysisBot import LogAnalysisBot +from microbots.constants import DOCKER_WORKING_DIR, LOG_FILE_DIR +from microbots.MicroBot import BotRunResult + + +@pytest.mark.integration +@pytest.mark.docker +@pytest.mark.slow +class TestLogAnalysisBotIntegration: + """Integration tests for LogAnalysisBot """ + + @pytest.fixture(scope="class") + def code_dir(self): + """Get the path to the calculator code directory""" + return Path(__file__).parent / "code" + + @pytest.fixture(scope="class") + def log_file(self): + """Get the path to the calculator log file""" + return Path(__file__).parent / "calculator.log" + + @pytest.fixture(scope="class") + def log_analysis_bot(self, code_dir): + """Create a LogAnalysisBot instance """ + # Ensure the code directory exists + if not code_dir.exists(): + pytest.skip(f"Code directory not found: {code_dir}") + + bot = LogAnalysisBot( + model="azure-openai/mini-swe-agent-gpt5", + folder_to_mount=str(code_dir), + ) + yield bot + + # Cleanup: stop the environment + if hasattr(bot, 'environment') and bot.environment: + try: + bot.environment.stop() + except Exception as e: + logger.warning(f"Error stopping environment: {e}") + + def test_analyze_calculator_log(self, log_analysis_bot, log_file): + """Test LogAnalysisBot analyzing calculator log file""" + # Ensure the log file exists + if not log_file.exists(): + pytest.skip(f"Log file not found: {log_file}") + + # Run the bot with the log analysis task + response: BotRunResult = log_analysis_bot.run( + str(log_file), + timeout_in_seconds=300, + ) + + # Log the response for debugging + logger.info(f"Status: {response.status}") + logger.info(f"Result: {response.result}") + if response.error: + logger.error(f"Error: {response.error}") + + # Assertions + assert response.status == True, f"Bot run failed with error: {response.error}" + assert response.result is not None, "Bot result should not be None" + assert response.error is None, f"Bot should not have errors! ERROR: {response.error}" + + # Check that the result contains analysis of the log + result_lower = response.result.lower() + assert len(response.result.strip()) > 0, "Result should not be empty" + + def test_analyze_nonexistent_log(self, log_analysis_bot): + """Test LogAnalysisBot behavior when trying to analyze a non-existent log file""" + nonexistent_log = Path(__file__).parent / "nonexistent.log" + + # LogAnalysisBot should raise a ValueError when trying to copy a nonexistent file + # This is expected behavior at the infrastructure level + with pytest.raises(ValueError, match="Failed to copy file to container"): + response: BotRunResult = log_analysis_bot.run( + str(nonexistent_log), + timeout_in_seconds=60, + ) + + logger.info(f"Successfully caught expected ValueError for nonexistent log file") + + + +# Manual test runner function (can be called directly) +def run_log_analysis_bot_manual_test(): + """Manual test function that can be run outside pytest""" + print("=== Manual LogAnalysisBot Integration Test ===") + + code_dir = Path(__file__).parent / "code" + log_file = Path(__file__).parent / "calculator.log" + + if not code_dir.exists(): + print(f"ERROR: Code directory not found: {code_dir}") + return + + if not log_file.exists(): + print(f"ERROR: Log file not found: {log_file}") + return + + print(f"Using code from: {code_dir}") + print(f"Using log file: {log_file}") + + try: + # Create LogAnalysisBot instance + myBot = LogAnalysisBot( + model="azure-openai/mini-swe-agent-gpt5", + folder_to_mount=str(code_dir), + ) + + # Run the log analysis task + response: BotRunResult = myBot.run( + str(log_file), + timeout_in_seconds=300, + ) + + # Print results + print(f"Status: {response.status}") + print(f"***Result:***\n{response.result}") + print(f"===\nError: {response.error}") + + # Cleanup + if hasattr(myBot, 'environment') and myBot.environment: + myBot.environment.stop() + + except Exception as e: + print(f"ERROR: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + # Allow running this file directly for manual testing + run_log_analysis_bot_manual_test() diff --git a/test/bot/calculator/test_writing_bot.py b/test/bot/calculator/test_writing_bot.py new file mode 100644 index 0000000..558c924 --- /dev/null +++ b/test/bot/calculator/test_writing_bot.py @@ -0,0 +1,175 @@ +""" +Integration tests for WritingBot +""" +import pytest +import logging +import os +import sys +from pathlib import Path + +# Set up logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + handlers=[ + logging.StreamHandler(), + logging.FileHandler('writing_bot_test.log') + ] +) +logger = logging.getLogger(__name__) + +# Add src directory to path to import from local source +sys.path.insert( + 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src")) +) + +from microbots.bot.WritingBot import WritingBot +from microbots.constants import DOCKER_WORKING_DIR +from microbots.MicroBot import BotRunResult + + +@pytest.mark.integration +@pytest.mark.docker +@pytest.mark.slow +class TestWritingBotIntegration: + """Integration tests for WritingBot with real environment and API""" + + @pytest.fixture(scope="class") + def calculator_data_dir(self): + """Get the path to the calculator test data directory""" + return Path(__file__).parent # The test file is already in the calculator directory + + @pytest.fixture(scope="class") + def writing_bot(self, calculator_data_dir): + """Create a WritingBot instance with real environment""" + # Ensure the calculator data directory exists + if not calculator_data_dir.exists(): + pytest.skip(f"Calculator data directory not found: {calculator_data_dir}") + + # Check if calculator.log exists + calc_log = calculator_data_dir / "calculator.log" + if not calc_log.exists(): + pytest.skip(f"Calculator log file not found: {calc_log}") + + bot = WritingBot( + model="azure-openai/mini-swe-agent-gpt5", + folder_to_mount=str(calculator_data_dir), + ) + yield bot + + # Cleanup: stop the environment + if hasattr(bot, 'environment') and bot.environment: + try: + bot.environment.stop() + except Exception as e: + logger.warning(f"Error stopping environment: {e}") + + def test_write_calculator_fix(self, writing_bot, calculator_data_dir): + """Test WritingBot reading calculator.log and fixing the calculator.py file""" + # Take a snapshot of git status before running the bot + import subprocess + git_before = subprocess.run(['git', 'status', '--porcelain'], + capture_output=True, text=True, + cwd=calculator_data_dir.parent.parent.parent) + + # Run the bot with the fix task + response: BotRunResult = writing_bot.run( + """Inside the mounted directory there is a calculator.log which have execution of code/calculator.py. + Read the log and fix the error.""", + timeout_in_seconds=300, + ) + + # Log the response for debugging (safely handle potential JSON serialization issues) + logger.info(f"Status: {response.status}") + + # Check if the calculator.py file was modified + calc_file = calculator_data_dir / "code" / "calculator.py" + calc_content = calc_file.read_text() + + # Check that the fix was applied (should contain some form of zero check) + assert "b == 0" in calc_content or "b != 0" in calc_content or "if not b" in calc_content or "if b == 0" in calc_content, \ + "Calculator code should contain a check for division by zero" + + logger.info(f"Fixed calculator.py content preview: {calc_content[:500]}...") + + + +# Manual test runner function (can be called directly) +def run_writing_bot_manual_test(): + """Manual test function that can be run outside pytest""" + print("=== Manual WritingBot Integration Test ===") + + calculator_data_dir = Path(__file__).parent.parent / "calculator" + + if not calculator_data_dir.exists(): + print(f"ERROR: Calculator data directory not found: {calculator_data_dir}") + return + + calc_log = calculator_data_dir / "calculator.log" + if not calc_log.exists(): + print(f"ERROR: Calculator log file not found: {calc_log}") + return + + print(f"Using calculator data from: {calculator_data_dir}") + + try: + # Create WritingBot instance + myBot = WritingBot( + model="azure-openai/mini-swe-agent-gpt5", + folder_to_mount=str(calculator_data_dir), + ) + + # Run the fix task + response: BotRunResult = myBot.run( + f"Read the /{DOCKER_WORKING_DIR}/calculator/calculator.log file and identify the error. " + f"Then examine /{DOCKER_WORKING_DIR}/calculator/code/calculator.py and fix the divide function " + f"to properly handle division by zero by adding a check before division and returning an appropriate message or value.", + timeout_in_seconds=300, + ) + + # Print results (safely handle potential serialization issues) + try: + print(f"Status: {response.status}") + print(f"***Result:***\n{response.result}") + print(f"===\nError: {response.error}") + except Exception as e: + print(f"Could not print response due to serialization issue: {e}") + try: + status_str = str(response.status) + result_str = str(response.result) if response.result is not None else "None" + error_str = str(response.error) if response.error is not None else "None" + print(f"Status: {status_str}") + print(f"***Result:***\n{result_str}") + print(f"===\nError: {error_str}") + except Exception as e2: + print(f"Status: {response.status}") + print(f"Result type: {type(response.result)}") + print(f"Error type: {type(response.error)}") + print(f"Serialization error: {e2}") + + # Check if file was modified + calc_file = calculator_data_dir / "code" / "calculator.py" + if calc_file.exists(): + print(f"✅ calculator.py exists!") + calc_content = calc_file.read_text() + if "b == 0" in calc_content or "b != 0" in calc_content or "if not b" in calc_content: + print(f"✅ Zero division check found in code!") + else: + print(f"⚠️ No obvious zero division check found") + print(f"Updated content preview: {calc_content[:500]}...") + else: + print("❌ calculator.py was not found") + + # Cleanup + if hasattr(myBot, 'environment') and myBot.environment: + myBot.environment.stop() + + except Exception as e: + print(f"ERROR: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + # Allow running this file directly for manual testing + run_writing_bot_manual_test() diff --git a/test/bot/countries_to_capital/reading_bot_test.py b/test/bot/countries_to_capital/reading_bot_test.py deleted file mode 100644 index 1801c7b..0000000 --- a/test/bot/countries_to_capital/reading_bot_test.py +++ /dev/null @@ -1,29 +0,0 @@ -import logging -import os -import sys -from pathlib import Path - -logger = logging.getLogger(__name__) -logging.basicConfig(level=logging.INFO) - -# Add src directory to path to import from local source -sys.path.insert( - 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src")) -) -from microbots.bot.ReadingBot import ReadingBot -from microbots.constants import DOCKER_WORKING_DIR -from microbots.MicroBot import BotRunResult - -myBot = ReadingBot( - model="azure-openai/mini-swe-agent-gpt5", - folder_to_mount=str(Path(__file__).parent / "countries_dir"), -) - -response: BotRunResult = myBot.run( - f"Read the /{DOCKER_WORKING_DIR}/countries_dir/countries.txt give me the capitals of each country.", - timeout_in_seconds=300, -) - -print( - f"Status: {response.status}\n***Result:***\n{response.result}\n===\nError: {response.error}" -) diff --git a/test/bot/countries_to_capital/test_reading_bot.py b/test/bot/countries_to_capital/test_reading_bot.py new file mode 100644 index 0000000..8dd498d --- /dev/null +++ b/test/bot/countries_to_capital/test_reading_bot.py @@ -0,0 +1,173 @@ +""" +Integration tests for ReadingBot +""" +import pytest +import logging +import os +import sys +from pathlib import Path + +# Set up logging +logger = logging.getLogger(__name__) + +# Add src directory to path to import from local source +sys.path.insert( + 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src")) +) + +from microbots.bot.ReadingBot import ReadingBot +from microbots.constants import DOCKER_WORKING_DIR +from microbots.MicroBot import BotRunResult + + +@pytest.mark.integration +@pytest.mark.docker +@pytest.mark.slow +class TestReadingBotIntegration: + """Integration tests for ReadingBot with real environment and API""" + + @pytest.fixture(scope="class") + def countries_data_dir(self): + """Get the path to the countries test data directory""" + return Path(__file__).parent / "countries_dir" + + @pytest.fixture(scope="class") + def reading_bot(self, countries_data_dir): + """Create a ReadingBot instance with real environment""" + # Ensure the countries data directory exists + if not countries_data_dir.exists(): + pytest.skip(f"Countries data directory not found: {countries_data_dir}") + + bot = ReadingBot( + model="azure-openai/mini-swe-agent-gpt5", + folder_to_mount=str(countries_data_dir), + ) + yield bot + + # Cleanup: stop the environment + if hasattr(bot, 'environment') and bot.environment: + try: + bot.environment.stop() + except Exception as e: + logger.warning(f"Error stopping environment: {e}") + + def test_read_countries_file(self, reading_bot): + """Test ReadingBot reading countries.txt file and extracting capitals""" + # Run the bot with the reading task + response: BotRunResult = reading_bot.run( + f"Read the /{DOCKER_WORKING_DIR}/countries_dir/countries.txt file and give me the capitals of each country.", + timeout_in_seconds=300, + ) + + # Log the response for debugging + logger.info(f"Status: {response.status}") + logger.info(f"Result: {response.result}") + if response.error: + logger.error(f"Error: {response.error}") + + # Assertions + assert response.status == True, f"Bot run failed with error: {response.error}" + assert response.result is not None, "Bot result should not be None" + assert response.error is None, f"Bot should not have errors: {response.error}" + + # Check that the result contains actual capitals from your countries + result_lower = response.result.lower() + expected_capitals = ["delhi", "washington", "brasília", "berlin", "singapore"] + assert any(capital in result_lower for capital in expected_capitals), \ + f"Result should contain capitals of the countries in the file. Got: {response.result}" + + def test_read_nonexistent_file(self, reading_bot): + """Test ReadingBot behavior when trying to read a non-existent file""" + response: BotRunResult = reading_bot.run( + f"Read the /{DOCKER_WORKING_DIR}/countries_dir/nonexistent.txt file.", + timeout_in_seconds=60, + ) + + # Log the response for debugging + logger.info(f"Status: {response.status}") + logger.info(f"Result: {response.result}") + if response.error: + logger.info(f"Error: {response.error}") + + # The bot should handle this gracefully - either return an error status + # or mention in the result that the file doesn't exist + assert response.status in [True, False], "Bot should handle missing files gracefully" + + if response.status == True: + # If successful, the result should mention the file doesn't exist + result_lower = response.result.lower() + assert any(phrase in result_lower for phrase in ["not found", "does not exist", "no such file"]), \ + f"Result should indicate file doesn't exist. Got: {response.result}" + + + @pytest.mark.parametrize("task", [ + "Count the number of countries in the countries.txt file", + "Tell me which country has Paris as its capital", + "What is the capital of Germany according to the file?", + ]) + def test_specific_reading_tasks(self, reading_bot, task): + """Test ReadingBot with specific reading tasks""" + full_task = f"Read /{DOCKER_WORKING_DIR}/countries_dir/countries.txt and {task.lower()}" + + response: BotRunResult = reading_bot.run( + full_task, + timeout_in_seconds=120, + ) + + # Log the response for debugging + logger.info(f"Task: {task}") + logger.info(f"Status: {response.status}") + logger.info(f"Result: {response.result}") + if response.error: + logger.error(f"Error: {response.error}") + + # Basic assertions + assert response.status == True, f"Task '{task}' failed with error: {response.error}" + assert response.result is not None, f"Task '{task}' result should not be None" + assert len(response.result.strip()) > 0, f"Task '{task}' should return non-empty result" + + +# Manual test runner function (can be called directly) +def run_reading_bot_manual_test(): + """Manual test function that can be run outside pytest""" + print("=== Manual ReadingBot Integration Test ===") + + countries_data_dir = Path(__file__).parent / "countries_dir" + + if not countries_data_dir.exists(): + print(f"ERROR: Countries data directory not found: {countries_data_dir}") + return + + print(f"Using countries data from: {countries_data_dir}") + + try: + # Create ReadingBot instance + myBot = ReadingBot( + model="azure-openai/mini-swe-agent-gpt5", + folder_to_mount=str(countries_data_dir), + ) + + # Run the reading task + response: BotRunResult = myBot.run( + f"Read the /{DOCKER_WORKING_DIR}/countries_dir/countries.txt file and give me the capitals of each country.", + timeout_in_seconds=300, + ) + + # Print results + print(f"Status: {response.status}") + print(f"***Result:***\n{response.result}") + print(f"===\nError: {response.error}") + + # Cleanup + if hasattr(myBot, 'environment') and myBot.environment: + myBot.environment.stop() + + except Exception as e: + print(f"ERROR: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + # Allow running this file directly for manual testing + run_reading_bot_manual_test() \ No newline at end of file diff --git a/test/bot/countries_to_capital/writing_bot_test.py b/test/bot/countries_to_capital/writing_bot_test.py deleted file mode 100644 index bcab3ab..0000000 --- a/test/bot/countries_to_capital/writing_bot_test.py +++ /dev/null @@ -1,27 +0,0 @@ -import logging -import os -import sys -from pathlib import Path - -logger = logging.getLogger(__name__) -logging.basicConfig(level=logging.INFO) - -# Add src directory to path to import from local source -sys.path.insert( - 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src")) -) -from microbots.bot.WritingBot import WritingBot -from microbots.constants import DOCKER_WORKING_DIR -from microbots.MicroBot import BotRunResult - -myBot = WritingBot( - model="azure-openai/mini-swe-agent-gpt5", - folder_to_mount=str(Path(__file__).parent / "countries_dir"), -) - -response: BotRunResult = myBot.run( - f"Read the /{DOCKER_WORKING_DIR}/countries_dir/countries.txt store their capitals in /{DOCKER_WORKING_DIR}/countries_dir/capitals.txt file", - timeout_in_seconds=300, -) - -print(f"Status: {response.status}, Result: {response.result}, Error: {response.error}") diff --git a/test/bot/test_browsing_bot.py b/test/bot/test_browsing_bot.py new file mode 100644 index 0000000..84a1c65 --- /dev/null +++ b/test/bot/test_browsing_bot.py @@ -0,0 +1,105 @@ +import logging +import os +import sys +import pytest + +# Setup logging for tests +logger = logging.getLogger(__name__) + +# Load environment variables +from dotenv import load_dotenv +load_dotenv() + +# Add src to path for imports +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../../src/"))) +from microbots.bot.BrowsingBot import BrowsingBot +from microbots.MicroBot import BotRunResult + + +@pytest.mark.integration +@pytest.mark.docker +@pytest.mark.slow +class TestBrowsingBot: + """Integration tests for BrowsingBot functionality.""" + + @pytest.fixture(scope="class") + def browsing_bot(self): + """Create a BrowsingBot instance for testing.""" + bot = BrowsingBot(model="azure-openai/mini-swe-agent-gpt5") + yield bot + # Cleanup: stop the environment + if hasattr(bot, 'environment') and bot.environment: + try: + bot.environment.stop() + except Exception as e: + logger.warning(f"Error stopping environment: {e}") + + def test_simple_question_response(self, browsing_bot): + """Test that the bot can answer a simple factual question.""" + response: BotRunResult = browsing_bot.run( + "What is the capital of France?", + timeout_in_seconds=300, + ) + + # Assert the response was successful + assert response.status == True, f"Bot failed with error: {response.error}" + assert response.result is not None, "Bot returned no result" + assert isinstance(response.result, str), "Result should be a string" + + # Check that the result contains the expected answer + result_lower = response.result.lower() + assert "paris" in result_lower, f"Expected 'Paris' in result, got: {response.result}" + + logger.info(f"Test passed. Bot response: {response.result}") + + + @pytest.mark.parametrize("query,expected_keywords", [ + ("What is the capital of Germany?", ["berlin"]), + ("What is 2+2?", ["4", "four"]), + ("Who is the current President of the United States?", ["Trump"]), + ]) + def test_multiple_queries(self, browsing_bot, query, expected_keywords): + """Test the bot with multiple different queries.""" + response: BotRunResult = browsing_bot.run(query, timeout_in_seconds=300) + + assert response.status == True, f"Query '{query}' failed: {response.error}" + assert response.result is not None, f"No result for query: {query}" + + result_lower = response.result.lower() + # At least one expected keyword should be in the result + keyword_found = any(keyword.lower() in result_lower for keyword in expected_keywords) + assert keyword_found, f"None of {expected_keywords} found in result: {response.result}" + + logger.info(f"Query '{query}' passed with result: {response.result[:100]}...") + +# Manual test runner function (can be called directly) +def run_browsing_bot_manual_test(): + """Manual test function that can be run outside pytest""" + print("=== Manual BrowsingBot Integration Test ===") + + try: + # Create BrowsingBot instance + myBot = BrowsingBot( + model="azure-openai/mini-swe-agent-gpt5", + ) + + response: BotRunResult = myBot.run( + "Find the current weather in New York City.", + timeout_in_seconds=300, + ) + + print(f"Status: {response.status}") + print(f"***Result:***\n{response.result}") + print(f"===\nError: {response.error}") + + print("\n=== Manual Test Completed ===") + + except Exception as e: + print(f"ERROR: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + # Allow running the test file directly for manual testing or pytest + run_browsing_bot_manual_test() \ No newline at end of file diff --git a/test/environment/local_docker/LocalDockerEnvironmentTest.py b/test/environment/local_docker/LocalDockerEnvironmentTest.py deleted file mode 100644 index 9e9db37..0000000 --- a/test/environment/local_docker/LocalDockerEnvironmentTest.py +++ /dev/null @@ -1,83 +0,0 @@ -"""Simple manual test script for LocalDockerEnvironment. - -This file demonstrates: - 1. Creating multiple containerized shell environments with different mount permissions. - 2. Executing commands through the HTTP bridge exposed by the FastAPI shell server. - 3. Expected behavior differences between READ_WRITE and READ_ONLY mounts. - -NOTES: - - Each environment must use a distinct host port to avoid binding conflicts. - - The container internally listens on port 8080; host ports map to that internal port. - - Commands like `cd` affect only the shell session state inside the container serving that env. - - A READ_ONLY mount should prevent file creation (e.g. "touch should_fail.txt") inside the mounted path. - - This script does not automatically stop containers so you can inspect them afterward. - Remember to call `env.stop()` (or prune with Docker) when done to free resources. - - Error handling here is minimal; in production wrap execute calls and inspect return values. - -USAGE: - python -m test.environment.local_docker.LocalDockerEnvironmentTest - -Clean Up Manually (example): - docker ps | grep - # then stop/remove as needed -""" - -from microbots.environment.local_docker import LocalDockerEnvironment - - -def LocalDockerEnvironmentTest(): - # Environment 1: Read-write mount of /home/kkaitepalli/MAP on host port 8085 - # Provide absolute path to a directory on your host machine - env1 = LocalDockerEnvironment( - port=8085, - folder_to_mount="/home/kkaitepalli/MAP", - permission="READ_WRITE", - ) - - # Environment 2: No mount (isolated filesystem view) on host port 8086 - env2 = LocalDockerEnvironment(port=8086) - - # Environment 3: Read-only mount of /home/kkaitepalli/telescope on host port 8087 - env3 = LocalDockerEnvironment( - port=8087, - folder_to_mount="/home/kkaitepalli/telescope", - permission="READ_ONLY", - ) - - try: - # Navigate inside env1 into the mounted directory - response = env1.execute("cd MAP") - print("env1 cd MAP:", response) - - # List contents to verify mount - response = env1.execute("ls -la") - print("env1 ls -la:\n", response) - - # Create a file in env2's working directory (no mount involved) - env2.execute("touch testfile.txt") - - # Change to subdirectory in env1 (assuming it exists in the mounted content) - env1.execute("cd mariner-aks-pipelines") - - # Attempt navigation inside env3's read-only mount - env3.execute("cd telescope") - - # Attempt to create a file in read-only environment (should fail) - response = env3.execute("touch should_fail.txt") - print("env3 touch should_fail.txt (expected failure or error msg):", response) - - # Show current working directory in env1 - response = env1.execute("pwd") - print("env1 pwd:", response) - - finally: - print( - "Containers left running for inspection. Call envX.stop() or \n" - "docker stop && docker rm when finished." - ) - # Example cleanup (uncomment as needed): - # env1.stop(); env2.stop(); env3.stop() - - -if __name__ == "__main__": # Allows running via: python -m test.environment.local_docker.LocalDockerEnvironmentTest - LocalDockerEnvironmentTest() diff --git a/test/environment/local_docker/TestFileCopy.py b/test/environment/local_docker/TestFileCopy.py deleted file mode 100644 index 883147c..0000000 --- a/test/environment/local_docker/TestFileCopy.py +++ /dev/null @@ -1,59 +0,0 @@ -#!/usr/bin/env python3 -""" -Simple test for file copy functionality -""" - -import os -import sys -from pathlib import Path -import logging -logger = logging.getLogger(__name__) -logging.basicConfig(level=logging.INFO) -# Add src directory to path to import from local source -sys.path.insert( - 0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src")) -) - -from microbots.environment.local_docker import LocalDockerEnvironment - -class TestFileCopy(): - """Simple test for file copy""" - - def test_copy_file(self): - """Test copying a file to container and from container to host""" - # Create environment - env = LocalDockerEnvironment(port=8081) - - try: - # Copy to container - # Get path to countries.txt file specifically - countries_file_path = Path(__file__).parent.parent.parent / "bot" / "countries_to_capital" / "countries_dir" / "countries.txt" - result = env.copy_to_container(str(countries_file_path), "/var/log/") - - # Verify - print(f"Copy result: {result}") - if result: - print("✅ Copy succeeded") - else: - print("❌ Copy failed") - - # Test copying from container to host - # Use /tmp/ which is available and writable on all systems - result_back = env.copy_from_container("/var/log/countries.txt", "/tmp/") - print(f"Copy back result: {result_back}") - if result_back: - print("✅ Copy back succeeded") - else: - print("❌ Copy back failed") - - finally: - # Cleanup - # os.unlink(test_file) - # env.stop() - print("Not stopping environment for debug") - - -if __name__ == "__main__": - # Run the tests - test_instance = TestFileCopy() - test_instance.test_copy_file() \ No newline at end of file diff --git a/test/environment/local_docker/test_local_docker_environment.py b/test/environment/local_docker/test_local_docker_environment.py new file mode 100644 index 0000000..c8d787b --- /dev/null +++ b/test/environment/local_docker/test_local_docker_environment.py @@ -0,0 +1,393 @@ +""" +Integration tests for LocalDockerEnvironment +""" +import pytest +import shutil +import os +import time +import socket +from pathlib import Path +import re + +# Add src to path for imports +import sys +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../src"))) + +from microbots.environment.local_docker.LocalDockerEnvironment import LocalDockerEnvironment +from microbots.environment.Environment import CmdReturn + +# Use the correct working directory path +DOCKER_WORKING_DIR = "workdir" + +class TestLocalDockerEnvironmentIntegration: + """Integration tests for LocalDockerEnvironment with real Docker containers""" + + @pytest.fixture(scope="class") + def available_port(self): + """Find an available port for testing - class scoped to reuse same port""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(('', 0)) + s.listen(1) + port = s.getsockname()[1] + return port + + @pytest.fixture + def test_dir(self): + """Use existing test/bot/calculator directory for testing instead of temp directory""" + # Get the base path of the minions project dynamically + current_file_dir = os.path.dirname(os.path.abspath(__file__)) # /path/to/minions/test/environment/local_docker + minions_base_path = os.path.dirname(os.path.dirname(os.path.dirname(current_file_dir))) # /path/to/minions + test_directory = os.path.join(minions_base_path, "test", "bot", "calculator") + + # Verify the directory exists and has test files + assert os.path.exists(test_directory), f"Test directory {test_directory} does not exist" + assert os.path.exists(os.path.join(test_directory, "calculator.log")), "Test file calculator.log not found" + assert os.path.exists(os.path.join(test_directory, "code")), "Test code subdirectory not found" + assert os.path.exists(os.path.join(test_directory, "code", "calculator.py")), "Test file calculator.py not found" + + return test_directory + + @pytest.fixture(scope="class") + def shared_env(self, available_port): + """Create a single LocalDockerEnvironment instance for all tests in this class""" + env = None + try: + env = LocalDockerEnvironment(port=available_port) + + # Wait for container to be ready + import time + time.sleep(2) + + # Verify it's working + result = env.execute("echo 'Environment ready'") + assert result.return_code == 0 + + yield env + finally: + if env: + env.stop() + + @pytest.mark.integration + @pytest.mark.docker + def test_basic_environment_lifecycle(self, shared_env): + """Test basic environment functionality using shared environment""" + # Test that container is running + assert shared_env.container is not None + shared_env.container.reload() + assert shared_env.container.status == 'running' + + # Test that we can connect and execute commands + result = shared_env.execute("echo 'Hello World'") + assert result.return_code == 0 + assert "Hello World" in result.stdout + + + @pytest.mark.integration + @pytest.mark.docker + def test_initialization_validation_errors(self): + """Test that initialization properly validates parameters""" + # Test permission without folder + with pytest.raises(ValueError, match="permission provided but folder_to_mount is None"): + LocalDockerEnvironment(port=8999, permission="READ_ONLY") + + # Test folder without permission + with pytest.raises(ValueError, match="folder_to_mount provided but permission is None"): + LocalDockerEnvironment(port=8999, folder_to_mount="/some/path") + + # Test invalid permission + with pytest.raises(ValueError, match="permission must be 'READ_ONLY' or 'READ_WRITE' when provided"): + LocalDockerEnvironment(port=8999, folder_to_mount="/some/path", permission="INVALID") + + @pytest.mark.integration + @pytest.mark.docker + def test_command_execution_basic(self, shared_env): + """Test basic command execution functionality using shared environment""" + # Test simple echo + result = shared_env.execute("echo 'test message'") + assert result.return_code == 0 + assert "test message" in result.stdout + assert result.stderr == "" + + # Test command with error + result = shared_env.execute("nonexistent_command") + assert result.return_code != 0 + assert result.stderr != "" + + # Test pwd + result = shared_env.execute("pwd") + assert result.return_code == 0 + assert "/app" in result.stdout + + @pytest.mark.integration + @pytest.mark.docker + def test_command_execution_complex(self, shared_env): + """Test that heredoc commands are automatically converted to safe alternatives""" + import time + + # Test the specific heredoc command that was causing timeouts + heredoc_command = """cat > /tmp/test_heredoc.py << EOF +#!/usr/bin/env python3 +import sys + +def missing_colon_error(): + # This function demonstrates a syntax error - missing colon after if statement + if True + print("This will cause a syntax error") + return True + + return False + +if __name__ == "__main__": + try: + result = missing_colon_error() + print(f"Function result: {result}") + except SyntaxError as e: + print(f"Syntax error caught: {e}") + sys.exit(1) +EOF""" + + print("Testing heredoc command execution...") + start_time = time.time() + + # Execute the heredoc command + result = shared_env.execute(heredoc_command, timeout=60) + end_time = time.time() + execution_time = end_time - start_time + + print(f"Heredoc command completed in {execution_time:.2f} seconds") + print(f"Return code: {result.return_code}") + print(f"Stdout: {result.stdout}") + print(f"Stderr: {result.stderr}") + + # The command should complete successfully (converted automatically) + assert result.return_code == 0, f"Heredoc command failed with return code {result.return_code}" + + # Should complete in reasonable time (less than 30 seconds) + assert execution_time < 30, f"Heredoc command took too long: {execution_time:.2f} seconds" + + # Verify the file was created correctly + verify_result = shared_env.execute("cat /tmp/test_heredoc.py") + assert verify_result.return_code == 0 + assert "missing_colon_error" in verify_result.stdout + assert re.search(r"if True$", verify_result.stdout, re.MULTILINE) is not None # Check for "if True" at end of line (missing colon) + print(verify_result) + + # Test that the Python file has the expected syntax error + python_result = shared_env.execute("python3 /tmp/test_heredoc.py") + # Should fail due to syntax error (missing colon) + assert python_result.return_code != 0 + assert "SyntaxError" in python_result.stderr or "invalid syntax" in python_result.stderr + + print("Heredoc command with automatic conversion test passed successfully") + + @pytest.mark.integration + @pytest.mark.docker + def test_read_write_mount(self, test_dir): + """Test READ_WRITE mount functionality - creates own env because mounting requires initialization-time config""" + # Get a fresh port for this test since shared_env is using the class-scoped port + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(('', 0)) + s.listen(1) + mount_port = s.getsockname()[1] + + env = None + try: + env = LocalDockerEnvironment( + port=mount_port, + folder_to_mount=test_dir, + permission="READ_WRITE" + ) + + folder_name = os.path.basename(test_dir) + mount_path = f"/{DOCKER_WORKING_DIR}/{folder_name}" + + # Test that mounted directory is accessible + result = env.execute(f"ls {mount_path}") + assert result.return_code == 0 + assert "calculator.log" in result.stdout + assert "code" in result.stdout + + # Test reading the mounted file + result = env.execute(f"cat {mount_path}/calculator.log") + assert result.return_code == 0 + assert "Calculator application started" in result.stdout + + # Test reading subdirectory + result = env.execute(f"ls {mount_path}/code") + assert result.return_code == 0 + assert "calculator.py" in result.stdout + + # Test writing to the mounted directory (should succeed with READ_WRITE) + result = env.execute(f"echo 'new content from container' > {mount_path}/new_test_file.txt") + assert result.return_code == 0 + + # Verify the file was created on the host + new_file_path = os.path.join(test_dir, "new_test_file.txt") + assert os.path.exists(new_file_path) + with open(new_file_path, 'r') as f: + content = f.read().strip() + assert "new content from container" in content + + # Clean up the created file + os.remove(new_file_path) + + finally: + if env: + env.stop() + + @pytest.mark.integration + @pytest.mark.docker + def test_read_only_mount(self, test_dir): + """Test READ_ONLY mount with overlay functionality - creates own env because mounting requires initialization-time config""" + # Get a fresh port for this test since shared_env is using the class-scoped port + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(('', 0)) + s.listen(1) + mount_port = s.getsockname()[1] + + env = None + try: + env = LocalDockerEnvironment( + port=mount_port, + folder_to_mount=test_dir, + permission="READ_ONLY" + ) + + folder_name = os.path.basename(test_dir) + mount_path = f"/{DOCKER_WORKING_DIR}/{folder_name}" + + # Test that mounted directory is accessible + result = env.execute(f"ls {mount_path}") + assert result.return_code == 0 + assert "calculator.log" in result.stdout + + # Test reading the mounted file + result = env.execute(f"cat {mount_path}/calculator.log") + assert result.return_code == 0 + assert "Calculator application started" in result.stdout + + # Test writing to the mounted directory (should appear to succeed with overlay) + result = env.execute(f"echo 'overlay content' > {mount_path}/overlay_file.txt") + assert result.return_code == 0 + + # Verify the file appears to exist in container + result = env.execute(f"cat {mount_path}/overlay_file.txt") + assert result.return_code == 0 + assert "overlay content" in result.stdout + + # Verify the file was NOT created on the host (read-only mount) + overlay_file_path = os.path.join(test_dir, "overlay_file.txt") + assert not os.path.exists(overlay_file_path) + + # Test modifying existing file (should work in overlay) + result = env.execute(f"echo 'overlay modification' >> {mount_path}/calculator.log") + assert result.return_code == 0 + + # Verify original file on host is unchanged + with open(os.path.join(test_dir, "calculator.log"), 'r') as f: + content = f.read() + assert "overlay modification" not in content + assert "Calculator application started" in content + + finally: + if env: + env.stop() + + @pytest.mark.integration + @pytest.mark.docker + def test_copy_to_container(self, shared_env, test_dir): + """Test copying files from host to container using shared environment""" + # Test copying a single file + source_file = os.path.join(test_dir, "calculator.log") + dest_dir = "/var/log/" + + success = shared_env.copy_to_container(source_file, dest_dir) + assert success is True + + # Verify file exists and has correct content in container + result = shared_env.execute(f"cat {dest_dir}calculator.log") + assert result.return_code == 0 + assert "Calculator application started" in result.stdout + + # Test copying non-existent file + success = shared_env.copy_to_container("/nonexistent/file.txt", "/tmp/fail.txt") + assert success is False + + @pytest.mark.integration + @pytest.mark.docker + def test_copy_from_container(self, shared_env): + """Test copying files from container to host using shared environment""" + # Create a file in container + container_file = "/tmp/container_created.txt" + result = shared_env.execute(f"echo 'Created in container' > {container_file}") + assert result.return_code == 0 + + # Copy file from container to host directory + host_dest_dir = "/tmp/" + success = shared_env.copy_from_container(container_file, host_dest_dir) + assert success is True + + # The file should be copied to /tmp/container_created.txt + copied_file_path = "/tmp/container_created.txt" + + # Verify file exists on host with correct content + assert os.path.exists(copied_file_path) + with open(copied_file_path, 'r') as f: + content = f.read() + assert "Created in container" in content + + # Clean up the created file + os.remove(copied_file_path) + + # Test copying non-existent file + success = shared_env.copy_from_container("/nonexistent/file.txt", "/tmp/fail.txt") + assert success is False + +# Manual test runner function +def run_local_docker_environment_manual_test(): + """Manual test function that can be run outside pytest""" + print("=== Manual LocalDockerEnvironment Integration Test ===") + + # Get available port + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(('', 0)) + s.listen(1) + available_port = s.getsockname()[1] + + try: + print(f"Creating LocalDockerEnvironment on port {available_port}") + env = LocalDockerEnvironment(port=available_port) + + print("--- Test 1: Basic Command Execution ---") + result = env.execute("echo 'Hello from Docker!'") + print(f"Return Code: {result.return_code}") + print(f"Stdout: {result.stdout}") + print(f"Stderr: {result.stderr}") + + print("\n--- Test 2: Working Directory ---") + result = env.execute("pwd") + print(f"Current directory: {result.stdout.strip()}") + + print("\n--- Test 3: File Operations ---") + result = env.execute("echo 'test content' > /tmp/test.txt && cat /tmp/test.txt") + print(f"File operation result: {result.stdout.strip()}") + + print("\n--- Test 4: Container Info ---") + result = env.execute("hostname && whoami") + print(f"Container info: {result.stdout.strip()}") + + print("\n--- Manual Test Completed Successfully ---") + + except Exception as e: + print(f"ERROR: {e}") + import traceback + traceback.print_exc() + finally: + if 'env' in locals(): + env.stop() + print("Environment stopped and cleaned up") + + +if __name__ == "__main__": + # Allow running the test file directly for manual testing + run_local_docker_environment_manual_test() \ No newline at end of file diff --git a/test/environment/swe-rex/LocalDockerTest.py b/test/environment/swe-rex/test_local_docker.py.disabled similarity index 100% rename from test/environment/swe-rex/LocalDockerTest.py rename to test/environment/swe-rex/test_local_docker.py.disabled