def test_explicit_grid(self):
        grid = TileGrid(res=[0.1, 0.05, 0.01])
        eq_(grid.resolution(0), 0.1)
        eq_(grid.resolution(1), 0.05)
        eq_(grid.resolution(2), 0.01)

        eq_(grid.closest_level(0.00001), 2)

class TestWGS84TileGrid(object):
    def setup(self):
        self.grid = TileGrid(is_geodetic=True)

    def test_resolution(self):
        assert_almost_equal(self.grid.resolution(0), 1.40625)
        assert_almost_equal(self.grid.resolution(1), 1.40625/2)

    def test_bbox(self):
        eq_(self.grid.bbox, (-180.0, -90.0, 180.0, 90.0))

    def test_grid_size(self):
        eq_(self.grid.grid_sizes[0], (1, 1))
        eq_(self.grid.grid_sizes[1], (2, 1))
        eq_(self.grid.grid_sizes[2], (4, 2))

    def test_affected_tiles(self):
        bbox, grid, tiles = self.grid.get_affected_tiles((-180,-90,180,90), (512,256))
        eq_(bbox, (-180.0, -90.0, 180.0, 90.0))
        eq_(grid, (2, 1))
        eq_(list(tiles), [(0, 0, 1), (1, 0, 1)])

    def test_affected_level_tiles(self):
        bbox, grid, tiles = self.grid.get_affected_level_tiles((-180,-90,180,90), 1)
        eq_(grid, (2, 1))
        eq_(bbox, (-180.0, -90.0, 180.0, 90.0))
        eq_(list(tiles), [(0, 0, 1), (1, 0, 1)])
        bbox, grid, tiles = self.grid.get_affected_level_tiles((0,0,180,90), 2)
        eq_(grid, (2, 1))
        eq_(bbox, (0.0, 0.0, 180.0, 90.0))
        eq_(list(tiles), [(2, 1, 2), (3, 1, 2)])

 def test_affected_tiles_out_of_grid_bounds(self):
     grid = TileGrid()
     #bbox from open layers
     req_bbox = (-30056262.509599999, -10018754.170400001, -20037508.339999996, -0.00080000050365924835)
     bbox, grid_size, tiles = \
         grid.get_affected_tiles(req_bbox, (256, 256))
     assert_almost_equal_bbox(bbox, req_bbox)
     eq_(grid_size, (1, 1))
     tiles = list(tiles)
     eq_(tiles, [None])

 def test_broken_bbox(self):
     grid = TileGrid()
     # broken request from "ArcGIS Client Using WinInet"
     req_bbox = (-10000855.0573254,2847125.18913603,-9329367.42767611,4239924.78564583)
     try:
         grid.get_affected_tiles(req_bbox, (256, 256), req_srs=SRS(31467))
     except TransformationError:
         pass
     else:
         assert False, 'Expected TransformationError'

 def test_adjacent_tile_bbox(self):
     grid = TileGrid(is_geodetic=True, bbox=(-10, 30, 10, 40), tile_size=(20, 20))
     t1 = grid.tile_bbox((0, 0, 2))
     t2 = grid.tile_bbox((1, 0, 2))
     t3 = grid.tile_bbox((0, 1, 2))
     assert t1[1] == t2[1]
     assert t1[3] == t2[3]
     assert t1[2] == t2[0]
     assert t1[0] == t3[0]
     assert t1[2] == t3[2]
     assert t1[3] == t3[1]

class TestClosestLevelTinyResFactor(object):
    def setup(self):
        self.grid = TileGrid(SRS(31467),
            bbox=[420000,30000,900000,350000], origin='ul',
            res=[4000,3750,3500,3250,3000,2750,2500,2250,2000,1750,1500,1250,1000,750,650,500,250,100,50,20,10,5,2.5,2,1.5,1,0.5],
        )

    def test_closest_level(self):
        eq_(self.grid.closest_level(5000), 0)
        eq_(self.grid.closest_level(4000), 0)
        eq_(self.grid.closest_level(3750), 1)
        eq_(self.grid.closest_level(3500), 2)
        eq_(self.grid.closest_level(3250), 3)
        eq_(self.grid.closest_level(3000), 4)

 def __init__(self, srs, target_dir):
     self.service_srs = srs
     self.sources = {}
     self.caches = {}
     self.layers = []
     self.yaml_file = os.path.join(target_dir, 'mapproxy.yaml')
     self.grid = TileGrid(SRS(3857))

