#!/usr/bin/env python3 """ Integration tests for Repository Sync Manager Tests the configuration validation and sync functionality """ import unittest import os import sys from unittest.mock import patch, MagicMock from pathlib import Path # Add scripts directory to path sys.path.insert(0, os.path.dirname(__file__)) try: from repo_sync_manager import RepoSyncManager except ImportError: print("⚠️ Could not import repo_sync_manager. Skipping tests.") sys.exit(0) class TestRepoSyncManager(unittest.TestCase): """Test cases for RepoSyncManager""" def setUp(self): """Set up test environment""" # Clear environment variables self.env_backup = {} env_vars = ['REPO_SYNC_TOKEN', 'L6_N9_REPO', 'L6_N9_SYNC_BRANCH', 'REST_API_KEY', 'REST_API_URL'] for var in env_vars: if var in os.environ: self.env_backup[var] = os.environ[var] del os.environ[var] def tearDown(self): """Restore environment""" # Restore backed up environment variables for var, value in self.env_backup.items(): os.environ[var] = value def test_init_defaults(self): """Test initialization with default values""" manager = RepoSyncManager() self.assertEqual(manager.source_repo, "A6-9V/MQL5-Google-Onedrive") self.assertEqual(manager.target_repo, "A6-9V/L6-N9") self.assertEqual(manager.sync_branch, "main") self.assertEqual(manager.rest_api_key, "") self.assertEqual(manager.rest_api_url, "") self.assertEqual(manager.sync_token, "") def test_init_custom_values(self): """Test initialization with custom environment values""" os.environ['L6_N9_REPO'] = "custom/repo" os.environ['L6_N9_SYNC_BRANCH'] = "develop" os.environ['REST_API_KEY'] = "test_key" os.environ['REST_API_URL'] = "https://test.example.com" os.environ['REPO_SYNC_TOKEN'] = "test_token" manager = RepoSyncManager() self.assertEqual(manager.target_repo, "custom/repo") self.assertEqual(manager.sync_branch, "develop") self.assertEqual(manager.rest_api_key, "test_key") self.assertEqual(manager.rest_api_url, "https://test.example.com") self.assertEqual(manager.sync_token, "test_token") def test_validate_config_no_token(self): """Test validation fails when no token is configured""" manager = RepoSyncManager() valid, msg = manager.validate_config() self.assertFalse(valid) self.assertIn("REPO_SYNC_TOKEN", msg) def test_validate_config_with_token(self): """Test validation succeeds with token""" os.environ['REPO_SYNC_TOKEN'] = "test_token" manager = RepoSyncManager() valid, msg = manager.validate_config() self.assertTrue(valid) self.assertEqual(msg, "Configuration valid") def test_validate_config_api_url_no_key(self): """Test validation warns when API URL set without key""" os.environ['REPO_SYNC_TOKEN'] = "test_token" os.environ['REST_API_URL'] = "https://test.example.com" manager = RepoSyncManager() valid, msg = manager.validate_config() self.assertTrue(valid) self.assertIn("Warning", msg) self.assertIn("REST_API_KEY", msg) @patch('subprocess.check_output') def test_get_git_info(self, mock_subprocess): """Test getting git information""" mock_subprocess.side_effect = [ "abc123def456\n", # commit hash "main\n" # branch name ] manager = RepoSyncManager() git_info = manager.get_git_info() self.assertEqual(git_info['commit'], "abc123def456") self.assertEqual(git_info['branch'], "main") self.assertIn('timestamp', git_info) @patch('subprocess.check_output') def test_get_git_info_error(self, mock_subprocess): """Test git info handles errors gracefully""" from subprocess import CalledProcessError mock_subprocess.side_effect = CalledProcessError(1, 'git') manager = RepoSyncManager() git_info = manager.get_git_info() self.assertEqual(git_info, {}) @patch('requests.post') def test_notify_rest_api_success(self, mock_post): """Test successful REST API notification""" os.environ['REST_API_KEY'] = "test_key" os.environ['REST_API_URL'] = "https://test.example.com" mock_response = MagicMock() mock_response.status_code = 200 mock_post.return_value = mock_response manager = RepoSyncManager() result = manager.notify_rest_api({"test": "data"}) self.assertTrue(result) mock_post.assert_called_once() # Check call arguments call_args = mock_post.call_args self.assertEqual(call_args[0][0], "https://test.example.com") self.assertIn('Authorization', call_args[1]['headers']) self.assertEqual(call_args[1]['headers']['Authorization'], "Bearer test_key") @patch('requests.post') def test_notify_rest_api_failure(self, mock_post): """Test REST API notification handles failures""" os.environ['REST_API_KEY'] = "test_key" os.environ['REST_API_URL'] = "https://test.example.com" mock_response = MagicMock() mock_response.status_code = 500 mock_post.return_value = mock_response manager = RepoSyncManager() result = manager.notify_rest_api({"test": "data"}) self.assertFalse(result) @patch('requests.post') def test_notify_rest_api_exception(self, mock_post): """Test REST API notification handles exceptions""" import requests os.environ['REST_API_KEY'] = "test_key" os.environ['REST_API_URL'] = "https://test.example.com" mock_post.side_effect = requests.RequestException("Connection error") manager = RepoSyncManager() result = manager.notify_rest_api({"test": "data"}) self.assertFalse(result) def test_notify_rest_api_not_configured(self): """Test REST API notification skips when not configured""" manager = RepoSyncManager() result = manager.notify_rest_api({"test": "data"}) self.assertTrue(result) # Should return True when skipped class TestRepoSyncManagerCLI(unittest.TestCase): """Test command-line interface""" def test_check_config_cli(self): """Test --check-config CLI argument""" # Clear any existing token old_token = os.environ.get('REPO_SYNC_TOKEN') if 'REPO_SYNC_TOKEN' in os.environ: del os.environ['REPO_SYNC_TOKEN'] try: manager = RepoSyncManager() valid, msg = manager.validate_config() # Should fail without token self.assertFalse(valid) finally: # Restore token if it existed if old_token: os.environ['REPO_SYNC_TOKEN'] = old_token def run_tests(): """Run all tests""" print("🧪 Running Repository Sync Manager Tests") print("=" * 50) # Discover and run tests loader = unittest.TestLoader() suite = loader.loadTestsFromModule(sys.modules[__name__]) runner = unittest.TextTestRunner(verbosity=2) result = runner.run(suite) print("=" * 50) if result.wasSuccessful(): print("✅ All tests passed!") return 0 else: print("❌ Some tests failed") return 1 if __name__ == '__main__': sys.exit(run_tests())