Source

heechee / heechee / tests / base.py

Full commit
"""
Base classes for the test suite.
"""

import os
import unittest
import tempfile
import subprocess
import shutil
import time
import socket
import signal
try:
    from cStringIO import StringIO
except ImportError:
    from StringIO import StringIO

class HeecheeTest(unittest.TestCase):
    
    use_hydra = False
    use_cache = False
    
    def setUp(self):
        "Sets up a target directory and a source repo."
        self.source_dir = tempfile.mkdtemp(suffix="-heechee-src")
        self.target_dir = tempfile.mkdtemp(suffix="-heechee-tgt")
        
        if self.use_hydra:
            path_pattern = os.path.join(self.source_dir, "%s")
            self.source_dir = os.path.join(self.source_dir, "subdir")
            os.mkdir(self.source_dir)
            args = ["python", "-m", "heechee.webdav.hydra", "^/([^/]+)", path_pattern]
        else:
            args = ["python", "-m", "heechee.webdav.__init__", self.source_dir]
        
        if self.use_cache:
            args.append("--cache")
        
        self.init_repo()
        self.server = subprocess.Popen(
            args,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        
        # Wait for server to come up
        for i in range(20):
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            try:
                s.connect(('localhost', 8080))
            except socket.error:
                time.sleep(0.25)
            else:
                return
        raise RuntimeError("Cannot seem to launch Heechee")
    
    def tearDown(self):
        "Deletes the target directory"
        shutil.rmtree(self.source_dir)
        shutil.rmtree(self.target_dir)
        # Kill server, and wait till it dies
        os.kill(self.server.pid, signal.SIGKILL)
#        self.server.kill()
#        self.server.wait()
    
    def init_repo(self):
        "Should be overridden to initialise the repository."
        raise NotImplementedError()
    
    def run_cmds(self, cmds, cwd=None):
        if isinstance(cmds, basestring):
            cmds = [cmds]
        output = ""
        for cmd in cmds:
            process = subprocess.Popen(cmd, shell=True, cwd=cwd, stdout=subprocess.PIPE)
            stdout, stderr = process.communicate()
            output += stdout
        return output
    
    def source_cmds(self, cmds):
        "Runs a command[s] in the source (mercurial/git) directory."
        return self.run_cmds(cmds, self.source_dir)
    
    def target_cmds(self, cmds):
        "Runs a command[s] in the target (svn) directory."
        return self.run_cmds(cmds, self.target_dir)
    
    def assert_exists(self, path):
        "Asserts that a file exists in the target."
        assert os.path.exists(os.path.join(self.target_dir, path)), "%s does not exist" % path
    
    def assert_exists_source(self, path):
        "Asserts that a file exists in the source."
        assert os.path.exists(os.path.join(self.source_dir, path)), "%s does not exist" % path
    
    def assert_not_exists(self, path):
        "Asserts that a file does not exist in the target."
        assert not os.path.exists(os.path.join(self.target_dir, path)), "%s exists" % path
    
    def assert_not_exists_source(self, path):
        "Asserts that a file exists in the source."
        assert not os.path.exists(os.path.join(self.source_dir, path)), "%s exists" % path
    
    def assert_contents(self, path, supposed_contents):
        "Asserts that a the contents of the given file are exactly as provided."
        self.assert_exists(path)
        fh = open(os.path.join(self.target_dir, path))
        contents = fh.read()
        fh.close()
        self.assertEqual(contents, supposed_contents)
    
    def assert_contents_source(self, path, supposed_contents):
        "Asserts that a the contents of the given file are exactly as provided."
        self.assert_exists_source(path)
        fh = open(os.path.join(self.source_dir, path))
        contents = fh.read()
        fh.close()
        self.assertEqual(contents, supposed_contents)
    
    def assert_contains(self, haystack, needle):
        "Asserts that needle is somewhere in the string haystack."
        assert needle in haystack, "String %r not found in %r" % (needle, haystack)
    
    def assert_not_contains(self, haystack, needle):
        "Asserts that needle is not somewhere in the string haystack."
        assert needle not in haystack, "String %r found in %r" % (needle, haystack)
    
    def assert_equal(self, item1, item2):
        "assertEqual doesn't follow PEP8, damnit."
        return self.assertEqual(item1, item2)