class TestGeodeticTileGrid(TileGridTest):
    def setup(self):
        self.grid = TileGrid(is_geodetic=True, )
    def test_auto_resolution(self):
        grid = TileGrid(is_geodetic=True, bbox=(-10, 30, 10, 40), tile_size=(20, 20))
        tile_bbox = grid.tile_bbox((0, 0, 0))
        assert tile_bbox == (-10, 30, 10, 50)
        assert grid.resolution(0) == 1.0

    def test_grid(self):
        for level, grid_size in [(0, (1, 1)), (1, (2, 1)), (2, (4, 2))]:
            yield self.check_grid, level, grid_size

    def test_adjacent_tile_bbox(self):
        grid = TileGrid(is_geodetic=True, bbox=(-10, 30, 10, 40), tile_size=(20, 20))
        t1 = grid.tile_bbox((0, 0, 2))
        t2 = grid.tile_bbox((1, 0, 2))
        t3 = grid.tile_bbox((0, 1, 2))
        assert t1[1] == t2[1]
        assert t1[3] == t2[3]
        assert t1[2] == t2[0]
        assert t1[0] == t3[0]
        assert t1[2] == t3[2]
        assert t1[3] == t3[1]

    def test_w_resolution(self):
        res = [1, 0.5, 0.2]
        grid = TileGrid(is_geodetic=True, bbox=(-10, 30, 10, 40), tile_size=(20, 20), res=res)
        assert grid.grid_sizes[0] == (1, 1)
        assert grid.grid_sizes[1] == (2, 1)
        assert grid.grid_sizes[2] == (5, 3)

    def test_tile(self):
        assert self.grid.tile(-180, -90, 0) == (0, 0, 0)
        assert self.grid.tile(-180, -90, 1) == (0, 0, 1)
        assert self.grid.tile(-180, -90, 2) == (0, 0, 2)
        assert self.grid.tile(180-0.001, 90-0.001, 0) == (0, 0, 0)
        assert self.grid.tile(10, 50, 1) == (1, 0, 1)

    def test_affected_tiles(self):
        bbox, grid_size, tiles = \
            self.grid.get_affected_tiles((-45,-45,45,45), (512,512))
        assert self.grid.grid_sizes[3] == (8, 4)
        assert bbox == (-45.0, -45.0, 45.0, 45.0)
        assert grid_size == (2, 2)
        tiles = list(tiles)
        assert tiles == [(3, 2, 3), (4, 2, 3), (3, 1, 3), (4, 1, 3)]

 def __init__(self, db_session, srs, target_dir, couchdb_url, template_dir):
     self.db_session = db_session
     self.couchdb_url = couchdb_url
     self.service_srs = srs
     self.template_dir = template_dir
     self.sources = {}
     self.caches = {}
     self.layers = []
     self.yaml_file = os.path.join(target_dir, 'mapproxy.yaml')
     self.grid = TileGrid(SRS(3857))

