<commit_before>import sys
import traceback
from StringIO import StringIO
from types import FileType
import os,shutil
from test import fixture
from migrate.versioning.repository import Repository
from migrate.versioning import genmodel, shell
from sqlalchemy import MetaData,Table

python_version = sys.version[0:3]

class Shell(fixture.Shell):
    _cmd=os.path.join('python migrate', 'versioning', 'shell.py')
    @classmethod
    def cmd(cls,*p):
        p = map(lambda s: str(s),p)
        ret = ' '.join([cls._cmd]+p)
        return ret
    def execute(self,shell_cmd,runshell=None):
        """A crude simulation of a shell command, to speed things up"""
        # If we get an fd, the command is already done
        if isinstance(shell_cmd, FileType) or isinstance(shell_cmd, StringIO):
            return shell_cmd
        # Analyze the command; see if we can 'fake' the shell
        try:
            # Forced to run in shell?
            #if runshell or '--runshell' in sys.argv:
            if runshell:
                raise Exception
            # Remove the command prefix
            if not shell_cmd.startswith(self._cmd):
                raise Exception
            cmd = shell_cmd[(len(self._cmd)+1):]
            params = cmd.split(' ')
            command = params[0]
        except: 
            return super(Shell,self).execute(shell_cmd)

        # Redirect stdout to an object; redirect stderr to stdout
        fd = StringIO()
        orig_stdout = sys.stdout
        orig_stderr = sys.stderr
        sys.stdout = fd
        sys.stderr = fd
        # Execute this command
        try:
            try:
                shell.main(params)
            except SystemExit,e:
                # Simulate the exit status
                fd_close=fd.close
                def close_():
                    fd_close()
                    return e.args[0]
                fd.close = close_
            except Exception,e:
                # Print the exception, but don't re-raise it
                traceback.print_exc()
                # Simulate a nonzero exit status
                fd_close=fd.close
                def close_():
                    fd_close()
                    return 2
                fd.close = close_
        finally:
            # Clean up
            sys.stdout = orig_stdout
            sys.stderr = orig_stderr
            fd.seek(0)
        return fd

    def cmd_version(self,repos_path):
        fd = self.execute(self.cmd('version',repos_path))
        ret = int(fd.read().strip())
        self.assertSuccess(fd)
        return ret
    def cmd_db_version(self,url,repos_path):
        fd = self.execute(self.cmd('db_version',url,repos_path))
        txt = fd.read()
        #print txt
        ret = int(txt.strip())
        self.assertSuccess(fd)
        return ret

class TestShellCommands(Shell):
    """Tests migrate.py commands"""

    def test_run(self):
        """Runs; displays help"""
        # Force this to run in shell...
        self.assertSuccess(self.cmd('-h'),runshell=True)
        self.assertSuccess(self.cmd('--help'),runshell=True)

    def test_help(self):
        """Display help on a specific command"""
        self.assertSuccess(self.cmd('-h'),runshell=True)
        self.assertSuccess(self.cmd('--help'),runshell=True)
        for cmd in shell.api.__all__:
            fd=self.execute(self.cmd('help',cmd))
            # Description may change, so best we can do is ensure it shows up
            #self.assertNotEquals(fd.read(),'')
            output = fd.read()
            self.assertNotEquals(output,'')
            self.assertSuccess(fd)
    
    def test_create(self):
        """Repositories are created successfully"""
        repos=self.tmp_repos()
        name='name'
        # Creating a file that doesn't exist should succeed
        cmd=self.cmd('create',repos,name)
        self.assertSuccess(cmd)
        # Files should actually be created
        self.assert_(os.path.exists(repos))
        # The default table should not be None
        repos_ = Repository(repos)
        self.assertNotEquals(repos_.config.get('db_settings','version_table'),'None')
        # Can't create it again: it already exists
        self.assertFailure(cmd)
    
    def test_script(self):
        """We can create a migration script via the command line"""
        repos=self.tmp_repos()
        self.assertSuccess(self.cmd('create',repos,'repository_name'))
        self.assertSuccess(self.cmd('script', '--repository=%s' % repos, 'Desc'))
        self.assert_(os.path.exists('%s/versions/001_Desc.py' % repos))
        # 's' instead of 'script' should work too
        self.assertSuccess(self.cmd('script', '--repository=%s' % repos, 'More'))
        self.assert_(os.path.exists('%s/versions/002_More.py' % repos))

    def test_script_sql(self):
        """We can create a migration sql script via the command line"""
        repos=self.tmp_repos()
        self.assertSuccess(self.cmd('create',repos,'repository_name'))
        self.assertSuccess(self.cmd('script_sql', '--repository=%s' % repos, 'mydb'))
        self.assert_(os.path.exists('%s/versions/001_mydb_upgrade.sql' % repos))
        self.assert_(os.path.exists('%s/versions/001_mydb_downgrade.sql' % repos))
        
        # Can't create it again: it already exists
        self.assertFailure(self.cmd('script_sql', '--repository=%s' % repos, 'mydb'))

    def test_manage(self):
        """Create a project management script"""
        script=self.tmp_py()
        self.assert_(not os.path.exists(script))
        # No attempt is made to verify correctness of the repository path here
        self.assertSuccess(self.cmd('manage',script,'--repository=/path/to/repository'))
        self.assert_(os.path.exists(script))

class TestShellRepository(Shell):
    """Shell commands on an existing repository/python script"""
    def setUp(self):
        """Create repository, python change script"""<commit_msg>integrate test case fix by <NAME><commit_after>import sys
import traceback
from StringIO import StringIO
from types import FileType
import os,shutil
from test import fixture
from migrate.versioning.repository import Repository
from migrate.versioning import genmodel, shell
from sqlalchemy import MetaData,Table

