def test_layers_with_opacity(self):
        # overlay with opacity -> request should not be combined
        common_params = (r'?SERVICE=WMS&FORMAT=image%2Fpng'
                                  '&REQUEST=GetMap&HEIGHT=200&SRS=EPSG%3A4326&styles='
                                  '&VERSION=1.1.1&BBOX=9.0,50.0,10.0,51.0'
                                  '&WIDTH=200')

        img_bg = create_tmp_image((200, 200), color=(0, 0, 0))
        img_fg = create_tmp_image((200, 200), color=(255, 0, 128))

        expected_req = [
                        ({'path': '/service_a' + common_params + '&layers=a_one'},
                         {'body': img_bg, 'headers': {'content-type': 'image/png'}}),
                        ({'path': '/service_a' + common_params + '&layers=a_two'},
                         {'body': img_fg, 'headers': {'content-type': 'image/png'}}),
                        ]

        with mock_httpd(('localhost', 42423), expected_req):
            self.common_map_req.params.layers = 'opacity_base,opacity_overlay'
            resp = self.app.get(self.common_map_req)
            eq_(resp.content_type, 'image/png')
            data = BytesIO(resp.body)
            assert is_png(data)
            img = Image.open(data)
            eq_(img.getcolors()[0], ((200*200),(127, 0, 64)))

    def test_get_tile_limited_to(self):
        def auth(service, layers, environ, query_extent, **kw):
            eq_(environ['PATH_INFO'], '/wmts/layer3/GLOBAL_MERCATOR/1/0/0.jpeg')
            eq_(service, 'wmts')
            eq_(len(layers), 1)
            eq_(query_extent[0], 'EPSG:900913')
            assert bbox_equals(query_extent[1], (-20037508.342789244, 0, 0, 20037508.342789244))
            return {
                'authorized': 'partial',
                'limited_to': {
                    'geometry': [-180, -89, -90, 89],
                    'srs': 'EPSG:4326',
                },
                'layers': {
                    'layer3': {'tile': True},
                }
            }

        serv = MockServ(port=42423)
        serv.expects('/1/0/1.png')
        serv.returns(create_tmp_image((256, 256), color=(255, 0, 0)), headers={'content-type': 'image/png'})
        with serv:
            resp = self.app.get('/wmts/layer3/GLOBAL_MERCATOR/1/0/0.jpeg', extra_environ={'mapproxy.authorize': auth})

        eq_(resp.content_type, 'image/png')

        img = img_from_buf(resp.body)
        img = img.convert('RGBA')
        # left part authorized, red
        eq_(img.crop((0, 0, 127, 255)).getcolors()[0], (127*255, (255, 0, 0, 255)))
        # right part not authorized, transparent
        eq_(img.crop((129, 0, 255, 255)).getcolors()[0][1][3], 0)

    def test_get_map_uncached(self):
        mbtiles_file = os.path.join(test_config['base_dir'], 'cache.mbtiles')
        tiles_lock_dir = os.path.join(test_config['base_dir'], 'testlockdir')

        assert os.path.exists(mbtiles_file) # already created on startup
        assert not os.path.exists(tiles_lock_dir)

        self.common_map_req.params.bbox = '-180,0,0,80'
        serv = MockServ(port=42423)
        serv.expects('/tiles/01/000/000/000/000/000/001.png')
        serv.returns(create_tmp_image((256, 256)))
        with serv:
            resp = self.app.get(self.common_map_req)
            eq_(resp.content_type, 'image/png')
            data = StringIO(resp.body)
            assert is_png(data)

        # now cached
        resp = self.app.get(self.common_map_req)
        eq_(resp.content_type, 'image/png')
        data = StringIO(resp.body)
        assert is_png(data)

        # custom tile_lock_dir created
        assert os.path.exists(tiles_lock_dir)

    def test_get_tile_limited_to_inside(self):
        def auth(service, layers, environ, **kw):
            eq_(environ['PATH_INFO'], '/wmts/layer3/GLOBAL_MERCATOR/1/0/0.jpeg')
            eq_(service, 'wmts')
            eq_(len(layers), 1)
            return {
                'authorized': 'partial',
                'limited_to': {
                    'geometry': [-180, -89, 180, 89],
                    'srs': 'EPSG:4326',
                },
                'layers': {
                    'layer3': {'tile': True},
                }
            }

        serv = MockServ(port=42423)
        serv.expects('/1/0/1.png')
        serv.returns(create_tmp_image((256, 256), color=(255, 0, 0)), headers={'content-type': 'image/png'})
        with serv:
            resp = self.app.get('/wmts/layer3/GLOBAL_MERCATOR/1/0/0.jpeg', extra_environ={'mapproxy.authorize': auth})

        eq_(resp.content_type, 'image/jpeg')

        img = img_from_buf(resp.body)
        eq_(img.getcolors()[0], (256*256, (255, 0, 0)))

 def test_sld_file(self):
     self.common_map_req.params['layers'] = 'sld_file'
     with mock_httpd(TESTSERVER_ADDRESS, [
       ({'path': self.common_wms_url + '&sld_body=' +quote('<sld>'), 'method': 'GET'},
        {'body': create_tmp_image((200, 200), format='png')}
       )]):
         resp = self.app.get(self.common_map_req)
         eq_(resp.content_type, 'image/png')

 def test_get_tile_flipped_axis(self):
     serv = MockServ(port=42423)
     # source is ll, cache/service ul
     serv.expects("/tiles/01/000/000/000/000/000/001.png")
     serv.returns(create_tmp_image((256, 256)))
     with serv:
         resp = self.app.get("/wmts/myrest/tms_cache_ul/ulgrid/01/0/0.png", status=200)
         eq_(resp.content_type, "image/png")

 def test_get_tile_flipped_axis(self):
     self.common_tile_req.params['layer'] = 'tms_cache_ul'
     self.common_tile_req.params['tilematrixset'] = 'ulgrid'
     self.common_tile_req.params['format'] = 'image/png'
     self.common_tile_req.tile = (0, 0, '01')
     serv = MockServ(port=42423)
     # source is ll, cache/service ul
     serv.expects('/tiles/01/000/000/000/000/000/001.png')
     serv.returns(create_tmp_image((256, 256)))
     with serv:
         resp = self.app.get(str(self.common_tile_req), status=200)
         eq_(resp.content_type, 'image/png')

    def test_get_map_cached(self):
        # mock_s3 interferes with MockServ, use boto to manually upload tile
        tile = create_tmp_image((256, 256))
        boto3.client("s3").upload_fileobj(
                BytesIO(tile),
                Bucket='default_bucket',
                Key='default_cache/WebMerc/4/1/9.png',
        )

        resp = self.app.get(self.common_map_req)
        eq_(resp.content_type, 'image/png')
        data = BytesIO(resp.body)
        assert is_png(data)

    def test_get_map_cached_quadkey(self):
        # mock_s3 interferes with MockServ, use boto to manually upload tile
        tile = create_tmp_image((256, 256))
        boto3.client("s3").upload_fileobj(
                BytesIO(tile),
                Bucket='tiles',
                Key='quadkeytiles/2003.png',
        )

        self.common_map_req.params.layers = 'quadkey'
        resp = self.app.get(self.common_map_req)
        eq_(resp.content_type, 'image/png')
        data = BytesIO(resp.body)
        assert is_png(data)

    def test_seed_refresh_remove_before_from_file(self):
        # tile already there but old
        t000 = self.make_tile((0, 0, 0), timestamp=time.time() - (60*60*25))

        # mtime is older than tile, no create of the tile
        timestamp = time.time() - (60*60*30)
        os.utime(self.seed_conf_file, (timestamp, timestamp))
        with local_base_config(self.mapproxy_conf.base_config):
            seed_conf  = load_seed_tasks_conf(self.seed_conf_file, self.mapproxy_conf)
            tasks = seed_conf.seeds(['refresh_from_file'])
            seed(tasks, dry_run=False)


        # touch the seed_conf file and refresh everything
        os.utime(self.seed_conf_file, None)
        img_data = create_tmp_image((256, 256), format='png')
        expected_req = ({'path': r'/service?LAYERS=foo&SERVICE=WMS&FORMAT=image%2Fpng'
                              '&REQUEST=GetMap&VERSION=1.1.1&bbox=-180.0,-90.0,180.0,90.0'
                              '&width=256&height=128&srs=EPSG:4326'},
                        {'body': img_data, 'headers': {'content-type': 'image/png'}})
        with mock_httpd(('localhost', 42423), [expected_req]):
            # touch the seed_conf file and refresh everything
            timestamp = time.time() - 60
            os.utime(self.seed_conf_file, (timestamp, timestamp))

            with local_base_config(self.mapproxy_conf.base_config):
                seed_conf  = load_seed_tasks_conf(self.seed_conf_file, self.mapproxy_conf)
                tasks = seed_conf.seeds(['refresh_from_file'])
                seed(tasks, dry_run=False)

                assert os.path.exists(t000)
                assert os.path.getmtime(t000) - 5 < time.time() < os.path.getmtime(t000) + 5

        # mtime is older than tile, no cleanup
        timestamp = time.time() - 5
        os.utime(self.seed_conf_file, (timestamp, timestamp))
        with local_base_config(self.mapproxy_conf.base_config):
            seed_conf  = load_seed_tasks_conf(self.seed_conf_file, self.mapproxy_conf)
            cleanup_tasks = seed_conf.cleanups(['remove_from_file'])
            cleanup(cleanup_tasks, verbose=False, dry_run=False)
        assert os.path.exists(t000)

        # now touch the seed_conf again and remove everything
        timestamp = time.time() + 5
        os.utime(self.seed_conf_file, (timestamp, timestamp))
        with local_base_config(self.mapproxy_conf.base_config):
            seed_conf  = load_seed_tasks_conf(self.seed_conf_file, self.mapproxy_conf)
            cleanup_tasks = seed_conf.cleanups(['remove_from_file'])
            cleanup(cleanup_tasks, verbose=False, dry_run=False)
        assert not os.path.exists(t000)

    def test_get_map_uncached(self):
        self.common_map_req.params.bbox = '-180,0,0,80'
        serv = MockServ(port=42423)
        serv.expects('/tiles/01/000/000/000/000/000/001.png')
        serv.returns(create_tmp_image((256, 256)))
        with serv:
            resp = self.app.get(self.common_map_req)
            eq_(resp.content_type, 'image/png')
            data = StringIO(resp.body)
            assert is_png(data)

        # now cached
        resp = self.app.get(self.common_map_req)
        eq_(resp.content_type, 'image/png')
        data = StringIO(resp.body)
        assert is_png(data)

    def test_get_map_uncached(self):
        assert os.path.exists(os.path.join(test_config['base_dir'], 'cache.gpkg')) # already created on startup

        self.common_map_req.params.bbox = '-180,0,0,80'
        serv = MockServ(port=42423)
        serv.expects('/tiles/01/000/000/000/000/000/001.png')
        serv.returns(create_tmp_image((256, 256)))
        with serv:
            resp = self.app.get(self.common_map_req)
            eq_(resp.content_type, 'image/png')
            data = BytesIO(resp.body)
            assert is_png(data)

        # now cached
        resp = self.app.get(self.common_map_req)
        eq_(resp.content_type, 'image/png')
        data = BytesIO(resp.body)
        assert is_png(data)

    def test_get_tile_flipped_axis(self):
        # test default tile lock directory
        tiles_lock_dir = os.path.join(test_config['base_dir'], 'cache_data', 'tile_locks')
        # make sure default tile_lock_dir was not created by other tests
        shutil.rmtree(tiles_lock_dir, ignore_errors=True)
        assert not os.path.exists(tiles_lock_dir)

        self.common_tile_req.params['layer'] = 'tms_cache_ul'
        self.common_tile_req.params['tilematrixset'] = 'ulgrid'
        self.common_tile_req.params['format'] = 'image/png'
        self.common_tile_req.tile = (0, 0, '01')
        serv = MockServ(port=42423)
        # source is ll, cache/service ul
        serv.expects('/tiles/01/000/000/000/000/000/001.png')
        serv.returns(create_tmp_image((256, 256)))
        with serv:
            resp = self.app.get(str(self.common_tile_req), status=200)
            eq_(resp.content_type, 'image/png')

        # test default tile lock directory was created
        assert os.path.exists(tiles_lock_dir)

    def check_get_tile_limited_to(self, auth_dict):
        def auth(service, layers, environ, query_extent, **kw):
            eq_(environ['PATH_INFO'], '/kml/layer3_EPSG900913/1/0/0.jpeg')
            eq_(service, 'kml')
            eq_(len(layers), 1)
            eq_(query_extent[0], 'EPSG:900913')
            assert bbox_equals(query_extent[1], (-20037508.342789244, -20037508.342789244, 0, 0))
            return auth_dict

        serv = MockServ(port=42423)
        serv.expects('/1/0/0.png')
        serv.returns(create_tmp_image((256, 256), color=(255, 0, 0)), headers={'content-type': 'image/png'})
        with serv:
            resp = self.app.get('/kml/layer3_EPSG900913/1/0/0.jpeg', extra_environ={'mapproxy.authorize': auth})

        eq_(resp.content_type, 'image/png')

        img = img_from_buf(resp.body)
        img = img.convert('RGBA')
        # left part authorized, red
        eq_(img.crop((0, 0, 127, 255)).getcolors()[0], (127*255, (255, 0, 0, 255)))
        # right part not authorized, transparent
        eq_(img.crop((129, 0, 255, 255)).getcolors()[0][1][3], 0)

            def do_POST(self):
                length = int(self.headers['content-length'])
                json_data = self.rfile.read(length)
                task = json.loads(json_data.decode('utf-8'))
                eq_(task['command'], 'tile')
                # request main tile of metatile
                eq_(task['tiles'], [[15, 17, 5]])
                eq_(task['cache_identifier'], 'wms_cache_GLOBAL_MERCATOR')
                eq_(task['priority'], 100)
                # this id should not change for the same tile/cache_identifier combination
                eq_(task['id'], 'aeb52b506e4e82d0a1edf649d56e0451cfd5862c')

                # manually create tile renderd should create
                tile_filename = os.path.join(test_self.config['cache_dir'],
                    'wms_cache_EPSG900913/05/000/000/016/000/000/016.jpeg')
                ensure_directory(tile_filename)
                with open(tile_filename, 'wb') as f:
                    f.write(create_tmp_image((256, 256), format='jpeg', color=(255, 0, 100)))

                self.send_response(200)
                self.send_header('Content-type', 'application/json')
                self.end_headers()
                self.wfile.write(b'{"status": "ok"}')