class TestFixedResolutionsTileGrid(TileGridTest):
    def setup(self):
        self.res = [1000.0, 500.0, 200.0, 100.0, 50.0, 20.0, 5.0]
        bbox = (3250000, 5230000, 3930000, 6110000)
        self.grid = TileGrid(SRS(31467), bbox=bbox, res=self.res)

    def test_resolution(self):
        for level, res in enumerate(self.res):
            assert res == self.grid.resolution(level)

    def test_closest_level(self):
        assert self.grid.closest_level(2000) == 0
        assert self.grid.closest_level(1000) == 0
        assert self.grid.closest_level(950) == 0
        assert self.grid.closest_level(210) == 2

    def test_affected_tiles(self):
        req_bbox = (3250000, 5230000, 3930000, 6110000)
        self.grid.max_shrink_factor = 10
        bbox, grid_size, tiles = \
            self.grid.get_affected_tiles(req_bbox, (256, 256))
        assert bbox == (req_bbox[0], req_bbox[1],
                        req_bbox[0]+1000*256*3, req_bbox[1]+1000*256*4)
        assert grid_size == (3, 4)
        tiles = list(tiles)
        assert tiles == [(0, 3, 0), (1, 3, 0), (2, 3, 0),
                         (0, 2, 0), (1, 2, 0), (2, 2, 0),
                         (0, 1, 0), (1, 1, 0), (2, 1, 0),
                         (0, 0, 0), (1, 0, 0), (2, 0, 0),
                         ]

    def test_affected_tiles_2(self):
        req_bbox = (3250000, 5230000, 3930000, 6110000)
        self.grid.max_shrink_factor = 2.0
        try:
            bbox, grid_size, tiles = \
                self.grid.get_affected_tiles(req_bbox, (256, 256))
        except NoTiles:
            pass
        else:
            assert False, 'got no exception'
    def test_grid(self):
        for level, grid_size in [(0, (3, 4)), (1, (6, 7)), (2, (14, 18))]:
            yield self.check_grid, level, grid_size

    def test_tile_bbox(self):
        tile_bbox = self.grid.tile_bbox((0, 0, 0)) # w: 1000x256
        assert tile_bbox == (3250000.0, 5230000.0, 3506000.0, 5486000.0)
        tile_bbox = self.grid.tile_bbox((0, 0, 1)) # w: 500x256
        assert tile_bbox == (3250000.0, 5230000.0, 3378000.0, 5358000.0)
        tile_bbox = self.grid.tile_bbox((0, 0, 2)) # w: 200x256
        assert tile_bbox == (3250000.0, 5230000.0, 3301200.0, 5281200.0)

 def test_above_first_res(self):
     grid = TileGrid(res=[1000, 500, 250, 100, 50], threshold_res=[1100, 750])
     grid.stretch_factor = 1.1
     eq_(grid.closest_level(1200), 0)
     eq_(grid.closest_level(1100), 0)
     eq_(grid.closest_level(1000), 0)
     eq_(grid.closest_level(800), 0)
     eq_(grid.closest_level(750.1), 0)
     eq_(grid.closest_level(750), 1)

 def test_lower_bound(self):
     # thresholds near the next lower res value
     grid = TileGrid(res=[1000, 500, 250, 100, 50], threshold_res=[300, 110])
     grid.stretch_factor = 1.1
     eq_(grid.closest_level(1100), 0)
     # regular transition (w/stretchfactor)
     eq_(grid.closest_level(950), 0)
     eq_(grid.closest_level(800), 1)
     eq_(grid.closest_level(500), 1)
     # transition at threshold
     eq_(grid.closest_level(301), 1)
     eq_(grid.closest_level(300), 2)
     eq_(grid.closest_level(250), 2)
     # transition at threshold
     eq_(grid.closest_level(111), 2)
     eq_(grid.closest_level(110), 3)
     eq_(grid.closest_level(100), 3)
     # regular transition (w/stretchfactor)
     eq_(grid.closest_level(92), 3)
     eq_(grid.closest_level(90), 4)

 def test_upper_bound(self):
     # thresholds near the next upper res value (within threshold)
     grid = TileGrid(res=[1000, 500, 250, 100, 50], threshold_res=[495, 240])
     grid.stretch_factor = 1.1
     eq_(grid.closest_level(1100), 0)
     # regular transition (w/stretchfactor)
     eq_(grid.closest_level(950), 0)
     eq_(grid.closest_level(800), 1)
     eq_(grid.closest_level(500), 1)
     # transition at threshold
     eq_(grid.closest_level(496), 1)
     eq_(grid.closest_level(495), 2)
     eq_(grid.closest_level(250), 2)
     # transition at threshold (within strechfactor)
     eq_(grid.closest_level(241), 2)
     eq_(grid.closest_level(240), 3)
     eq_(grid.closest_level(100), 3)
     # regular transition (w/stretchfactor)
     eq_(grid.closest_level(92), 3)
     eq_(grid.closest_level(90), 4)