python_version = sys.version[0:3]

class Shell(fixture.Shell):
    _cmd=os.path.join('python migrate', 'versioning', 'shell.py')
    @classmethod
    def cmd(cls,*p):
        p = map(lambda s: str(s),p)
        ret = ' '.join([cls._cmd]+p)
        return ret
    def execute(self,shell_cmd,runshell=None):
        """A crude simulation of a shell command, to speed things up"""
        # If we get an fd, the command is already done
        if isinstance(shell_cmd, FileType) or isinstance(shell_cmd, StringIO):
            return shell_cmd
        # Analyze the command; see if we can 'fake' the shell
        try:
            # Forced to run in shell?
            #if runshell or '--runshell' in sys.argv:
            if runshell:
                raise Exception
            # Remove the command prefix
            if not shell_cmd.startswith(self._cmd):
                raise Exception
            cmd = shell_cmd[(len(self._cmd)+1):]
            params = cmd.split(' ')
            command = params[0]
        except: 
            return super(Shell,self).execute(shell_cmd)

        # Redirect stdout to an object; redirect stderr to stdout
        fd = StringIO()
        orig_stdout = sys.stdout
        orig_stderr = sys.stderr
        sys.stdout = fd
        sys.stderr = fd
        # Execute this command
        try:
            try:
                shell.main(params)
            except SystemExit,e:
                # Simulate the exit status
                fd_close=fd.close
                def close_():
                    fd_close()
                    return e.args[0]
                fd.close = close_
            except Exception,e:
                # Print the exception, but don't re-raise it
                traceback.print_exc()
                # Simulate a nonzero exit status
                fd_close=fd.close
                def close_():
                    fd_close()
                    return 2
                fd.close = close_
        finally:
            # Clean up
            sys.stdout = orig_stdout
            sys.stderr = orig_stderr
            fd.seek(0)
        return fd

    def cmd_version(self,repos_path):
        fd = self.execute(self.cmd('version',repos_path))
        ret = int(fd.read().strip())
        self.assertSuccess(fd)
        return ret
    def cmd_db_version(self,url,repos_path):
        fd = self.execute(self.cmd('db_version',url,repos_path))
        txt = fd.read()
        #print txt
        ret = int(txt.strip())
        self.assertSuccess(fd)
        return ret

class TestShellCommands(Shell):
    """Tests migrate.py commands"""

    def test_run(self):
        """Runs; displays help"""
        # Force this to run in shell...
        self.assertSuccess(self.cmd('-h'),runshell=True)
        self.assertSuccess(self.cmd('--help'),runshell=True)

    def test_help(self):
        """Display help on a specific command"""
        self.assertSuccess(self.cmd('-h'),runshell=True)
        self.assertSuccess(self.cmd('--help'),runshell=True)
        for cmd in shell.api.__all__:
            fd=self.execute(self.cmd('help',cmd))
            # Description may change, so best we can do is ensure it shows up
            #self.assertNotEquals(fd.read(),'')
            output = fd.read()
            self.assertNotEquals(output,'')
            self.assertSuccess(fd)
    
    def test_create(self):
        """Repositories are created successfully"""
        repos=self.tmp_repos()
        name='name'
        # Creating a file that doesn't exist should succeed
        cmd=self.cmd('create',repos,name)
        self.assertSuccess(cmd)
        # Files should actually be created
        self.assert_(os.path.exists(repos))
        # The default table should not be None
        repos_ = Repository(repos)
        self.assertNotEquals(repos_.config.get('db_settings','version_table'),'None')
        # Can't create it again: it already exists
        self.assertFailure(cmd)
    
    def test_script(self):
        """We can create a migration script via the command line"""
        repos=self.tmp_repos()
        self.assertSuccess(self.cmd('create',repos,'repository_name'))
        self.assertSuccess(self.cmd('script', '--repository=%s' % repos, 'Desc'))
        self.assert_(os.path.exists('%s/versions/001_Desc.py' % repos))
        # 's' instead of 'script' should work too
        self.assertSuccess(self.cmd('script', '--repository=%s' % repos, 'More'))
        self.assert_(os.path.exists('%s/versions/002_More.py' % repos))

    def test_script_sql(self):
        """We can create a migration sql script via the command line"""
        repos=self.tmp_repos()
        self.assertSuccess(self.cmd('create',repos,'repository_name'))
        self.assertSuccess(self.cmd('script_sql', '--repository=%s' % repos, 'mydb'))
        self.assert_(os.path.exists('%s/versions/001_mydb_upgrade.sql' % repos))
        self.assert_(os.path.exists('%s/versions/001_mydb_downgrade.sql' % repos))

        # Test creating a second
        self.assertSuccess(self.cmd('script_sql', '--repository=%s' % repos, 'mydb'))
        self.assert_(os.path.exists('%s/versions/002_mydb_upgrade.sql' % repos))
        self.assert_(os.path.exists('%s/versions/002_mydb_downgrade.sql' % repos))

    def test_manage(self):
        """Create a project management script"""
        script=self.tmp_py()
        self.assert_(not os.path.exists(script))
        # No attempt is made to verify correctness of the repository path here
        self.assertSuccess(self.cmd('manage',script,'--repository=/path/to/repository'))
        self.assert_(os.path.exists(script))

class TestShellRepository(Shell):
    """Shell commands on an existing repository/python script"""
    def setUp(self):
        """Create repository, python change script"""
{'fraction_non_alphanumeric': 0.11480312344243229, 'fraction_numerical': 0.0036550922080079745, 'mean_word_length': 2.231946308724832, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3687371', 'n_tokens_mistral': 3479, 'n_tokens_neox': 3335, 'n_words': 843}