<commit_before>#!/usr/bin/env python
"""Test Mapchete main module and processing."""

import os
import yaml
import shutil

from mapchete.cli.main import MapcheteCLI

scriptdir = os.path.dirname(os.path.realpath(__file__))
out_dir = os.path.join(scriptdir, "testdata/tmp")


def test_execute():
    """Run mapchete execute."""
    temp_mapchete = "temp.mapchete"
    temp_process = "temp.py"
    out_format = "GTiff"
    out_dir = os.path.join(scriptdir, "testdata/tmp")
    try:
        # create from template
        args = [
            None, 'create', temp_mapchete, temp_process, out_format,
            "--pyramid_type", "geodetic"]
        MapcheteCLI(args)
        # edit configuration
        with open(temp_mapchete, "r") as config_file:
            config = yaml.load(config_file)
            config["output"].update(bands=1, dtype="uint8", path=".")
        with open(temp_mapchete, "w") as config_file:
            config_file.write(yaml.dump(config, default_flow_style=False))
        # run process for single tile
        input_file = os.path.join(scriptdir, "testdata/cleantopo_br.tif")
        args = [
            None, 'execute', temp_mapchete, '--tile', '6', '62', '124',
            '--input_file', input_file]
        try:
            MapcheteCLI(args)
        except RuntimeError:
            pass
    except:
        raise
    finally:
        delete_files = [temp_mapchete, temp_process, "temp.pyc", "temp.log"]
        for delete_file in delete_files:
            try:
                os.remove(delete_file)
            except:
                pass
        try:
            shutil.rmtree(out_dir)
        except:
            pass


def test_execute_multiprocessing():
    """Run mapchete execute with multiple workers."""
    temp_mapchete = "temp.mapchete"
    temp_process = "temp.py"
    out_format = "GTiff"
    out_dir = os.path.join(scriptdir, "testdata/tmp")
    try:
        # create from template
        args = [
            None, 'create', temp_mapchete, temp_process, out_format,
            "--pyramid_type", "geodetic"]
        MapcheteCLI(args)
        # edit configuration
        with open(temp_mapchete, "r") as config_file:
            config = yaml.load(config_file)
            config["output"].update(bands=1, dtype="uint8", path=".")
        with open(temp_mapchete, "w") as config_file:
            config_file.write(yaml.dump(config, default_flow_style=False))
        # run process with multiprocessing
        input_file = os.path.join(scriptdir, "testdata/cleantopo_br.tif")
        args = [
            None, 'execute', temp_mapchete, '--zoom', '6',
            '--input_file', input_file]
        try:
            MapcheteCLI(args)
        except RuntimeError:
            pass
        # run example process with multiprocessing
        args = [
            None, 'execute', os.path.join(
                scriptdir, "testdata/cleantopo_br.mapchete"),
            '--zoom', '8'
        ]
        MapcheteCLI(args)
    except:
        raise
    finally:
        delete_files = [temp_mapchete, temp_process, "temp.pyc", "temp.log"]
        for delete_file in delete_files:
            try:
                os.remove(delete_file)
            except:
                pass
        try:
            shutil.rmtree(out_dir)
        except:
            pass


# TODO mapchete serve
# TODO mapchete pyramid<commit_msg>added test for mapchete formats CLI command<commit_after>#!/usr/bin/env python
"""Test Mapchete main module and processing."""

import os
import yaml
import shutil

from mapchete.cli.main import MapcheteCLI

scriptdir = os.path.dirname(os.path.realpath(__file__))
out_dir = os.path.join(scriptdir, "testdata/tmp")


def test_execute():
    """Run mapchete execute."""
    temp_mapchete = "temp.mapchete"
    temp_process = "temp.py"
    out_format = "GTiff"
    out_dir = os.path.join(scriptdir, "testdata/tmp")
    try:
        # create from template
        args = [
            None, 'create', temp_mapchete, temp_process, out_format,
            "--pyramid_type", "geodetic"]
        MapcheteCLI(args)
        # edit configuration
        with open(temp_mapchete, "r") as config_file:
            config = yaml.load(config_file)
            config["output"].update(bands=1, dtype="uint8", path=".")
        with open(temp_mapchete, "w") as config_file:
            config_file.write(yaml.dump(config, default_flow_style=False))
        # run process for single tile
        input_file = os.path.join(scriptdir, "testdata/cleantopo_br.tif")
        args = [
            None, 'execute', temp_mapchete, '--tile', '6', '62', '124',
            '--input_file', input_file]
        try:
            MapcheteCLI(args)
        except RuntimeError:
            pass
    except Exception:
        raise
    finally:
        delete_files = [temp_mapchete, temp_process, "temp.pyc", "temp.log"]
        for delete_file in delete_files:
            try:
                os.remove(delete_file)
            except Exception:
                pass
        try:
            shutil.rmtree(out_dir)
        except Exception:
            pass


def test_execute_multiprocessing():
    """Run mapchete execute with multiple workers."""
    temp_mapchete = "temp.mapchete"
    temp_process = "temp.py"
    out_format = "GTiff"
    out_dir = os.path.join(scriptdir, "testdata/tmp")
    try:
        # create from template
        args = [
            None, 'create', temp_mapchete, temp_process, out_format,
            "--pyramid_type", "geodetic"]
        MapcheteCLI(args)
        # edit configuration
        with open(temp_mapchete, "r") as config_file:
            config = yaml.load(config_file)
            config["output"].update(bands=1, dtype="uint8", path=".")
        with open(temp_mapchete, "w") as config_file:
            config_file.write(yaml.dump(config, default_flow_style=False))
        # run process with multiprocessing
        input_file = os.path.join(scriptdir, "testdata/cleantopo_br.tif")
        args = [
            None, 'execute', temp_mapchete, '--zoom', '6',
            '--input_file', input_file]
        try:
            MapcheteCLI(args)
        except RuntimeError:
            pass
        # run example process with multiprocessing
        args = [
            None, 'execute', os.path.join(
                scriptdir, "testdata/cleantopo_br.mapchete"),
            '--zoom', '8'
        ]
        MapcheteCLI(args)
    except Exception:
        raise
    finally:
        delete_files = [temp_mapchete, temp_process, "temp.pyc", "temp.log"]
        for delete_file in delete_files:
            try:
                os.remove(delete_file)
            except Exception:
                pass
        try:
            shutil.rmtree(out_dir)
        except Exception:
            pass


def test_formats(capfd):
    """Output of mapchete formats command."""
    MapcheteCLI([None, 'formats'])
    out, err = capfd.readouterr()
    assert not err
    MapcheteCLI([None, 'formats', '-i'])
    out, err = capfd.readouterr()
    assert not err
    MapcheteCLI([None, 'formats', '-o'])
    out, err = capfd.readouterr()
    assert not err


# TODO mapchete serve
# TODO mapchete pyramid
{'fraction_non_alphanumeric': 0.11308111126622923, 'fraction_numerical': 0.003350551444925311, 'mean_word_length': 2.1134289439374183, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7439878', 'n_tokens_mistral': 2253, 'n_tokens_neox': 2210, 'n_words': 486}