from io import BytesIO
from mapproxy.request.wms import WMS111FeatureInfoRequest
from mapproxy.test.image import is_png, create_tmp_image
from mapproxy.test.http import mock_httpd
from mapproxy.test.system import module_setup, module_teardown, SystemTest
from nose.tools import eq_

test_config = {}

def setup_module():
    module_setup(test_config, 'arcgis.yaml')

def teardown_module():
    module_teardown(test_config)

transp = create_tmp_image((512, 512), mode='RGBA', color=(0, 0, 0, 0))


class TestArcgisSource(SystemTest):
    config = test_config
    def setup(self):
        SystemTest.setup(self)
        self.common_fi_req = WMS111FeatureInfoRequest(url='/service?',
            param=dict(x='10', y='20', width='200', height='200', layers='app2_with_layers_fi_layer',
                       format='image/png', query_layers='app2_with_layers_fi_layer', styles='',
                       bbox='1000,400,2000,1400', srs='EPSG:3857', info_format='application/json'))

    def test_get_tile(self):
        expected_req = [({'path': '/arcgis/rest/services/ExampleLayer/ImageServer/exportImage?f=image&format=png&imageSR=900913&bboxSR=900913&bbox=-20037508.342789244,-20037508.342789244,20037508.342789244,20037508.342789244&size=512,512'},
                 {'body': transp, 'headers': {'content-type': 'image/png'}}),
                ]

