mirror of
https://github.com/A6-9V/MQL5-Google-Onedrive.git
synced 2026-04-11 11:20:57 +00:00
227 lines
7.7 KiB
Python
Executable file
227 lines
7.7 KiB
Python
Executable file
#!/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())
|