class TestGKTileGrid(TileGridTest):
    def setup(self):
        self.grid = TileGrid(SRS(31467), bbox=(3250000, 5230000, 3930000, 6110000))

    def test_bbox(self):
        assert self.grid.bbox == (3250000, 5230000, 3930000, 6110000)

    def test_resolution(self):
        res = self.grid.resolution(0)
        width = self.grid.bbox[2] - self.grid.bbox[0]
        height = self.grid.bbox[3] - self.grid.bbox[1]
        assert height == 880000.0 and width == 680000.0
        assert res == 880000.0/256

    def test_tile_bbox(self):
        tile_bbox = self.grid.tile_bbox((0, 0, 0))
        assert tile_bbox == (3250000.0, 5230000.0, 4130000.0, 6110000.0)

    def test_tile(self):
        x, y = 3450000, 5890000
        assert [self.grid.tile(x, y, level) for level in range(5)] == \
            [(0, 0, 0), (0, 1, 1), (0, 3, 2), (1, 6, 3), (3, 12, 4)]

    def test_grids(self):
        for level, grid_size in [(0, (1, 1)), (1, (2, 2)), (2, (4, 4)), (3, (7, 8))]:
            yield self.check_grid, level, grid_size

    def test_closest_level(self):
        assert self.grid.closest_level(880000.0/256) == 0
        assert self.grid.closest_level(600000.0/256) == 1
        assert self.grid.closest_level(440000.0/256) == 1
        assert self.grid.closest_level(420000.0/256) == 1

    def test_adjacent_tile_bbox(self):
        t1 = self.grid.tile_bbox((0, 0, 1))
        t2 = self.grid.tile_bbox((1, 0, 1))
        t3 = self.grid.tile_bbox((0, 1, 1))
        assert t1[1] == t2[1]
        assert t1[3] == t2[3]
        assert t1[2] == t2[0]
        assert t1[0] == t3[0]
        assert t1[3] == t3[1]

class MapProxyConfiguration(object):
    def __init__(self, srs, target_dir):
        self.service_srs = srs
        self.sources = {}
        self.caches = {}
        self.layers = []
        self.yaml_file = os.path.join(target_dir, 'mapproxy.yaml')
        self.grid = TileGrid(SRS(3857))

    def _load_sources(self):
        public_wmts = db.session.query(WMTS, ST_Transform(WMTS.view_coverage, 3857)).filter_by(is_public=True).group_by(WMTS).all()

        for wmts, view_coverage in public_wmts:
            self.sources['%s_source' % wmts.name] = {
                'type': 'tile',
                'url': wmts.url,
                'grid': 'GoogleMapsCompatible',
                'coverage': {
                    'srs': 'EPSG:3857',
                    'bbox': list(to_shape(view_coverage).bounds)
                }
            }
            self.caches['%s_cache' % wmts.name] = {
                'sources': ['%s_source' % wmts.name],
                'grids': ['GoogleMapsCompatible'],
                'disable_storage': True
            }
            self.layers.append({
                'name': '%s_layer' % wmts.name,
                'title': wmts.title,
                'sources': ['%s_cache' % wmts.name],
                'min_res': self.grid.resolution(wmts.view_level_start),
                # increase max_res to allow a bit of oversampling
                'max_res': self.grid.resolution(wmts.view_level_end) / 2,
            })

    def _write_mapproxy_yaml(self):
        grids = {
            'GoogleMapsCompatible': {
                'base': 'GLOBAL_MERCATOR',
                'srs': 'EPSG:3857',
                'num_levels': 20,
                'origin': 'nw'
            }
        }

        services = {
            'demo': None,
            'wms': {
                'srs': self.service_srs,
                'md': {'title': _('external_wms_title')}
            },
        }

        globals_ = {
            'image': {
                'paletted': True,
            },
            'cache': {
                'meta_size': [8, 8],
                'meta_buffer': 50,
            },
            'http': {
                'client_timeout': 120,
            },
        }

        config = {}

        if globals_: config['globals'] = globals_
        if grids: config['grids'] = grids
        if self.layers:
            config['layers'] = self.layers
            config['services'] = services
        if self.caches: config['caches'] = self.caches
        if self.sources: config['sources'] = self.sources


        # safe_dump does not output !!python/unicode, etc.
        mapproxy_yaml = yaml.safe_dump(config, default_flow_style=False)
        f = open(self.yaml_file, 'w')
        f.write(mapproxy_yaml)
        f.close()

 def test_tile_out_of_grid_bounds(self):
     grid = TileGrid(is_geodetic=True)
     eq_(grid.tile(-180.01, 50, 1), (-1, 0, 1))

 def test_factor_grid(self):
     grid = TileGrid(is_geodetic=True, res=1/0.75, tile_size=(360, 180))
     eq_(grid.resolution(0), 1.0)
     eq_(grid.resolution(1), 0.75)
     eq_(grid.resolution(2), 0.75*0.75)

 def test_auto_resolution(self):
     grid = TileGrid(is_geodetic=True, bbox=(-10, 30, 10, 40), tile_size=(20, 20))
     tile_bbox = grid.tile_bbox((0, 0, 0))
     assert tile_bbox == (-10, 30, 10, 50)
     assert grid.resolution(0) == 1.0

 def setup(self):
     self.grid = TileGrid(4326, bbox=(-180, -90, 180, 90), origin='ul')

