Commits

Andrew Godwin  committed 1385f19

Add in proper Hydra support, and tests for same. It can checkout, but some tests still fail (log, commit, etc.)

  • Participants
  • Parent commits aa65016

Comments (0)

Files changed (6)

File heechee/repo/__init__.py

                 type = t
                 break
     if not type:
-        raise NoRepoError('Repository type is unknown')
+        raise NoRepoError('Repository type is unknown (dir %r)' % path)
     if type not in repotypes:
-        raise NoRepoError('Repository type is not supported')
+        raise NoRepoError('Repository type is not supported (dir %r)' % path)
     # Import the module of the right type
     module = __import__("heechee.repo.%s" % type, {}, {}, ['*'])
     # Return the appropriate class

File heechee/tests/__init__.py

 
 import unittest
 
-from heechee.tests.mercurial import MercurialTests
+from heechee.tests.mercurial import MercurialTests, MercurialHydraTests
 from heechee.tests.svndiff import SvnDiffTests
 
 if __name__ == '__main__':

File heechee/tests/base.py

 
 class HeecheeTest(unittest.TestCase):
     
+    use_hydra = False
+    
     def setUp(self):
         "Sets up a target directory and a source repo."
         self.source_dir = tempfile.mkdtemp(suffix="-heechee-src")
-        self.init_repo()
         self.target_dir = tempfile.mkdtemp(suffix="-heechee-tgt")
-        self.server = subprocess.Popen(
-            ["python", "-m", "heechee.webdav.__init__", self.source_dir],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        )
+        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)
+            self.init_repo()
+            self.server = subprocess.Popen(
+                ["python", "-m", "heechee.webdav.hydra", "^/([^/]+)/", path_pattern],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+            )
+        else:
+            self.init_repo()
+            self.server = subprocess.Popen(
+                ["python", "-m", "heechee.webdav.__init__", self.source_dir],
+                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)

File heechee/tests/mercurial.py

 
 class MercurialTests(HeecheeTest):
     