eq_xpath_wmts = functools.partial(eq_xpath, namespaces=ns_wmts)

DIMENSION_LAYER_BASE_REQ = (
    '/service1?styles=&format=image%2Fpng&height=256'
    '&bbox=-20037508.3428,0.0,0.0,20037508.3428'
    '&layers=foo,bar&service=WMS&srs=EPSG%3A900913'
    '&request=GetMap&width=256&version=1.1.1'
)
NO_DIMENSION_LAYER_BASE_REQ = DIMENSION_LAYER_BASE_REQ.replace('/service1?', '/service2?')

WMTS_KVP_URL = (
    '/service?service=wmts&request=GetTile&version=1.0.0'
    '&tilematrixset=GLOBAL_MERCATOR&tilematrix=01&tilecol=0&tilerow=0&format=png&style='
)

TEST_TILE = create_tmp_image((256, 256))

class TestWMTS(SystemTest):
    config = test_config
    def setup(self):
        SystemTest.setup(self)

    def test_capabilities(self):
        resp = self.app.get('/wmts/myrest/1.0.0/WMTSCapabilities.xml')
        xml = resp.lxml
        assert validate_with_xsd(xml, xsd_name='wmts/1.0/wmtsGetCapabilities_response.xsd')

        eq_(len(xml.xpath('//wmts:Layer', namespaces=ns_wmts)), 2)
        eq_(len(xml.xpath('//wmts:Contents/wmts:TileMatrixSet', namespaces=ns_wmts)), 1)

        eq_(set(xml.xpath('//wmts:Contents/wmts:Layer/wmts:ResourceURL/@template', namespaces=ns_wmts)),

from mapproxy.test.system.test_wms import is_111_exception
from nose.tools import eq_

test_config = {}
test_config_raise = {}

def setup_module():
    module_setup(test_config, 'source_errors.yaml')
    module_setup(test_config_raise, 'source_errors_raise.yaml')

def teardown_module():
    module_teardown(test_config)
    module_teardown(test_config_raise)


transp = create_tmp_image((200, 200), mode='RGBA', color=(0, 0, 0, 0))

class TestWMS(SystemTest):
    config = test_config
    def setup(self):
        SystemTest.setup(self)
        self.common_map_req = WMS111MapRequest(url='/service?', param=dict(service='WMS',
             version='1.1.1', bbox='9,50,10,51', width='200', height='200',
             layers='online', srs='EPSG:4326', format='image/png',
             styles='', request='GetMap', transparent=True))

    def test_online(self):
        common_params = (r'?SERVICE=WMS&FORMAT=image%2Fpng'
                                  '&REQUEST=GetMap&HEIGHT=200&SRS=EPSG%3A4326&styles='
                                  '&VERSION=1.1.1&BBOX=9.0,50.0,10.0,51.0'
                                  '&WIDTH=200&transparent=True')

                                  '&REQUEST=GetMap&VERSION=1.1.1&bbox=-180.0,-90.0,180.0,90.0'
                                  '&width=256&height=128&srs=EPSG:4326'},
                            {'body': img_data, 'headers': {'content-type': 'image/png'}})
            with mock_httpd(('localhost', 42423), [expected_req]):
                seed_conf  = load_seed_tasks_conf(self.seed_conf_file, self.mapproxy_conf)
                tasks, cleanup_tasks = seed_conf.seeds(), seed_conf.cleanups()
                seed(tasks, dry_run=False)
                cleanup(cleanup_tasks, verbose=False, dry_run=False)
        
        assert os.path.exists(t000)
        assert os.path.getmtime(t000) - 5 < time.time() < os.path.getmtime(t000) + 5
        assert not os.path.exists(t001)