class MapProxyConfiguration(object):
    def __init__(self, db_session, srs, target_dir, couchdb_url, template_dir):
        self.db_session = db_session
        self.couchdb_url = couchdb_url
        self.service_srs = srs
        self.template_dir = template_dir
        self.sources = {}
        self.caches = {}
        self.layers = []
        self.yaml_file = os.path.join(target_dir, 'mapproxy.yaml')
        self.grid = TileGrid(SRS(3857))

    def _load_sources(self):
        local_sources = self.db_session.query(LocalWMTSSource).all()
        for local_source in local_sources:
            wmts_source = local_source.wmts_source
            self.sources[wmts_source.name + '_source'] = {
                'type': 'wms',
                'req': {
                    'url': 'http://dummy.example.org/service?'
                },
                'seed_only': True,
                'coverage': {
                    'srs': 'EPSG:3857',
                    'bbox': list(coverage_from_geojson(wmts_source.download_coverage).bbox)
                }

            }
            self.caches[wmts_source.name + '_cache'] = {
                'sources': [wmts_source.name + '_source'],
                'grids': [wmts_source.matrix_set],
                'cache': {
                    'type': 'couchdb',
                    'url': '%s' % self.couchdb_url,
                    'db_name': wmts_source.name,
                    'tile_metadata': {
                        'tile_col': '{{x}}',
                        'tile_row': '{{y}}',
                        'tile_level': '{{z}}',
                        'created_ts': '{{timestamp}}',
                        'created': '{{utc_iso}}',
                        'center': '{{wgs_tile_centroid}}'
                    }
                }
            }
            self.layers.append({
                'name': wmts_source.name + '_layer',
                'title': wmts_source.title,
                'sources': [wmts_source.name + '_cache'],
                'min_res': self.grid.resolution(local_source.download_level_start),
                # increase max_res to allow a bit of oversampling
                'max_res': self.grid.resolution(local_source.download_level_end) / 2,
            })

    def _write_mapproxy_yaml(self):
        grids = {
            'GoogleMapsCompatible': {
                'base': 'GLOBAL_MERCATOR',
                'srs': 'EPSG:3857',
                'num_levels': 19,
                'origin': 'nw'
            }
        }

        services = {
            'demo': None,
            'wmts': None,
            'wms': {
                'srs': self.service_srs,
                'md': {'title': 'Geobox WMS'}
            },
        }

        globals_ = {
            'image': {
                'paletted': True,
            },
            'cache': {
                'meta_size': [8, 8],
                'meta_buffer': 50,
            },
            'http': {
                'client_timeout': 120,
            },
        }
        if self.template_dir:
            globals_['template_dir'] = self.template_dir

        config = {}

        if globals_: config['globals'] = globals_
        if grids: config['grids'] = grids
        if self.layers:
            config['layers'] = self.layers
            config['services'] = services
        if self.caches: config['caches'] = self.caches
        if self.sources: config['sources'] = self.sources


        # safe_dump does not output !!python/unicode, etc.
#.........这里部分代码省略.........