cvs-proxy/cvs_proxy/test_cvs_client.py

import unittest
from unittest.mock import patch, MagicMock, mock_open
from cvs_proxy.cvs_client import CVSClient
import subprocess
import os
import tempfile
import shutil

class TestCVSClient(unittest.TestCase):
    def setUp(self):
        # Use a test repository URL directly
        self.test_url = ':pserver:testuser@test.repo.com:/path/to/repo'
        
        # Create a temporary directory to simulate checkouts
        self.test_checkouts_dir = tempfile.mkdtemp()
        
        # Patch the checkout method to prevent actual checkout
        CVSClient._checkout_repository = lambda self: None
        
        # Create the CVS client with explicit repos_checkout and cvs_module
        self.cvs_client = CVSClient(self.test_url, repos_checkout=self.test_checkouts_dir, cvs_module='test_module')
        
        # Override local_repo_path with a test directory
        self.test_repo_dir = os.path.join(self.test_checkouts_dir, 'testuser_test_repo_com_path_to_repo')
        os.makedirs(self.test_repo_dir, exist_ok=True)
        self.cvs_client.local_repo_path = self.test_repo_dir

    def tearDown(self):
        # Clean up the temporary directory
        shutil.rmtree(self.test_checkouts_dir)
        
        # Remove the environment variable
        if 'REPO_CHECKOUTS' in os.environ:
            del os.environ['REPO_CHECKOUTS']

    def _create_test_files(self):
        """
        Create a test directory structure for repository tree testing
        """
        # Create test files and directories
        os.makedirs(os.path.join(self.test_repo_dir, 'src'), exist_ok=True)
        os.makedirs(os.path.join(self.test_repo_dir, 'docs'), exist_ok=True)
        
        # Create some files
        with open(os.path.join(self.test_repo_dir, 'README.md'), 'w') as f:
            f.write('Test README')
        
        with open(os.path.join(self.test_repo_dir, 'src', 'main.py'), 'w') as f:
            f.write('def main():\n    pass')
        
        with open(os.path.join(self.test_repo_dir, 'docs', 'manual.txt'), 'w') as f:
            f.write('Documentation')

    def test_initialization(self):
        """Test CVS client initialization"""
        # Verify repository URL
        self.assertEqual(self.cvs_client.repo_url, ':pserver:testuser@test.repo.com:/path/to/repo')
        
        # Verify checkout directory is created in the specified REPO_CHECKOUTS location
        self.assertTrue(os.path.exists(self.test_checkouts_dir))
        self.assertTrue(os.path.exists(self.test_repo_dir))

    def test_missing_url(self):
        """Test initialization without URL"""
        with self.assertRaises(ValueError):
            CVSClient()

    def test_list_repository_tree(self):
        """Test listing repository tree structure"""
        # Create test files
        self._create_test_files()

        # Test full repository tree
        tree = self.cvs_client.list_repository_tree()
        expected_tree = [
            'README.md',
            'docs/manual.txt', 
            'src/main.py'
        ]
        self.assertEqual(sorted(tree), sorted(expected_tree))

        # Test subdirectory listing
        src_tree = self.cvs_client.list_repository_tree('src')
        self.assertEqual(src_tree, ['main.py'])

        # Test non-existent directory
        empty_tree = self.cvs_client.list_repository_tree('nonexistent')
        self.assertEqual(empty_tree, [])

    def test_checkout_directory_configuration(self):
        """Test that checkout directory can be configured via environment variable"""
        # Create another test directory
        another_checkouts_dir = tempfile.mkdtemp()
        os.environ['REPO_CHECKOUTS'] = another_checkouts_dir
        
        # Create a new CVS client
        another_client = CVSClient(':pserver:another@example.com:/another/repo')
        
        # Verify the checkout is in the new directory
        expected_repo_path = os.path.join(another_checkouts_dir, 'another_example_com_another_repo')
        self.assertEqual(another_client.local_repo_path, expected_repo_path)
        
        # Clean up
        shutil.rmtree(another_checkouts_dir)
        del os.environ['REPO_CHECKOUTS']

    @patch('cvs_proxy.cvs_client.subprocess.run')
    def test_run_cvs_command(self, mock_run):
        """Test the internal _run_cvs_command method"""
        mock_run.return_value = MagicMock(
            stdout='Command output',
            stderr='',
            returncode=0
        )

        result = self.cvs_client._run_cvs_command(['rlog', 'test.txt'])
        self.assertEqual(result, 'Command output')
        
        # Verify the command was constructed correctly
        mock_run.assert_called_once()
        call_args = mock_run.call_args[0][0]
        self.assertEqual(call_args, [
            'cvs', 
            '-d', 
            ':pserver:testuser@test.repo.com:/path/to/repo', 
            'rlog', 
            'test.txt'
        ])

    @patch('cvs_proxy.cvs_client.subprocess.run')
    def test_get_file_diff(self, mock_run):
        """Test getting file differences"""
        mock_run.return_value = MagicMock(
            stdout='--- a/file.txt\n+++ b/file.txt\n@@ -1,3 +1,4 @@\n line1\n-old line\n+new line\n line3',
            stderr='',
            returncode=0
        )

        diff = self.cvs_client.get_file_diff('file.txt', '1.1', '1.2')
        self.assertIn('line1', diff)
        self.assertIn('-old line', diff)
        self.assertIn('+new line', diff)

    @patch('cvs_proxy.cvs_client.subprocess.run')
    def test_get_file_history(self, mock_run):
        """Test retrieving file history using cvs log"""
        mock_run.return_value = MagicMock(
            stdout='''
revision 1.2
date: 2023/11/20 10:00:00;  author: testuser;  state: Exp;  lines: +5 -2
revision 1.1
date: 2023/11/19 15:30:00;  author: testuser;  state: Exp;  lines: +10 -0
''',
            stderr='',
            returncode=0
        )

        history = self.cvs_client.get_file_history('file.txt')
        self.assertIsInstance(history, list)
        self.assertEqual(len(history), 2)
        self.assertIn('revision', history[0])
        self.assertEqual(history[0]['revision'], '1.2')
        self.assertEqual(history[0]['author'], 'testuser')
        self.assertEqual(history[0]['date'], '2023/11/20 10:00:00')
        self.assertEqual(history[0]['state'], 'Exp')
        
        # Verify the cvs log command was called correctly
        mock_run.assert_called()
        call_args = mock_run.call_args[0][0]
        self.assertEqual(call_args[0], 'cvs')
        self.assertIn('log', call_args)
        self.assertIn('file.txt', call_args)

    @patch('cvs_proxy.cvs_client.subprocess.run')
    def test_get_file_content(self, mock_run):
        """Test retrieving file content"""
        mock_run.return_value = MagicMock(
            stdout='File contents\nMultiple lines\nof text',
            stderr='',
            returncode=0
        )

        content = self.cvs_client.get_file_content('file.txt', '1.2')
        self.assertEqual(content, 'File contents\nMultiple lines\nof text')

    @patch('cvs_proxy.cvs_client.subprocess.run')
    def test_command_failure(self, mock_run):
        """Test handling of CVS command failure"""
        mock_run.side_effect = subprocess.CalledProcessError(
            returncode=1, 
            cmd=['cvs', 'rlog'], 
            stderr='Error: Repository not found'
        )

        with self.assertRaises(subprocess.CalledProcessError):
            self.cvs_client._run_cvs_command(['rlog'])

if __name__ == '__main__':
    unittest.main()