tile_image_buf = create_tmp_image_buf((256, 256), color='blue')
tile_image = create_tmp_image((256, 256), color='blue')

class TestSeed(SeedTestBase):
    seed_conf_name = 'seed.yaml'
    mapproxy_conf_name = 'seed_mapproxy.yaml'
    
    def test_cleanup_levels(self):
        seed_conf  = load_seed_tasks_conf(self.seed_conf_file, self.mapproxy_conf)
        cleanup_tasks = seed_conf.cleanups(['cleanup'])
        
        self.make_tile((0, 0, 0))
        self.make_tile((0, 0, 1))
        self.make_tile((0, 0, 2))
        self.make_tile((0, 0, 3))
        
        cleanup(cleanup_tasks, verbose=False, dry_run=False)

    def read_tiles(self):
        with open('/tmp/foo.metatile', 'rb') as f:
            tile_positions = self._read_header(f)

            for i, (offset, size) in enumerate(tile_positions):
                f.seek(offset, 0)
                # img = ImageSource(BytesIO(f.read(size)))
                open('/tmp/img-%02d.png' % i, 'wb').write(f.read(size))

if __name__ == '__main__':
    from io import BytesIO
    from mapproxy.cache.tile import Tile
    from mapproxy.test.image import create_tmp_image

    tiles = []
    img = create_tmp_image((256, 256))
    for x in range(8):
        for y in range(8):
            tiles.append(Tile((x, y, 4), ImageSource(BytesIO(img))))

    m = MetaTileFile(None)
    print('!')
    m.write_tiles(tiles)
    print('!')
    m.read_tiles()
    print('!')

    x = y = 0
    METATILE = 8
    for meta in range(METATILE ** 2):
        print(x + (meta / METATILE), y + (meta % METATILE));