Commits

Oliver Tonnhofer  committed 4e6061f

make MBTilesCache API compatible to FileCache implementation

  • Participants
  • Parent commits 829d183

Comments (0)

Files changed (2)

File mapproxy/cache/mbtiles.py

     def is_cached(self, tile):
         if tile.coord is None:
             return True
+        if tile.source:
+            return True
+        
         cur = self.db.cursor()
         cur.execute('''SELECT 1 FROM tiles
             WHERE tile_column = ? AND
             return False
     
     def store_tile(self, tile):
+        if tile.stored:
+            return True
         with tile_buffer(tile) as buf:
             content = buffer(buf.read())
             x, y, level = tile.coord
         content = cur.fetchone()
         if content:
             tile.source = ImageSource(StringIO(content[0]))
+            return True
+        else:
+            return False
     
     def load_tiles_(self, tiles, with_metadata=False):
         #associate the right tiles with the cursor
             tile.source = ImageSource(StringIO(data), size=tile.size)
         cursor.close()
         return tiles
+    
+    def remove_tile(self, tile):
+        cursor = self.db.cursor()
+        cursor.execute(
+            "DELETE FROM tiles WHERE (zoom_level = ? AND tile_column = ? AND tile_row = ?)",
+            tile.coord)
+        if cursor.rowcount:
+            return True
+        return False

File mapproxy/test/unit/test_cache_tile.py

 
 from mapproxy.cache.tile import Tile
 from mapproxy.cache.file import FileCache
+from mapproxy.cache.mbtiles import MBTilesCache
 from mapproxy.image import ImageSource
 from mapproxy.image.opts import ImageOptions
 from mapproxy.test.image import create_tmp_image_buf, is_png
     return abs(timestamp - time.time()) <= delta
 
 class TileCacheTestBase(object):
+    def setup(self):
+        self.cache_dir = tempfile.mkdtemp()
+    
+    def teardown(self):
+        if hasattr(self, 'cache_dir') and os.path.exists(self.cache_dir):
+            shutil.rmtree(self.cache_dir)
+
     def create_tile(self, coord=(0, 0, 0)):
         return Tile(coord,
             ImageSource(tile_image,
         assert not self.cache.is_cached(Tile((0, 0, 0)))
     
     def test_is_cached_hit(self):
-        tile = Tile((0, 0, 0))
+        tile = self.create_tile()
         self.create_cached_tile(tile)
         assert self.cache.is_cached(Tile((0, 0, 0)))
     
         assert tile.is_missing()
     
     def test_load_tile_cached(self):
+        tile = self.create_tile()
+        self.create_cached_tile(tile)
         tile = Tile((0, 0, 0))
-        self.create_cached_tile(tile)
         assert self.cache.load_tile(tile) == True
         assert not tile.is_missing()
 
         assert tile.source is None
         assert self.cache.load_tile(tile, with_metadata=True)
         assert tile.source is not None
-        assert timestamp_is_now(tile.timestamp, delta=5)
-        assert tile.size == size
+        if tile.timestamp:
+            assert timestamp_is_now(tile.timestamp, delta=5)
+        if tile.size:
+            assert tile.size == size
         
     def test_overwrite_tile(self):
         tile = self.create_tile((5, 12, 2))
     def test_store_tile_already_stored(self):
         # tile object is marked as stored, do not save
         # (used for disable_storage)
-        tile = Tile((0, 0, 0), StringIO('foo'))
+        tile = Tile((0, 0, 0), ImageSource(StringIO('foo')))
         tile.stored = True
         self.cache.store_tile(tile)
         
         assert not self.cache.is_cached(tile)
     
     def test_remove(self):
-        tile = Tile((0, 0, 0))
+        tile = self.create_tile((0, 0, 0))
         self.create_cached_tile(tile)
         assert self.cache.is_cached(Tile((0, 0, 0)))
         
-        self.cache.remove_tile(Tile((0, 0, 0)))
+        self.cache.remove_tile(self.create_tile())
         assert not self.cache.is_cached(Tile((0, 0, 0)))
     
+    def create_cached_tile(self, tile):
+        self.cache.store_tile(tile)
     
 class TestFileTileCache(TileCacheTestBase):
     def setup(self):
-        self.cache_dir = tempfile.mkdtemp()
+        TileCacheTestBase.setup(self)
         self.cache = FileCache(self.cache_dir, 'png')
     
     def test_store_tile(self):
         assert os.path.islink(loc)
         assert os.path.realpath(loc).endswith('ff0105ff.png')
         assert is_png(open(loc, 'rb'))
-    
-    
-    def teardown(self):
-        shutil.rmtree(self.cache_dir)
 
     def create_cached_tile(self, tile):
         loc = self.cache.tile_location(tile, create_dir=True)
         with open(loc, 'w') as f:
             f.write('foo')
+
+class TestMBTileCache(TileCacheTestBase):
+    def setup(self):
+        TileCacheTestBase.setup(self)
+        self.cache = MBTilesCache(os.path.join(self.cache_dir, 'tmp.mbtiles'))