Look up directories in a generation.

2 files changed, 124 insertions, 0 deletions

diff --git a/obnam/store.py b/obnam/store.py
index a90a87ef..abd2cc65 100644
--- a/obnam/store.py
+++ b/obnam/store.py
@@ -19,6 +19,7 @@
 
 
 import logging
+import os
 
 import obnam
 
@@ -165,3 +166,42 @@ class Store:
         if object:
             return object
         raise ObjectNotFoundInStore(id)
+
+    def parse_pathname(self, pathname):
+        """Return list of components in pathname."""
+
+        list = []
+        while pathname:
+            dirname = os.path.dirname(pathname)
+            basename = os.path.basename(pathname)
+            if basename:
+                list.insert(0, basename)
+            elif dirname == os.sep:
+                list.insert(0, "/")
+                dirname = ""
+            pathname = dirname
+
+        return list
+
+    def _lookup_dir_from_refs(self, dirrefs, parts):
+        for ref in dirrefs:
+            dir = self.get_object(ref)
+            if dir.get_name() == parts[0]:
+                parts = parts[1:]
+                if parts:
+                    dirrefs = dir.get_dirrefs()
+                    return self._lookup_dir_from_refs(dirrefs, parts)
+                else:
+                    return dir
+        return None
+
+    def lookup_dir(self, generation, pathname):
+        """Return a DirObject that corresponds to pathname in a generation.
+        
+        Look up the directory in the generation. If it does not exist,
+        return None.
+        
+        """
+
+        parts = self.parse_pathname(pathname)
+        return self._lookup_dir_from_refs(generation.get_dirrefs(), parts)
diff --git a/obnam/storeTests.py b/obnam/storeTests.py
index abace511..196547c4 100644
--- a/obnam/storeTests.py
+++ b/obnam/storeTests.py
@@ -18,8 +18,10 @@
 """Unit tests for abstraction for storing backup data, for Obnam."""
 
 
+import os
 import shutil
 import socket
+import tempfile
 import unittest
 
 import obnam
@@ -187,3 +189,85 @@ class StoreMapTests(unittest.TestCase):
         obnam.map.add(self.context.contmap, "pink", "pretty")
         self.store.update_content_maps()
         self.failUnlessEqual(obnam.map.count(self.context.contmap), 1)
+
+
+class StorePathnameParserTests(unittest.TestCase):
+
+    def setUp(self):
+        context = obnam.context.Context()
+        self.store = obnam.Store(context)
+
+    def testReturnsRootForRoot(self):
+        self.failUnlessEqual(self.store.parse_pathname("/"), ["/"])
+
+    def testReturnsDotForDot(self):
+        self.failUnlessEqual(self.store.parse_pathname("."), ["."])
+
+    def testReturnsItselfForSingleElement(self):
+        self.failUnlessEqual(self.store.parse_pathname("foo"), ["foo"])
+
+    def testReturnsListOfPartsForMultipleElements(self):
+        self.failUnlessEqual(self.store.parse_pathname("foo/bar"), 
+                             ["foo", "bar"])
+
+    def testReturnsListOfPartsFromRootForAbsolutePathname(self):
+        self.failUnlessEqual(self.store.parse_pathname("/foo/bar"), 
+                             ["/", "foo", "bar"])
+
+    def testIgnoredTrailingSlashIfNotRoot(self):
+        self.failUnlessEqual(self.store.parse_pathname("foo/bar/"), 
+                             ["foo", "bar"])
+
+
+class StoreLookupTests(unittest.TestCase):
+
+    def create_data_dir(self):
+        dirname = tempfile.mkdtemp()
+        file(os.path.join(dirname, "file"), "w").close()
+        os.mkdir(os.path.join(dirname, "dir1"))
+        os.mkdir(os.path.join(dirname, "dir1", "dir2"))
+        file(os.path.join(dirname, "dir1", "dir2", "file"), "w").close()
+        return dirname
+
+    def create_context(self):
+        context = obnam.context.Context()
+        context.cache = obnam.cache.Cache(context.config)
+        context.be = obnam.backend.init(context.config, context.cache)
+        return context
+
+    def setUp(self):
+        self.datadir = self.create_data_dir()
+        self.dirbasename = os.path.basename(self.datadir)
+
+        app = obnam.Application(self.create_context())
+        app.load_host()
+        gen = app.backup([self.datadir])
+        app.get_store().commit_host_block([gen])
+        
+        self.store = obnam.Store(self.create_context())
+        self.store.fetch_host_block()
+        self.store.load_maps()
+        gen_ids = self.store.get_host_block().get_generation_ids()
+        self.gen = self.store.get_object(gen_ids[0])
+
+    def tearDown(self):
+        shutil.rmtree(self.datadir)
+        shutil.rmtree(self.store._context.config.get("backup", "store"))
+
+    def testFindsBackupRoot(self):
+        dir = self.store.lookup_dir(self.gen, self.dirbasename)
+        self.failUnless(dir.get_name(), self.dirbasename)
+
+    def testFindsFirstSubdir(self):
+        pathname = os.path.join(self.dirbasename, "dir1")
+        dir = self.store.lookup_dir(self.gen, pathname)
+        self.failUnless(dir.get_name(), "dir1")
+
+    def testFindsSecondSubdir(self):
+        pathname = os.path.join(self.dirbasename, "dir1", "dir2")
+        dir = self.store.lookup_dir(self.gen, pathname)
+        self.failUnless(dir.get_name(), "dir2")
+
+    def testDoesNotFindNonExistentDir(self):
+        self.failUnlessEqual(self.store.lookup_dir(self.gen, "notexist"),
+                             None)