+    url_prefix = ""
+    
     def init_repo(self):
         self.source_cmds([
             "hg init",
     def test_checkout_all(self):
         # Checkout the whole repo
         self.target_cmds([
-            "svn co http://localhost:8080/ checkout"
+            "svn co http://localhost:8080/%s checkout" % self.url_prefix
         ])
         # Check things are there
         self.assert_contents("checkout/trunk/a", "a\n")
     def test_checkout_trunk(self):
         # Checkout a subdir
         self.target_cmds([
-            "svn co http://localhost:8080/trunk/ checkout"
+            "svn co http://localhost:8080/%strunk/ checkout" % self.url_prefix
         ])
         # Make sure it only has file a
         self.assert_contents("checkout/a", "a\n")
     def test_checkout_tag(self):
         # Checkout the tag
         self.target_cmds([
-            "svn co http://localhost:8080/tags/tag1/ checkout"
+            "svn co http://localhost:8080/%stags/tag1/ checkout" % self.url_prefix
         ])
         # Make sure the files are there
         self.assert_contents("checkout/a", "a\n")
     
     def test_checkout_subdir(self):
         self.target_cmds([
-            "svn co http://localhost:8080/branches/branch1/dir/ checkout"
+            "svn co http://localhost:8080/%sbranches/branch1/dir/ checkout" % self.url_prefix
         ])
         self.assert_contents("checkout/c", "c\n")
         self.assert_not_exists("checkout/a")
     
     def test_ls(self):
         self.assert_equal(
-            self.target_cmds("svn ls http://localhost:8080/"),
+            self.target_cmds("svn ls http://localhost:8080/%s" % self.url_prefix),
             "branches/\ntags/\ntrunk/\n",
         )
         self.assert_equal(
-            self.target_cmds("svn ls http://localhost:8080/trunk/"),
+            self.target_cmds("svn ls http://localhost:8080/%strunk/" % self.url_prefix),
             "a\n",
         )
 
     def test_cat(self):
         self.assert_equal(
-            self.target_cmds("svn cat http://localhost:8080/trunk/a"),
+            self.target_cmds("svn cat http://localhost:8080/%strunk/a" % self.url_prefix),
             "a\n",
         )    
     
     def test_update(self):
         # First, get a checkout.
-        self.target_cmds("svn co http://localhost:8080/trunk/ checkout")
+        self.target_cmds("svn co http://localhost:8080/%strunk/ checkout" % self.url_prefix)
         # Make sure there's no file d (or b)
         self.assert_not_exists("checkout/b")
         self.assert_not_exists("checkout/d")
     
     def test_log(self):
         # Get the log directly
-        log = self.target_cmds("svn log http://localhost:8080/")
+        log = self.target_cmds("svn log http://localhost:8080/%s" % self.url_prefix)
         # Check it contains certain vital things
         self.assert_contains(log, "r1 |")
         self.assert_contains(log, "r2 |")
         self.assert_contains(log, "\naddc\n")
         self.assert_contains(log, "\nAdded tag tag1")
         # Get the log for trunk (only had one commit)
-        log = self.target_cmds("svn log http://localhost:8080/trunk/")
+        log = self.target_cmds("svn log http://localhost:8080/%strunk/" % self.url_prefix)
         self.assert_contains(log, "r1 |")
         self.assert_not_contains(log, "r2 |")
         # Check out the code, get the log from there (well, server still...)
-        self.target_cmds("svn co http://localhost:8080/ checkout")
+        self.target_cmds("svn co http://localhost:8080/%s checkout" % self.url_prefix)
         log = self.target_cmds("svn log checkout")
         # Check it contains certain vital things
         self.assert_contains(log, "r1 |")
         "Tests committing to the fake SVN repo"
         # Check out a copy of trunk, create a new file, add it, and commit it
         self.target_cmds([
-            "svn co http://localhost:8080/trunk/ checkout",
+            "svn co http://localhost:8080/%strunk/ checkout" % self.url_prefix,
             "echo e > checkout/e",
             "echo aa > checkout/a",
             "svn add checkout/e",
         "Tests committing directories. Mercurial should ignore these."
         # Check out a copy of trunk, create a new file, add it, and commit it
         self.target_cmds([
-            "svn co http://localhost:8080/trunk/ checkout",
+            "svn co http://localhost:8080/%strunk/ checkout" % self.url_prefix,
             "svn mkdir checkout/newdir",
             "svn ci checkout -m 'Commit from SVN'",
         ])
         self.assert_not_exists_source("newdir")
 
 
+class MercurialHydraTests(MercurialTests):
+    
+    use_hydra = True
+    url_prefix = "subdir/"
+
+
 if __name__ == '__main__':
     unittest.main()

File heechee/webdav/hydra.py

 
 import re
-from werkzeug.wrappers import Request
+import logging
+
+from werkzeug.wrappers import Request, Response
+from werkzeug.exceptions import NotFound
+from werkzeug.utils import DispatcherMiddleware
+
+from heechee.webdav import RepositoryServer, get_repository
 
 class RepositoryHydra(object):
     
         url_pattern = r'
         """
         
-        self.url_pattern = url_pattern
+        self.url_pattern = re.compile(url_pattern)
         self.path_pattern = path_pattern
 
-    def __call__(self):
-        raise NotImplementedError
+    def __call__(self, environ, start_response):
+        # Work out the repository
+        script = environ.get('PATH_INFO', '')
+        match = self.url_pattern.match(script)
+        try:
+            repo_path = self.path_pattern % match.groups()
+        except AttributeError:
+            return NotFound("Path not valid.")
+        # Make a repository app for that
+        app = RepositoryServer(get_repository(repo_path))
+        # Delegate
+        original_script_name = environ.get('SCRIPT_NAME', '')
+        environ['SCRIPT_NAME'] = original_script_name + script[:match.end()]
+        environ['PATH_INFO'] = script[match.end():]
+        return app(environ, start_response)
+
+
+# As-script behaviour
+if __name__ == "__main__":
+    import optparse
+    parser = optparse.OptionParser("%prog [options] REPOPATH")
+    parser.add_option("-p", "--port", type="int", default=8080,
+        help="Port on which the server should listen. Defaults to 8080.")
+    parser.add_option("--debug", action='store_true', default=False,
+        help="Display debug output (warning: there's lots)")
+    options, args = parser.parse_args()
+    if len(args) < 2:
+        raise Exception('Not enough arguments; need a url pattern and a path pattern')
+   
+    # Set up logging.
+    logging.basicConfig(
+        format="%(asctime)s - %(levelname)8s - %(message)s",
+        level=options.debug and logging.DEBUG or logging.INFO,
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    from werkzeug import run_simple
+    run_simple(
+        'localhost',
+        options.port,
+        RepositoryHydra(args[0], args[1]), #, "/home/andrew/Programs/%s/"),
+    )

File heechee/webdav/report.py

             # Find what part of the path they want
             path_parts = urlsplit(tree.find(SVN_NS+"src-path").text)
             path = path_parts[2].strip('/')
+            path_start = request.script_root.strip('/')
+            path = path.lstrip(path_start).strip('/')
             
             # If they asked to start-empty, they want a new checkout
             if entry.attrib.get('start-empty', "false") == "true":