<commit_before># -*- coding: utf-8 -*-
import logging
import os
from osgeo import gdal, ogr
from mock import Mock, patch, call
from uuid import uuid4
from django.test import TestCase

from ..gdalutils import open_ds, cleanup_ds, is_envelope, clip_dataset, convert, get_meta

logger = logging.getLogger(__name__)


class TestGdalUtils(TestCase):
    def setUp(self, ):
        self.path = os.path.dirname(os.path.realpath(__file__))
        self.task_process_patcher = patch('eventkit_cloud.utils.gdalutils.TaskProcess')
        self.task_process = self.task_process_patcher.start()
        self.addCleanup(self.task_process_patcher.stop)
        self.task_uid = uuid4()

    @patch('eventkit_cloud.utils.gdalutils.os.path.isfile')
    @patch('eventkit_cloud.utils.gdalutils.open_ds')
    def test_get_meta(self, open_ds_mock, isfile):

        dataset_path = "/path/to/dataset"
        isfile.return_value = True
        self.task_process.return_value = Mock(exitcode=0)

        ds_mock = Mock(spec=gdal.Dataset)
        ds_mock.RasterCount = 0
        open_ds_mock.return_value = ds_mock
        ds_mock.GetDriver.return_value.ShortName = 'gtiff'
        expected_meta = {'driver': 'gtiff', 'is_raster': True, 'nodata': None}
        returned_meta = get_meta(dataset_path)
        self.assertEqual(expected_meta, returned_meta)

        ds_mock.RasterCount = 2
        ds_mock.GetRasterBand.return_value.GetNoDataValue.return_value = -32768.0
        expected_meta = {'driver': 'gtiff', 'is_raster': True, 'nodata': -32768.0}
        returned_meta = get_meta(dataset_path)
        self.assertEqual(expected_meta, returned_meta)

        ds_mock = Mock(spec=ogr.DataSource)
        open_ds_mock.return_value = ds_mock
        ds_mock.GetDriver.return_value.GetName.return_value = 'gpkg'
        expected_meta = {'driver': 'gpkg', 'is_raster': False, 'nodata': None}
        returned_meta = get_meta(dataset_path)
        self.assertEqual(expected_meta, returned_meta)

        open_ds_mock.return_value = None
        expected_meta = {'driver': None, 'is_raster': None, 'nodata': None}
        returned_meta = get_meta(dataset_path)
        self.assertEqual(expected_meta, returned_meta)

    def test_is_envelope(self):
        envelope_gj = """{"type": "MultiPolygon",
            "coordinates": [ [
                [   [0,0],
                    [1,0],
                    [1,1],
                    [0,1],
                    [0,0]
                ]
            ] ]
        }"""
        triangle_gj = """{"type": "MultiPolygon",
            "coordinates": [ [
                [   [0,0],
                    [1,0],
                    [0,1],
                    [0,0]
                ]
            ] ]
        }"""
        non_env_gj = """{"type": "MultiPolygon",
            "coordinates": [ [
                [   [0,0],
                    [1.5,0],
                    [1,1],
                    [0,1],
                    [0,0]
                ]
            ] ]
        }"""
        empty_gj = ""

        self.assertTrue(is_envelope(envelope_gj))
        self.assertFalse(is_envelope(triangle_gj))
        self.assertFalse(is_envelope(non_env_gj))
        self.assertFalse(is_envelope(empty_gj))

    @patch('eventkit_cloud.utils.gdalutils.is_envelope')
    @patch('eventkit_cloud.utils.gdalutils.get_meta')
    @patch('eventkit_cloud.utils.gdalutils.os.path.isfile')
    def test_clip_dataset(self, isfile, get_meta_mock, is_envelope_mock):

        isfile.return_value = True

        with self.assertRaises(Exception):
            clip_dataset(boundary=None, in_dataset=None)

        # Raster geopackage
        geojson_file = "/path/to/geojson"
        dataset = "/path/to/dataset"
        in_dataset = "/path/to/old_dataset"
        fmt = "gpkg"
        band_type = "-ot byte"
        expected_cmd = "gdalwarp -cutline {} -crop_to_cutline {} -of {} {} {} {}".format(
            geojson_file,
            "-dstalpha",
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        get_meta_mock.return_value = {'driver': 'gpkg', 'is_raster': True, 'nodata': None}
        self.task_process.return_value = Mock(exitcode=0)
        clip_dataset(boundary=geojson_file, in_dataset=in_dataset, out_dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Geotiff
        fmt = "gtiff"
        band_type = ""
        expected_cmd = "gdalwarp -cutline {} -crop_to_cutline {} -of {} {} {} {}".format(
            geojson_file,
            "",
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        get_meta_mock.return_value = {'driver': 'gtiff', 'is_raster': True, 'nodata': None}
        is_envelope_mock.return_value = True  # So, no need for -dstalpha
        clip_dataset(boundary=geojson_file, in_dataset=in_dataset, out_dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Geotiff with non-envelope polygon cutline
        expected_cmd = "gdalwarp -cutline {} -crop_to_cutline {} -of {} {} {} {}".format(
            geojson_file,
            "-dstalpha",
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        is_envelope_mock.return_value = False
        clip_dataset(boundary=geojson_file, in_dataset=in_dataset, out_dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Vector
        fmt = "gpkg"
        expected_cmd = "ogr2ogr -f {0} -clipsrc {1} {2} {3}".format(
            fmt,
            geojson_file,
            dataset,
            in_dataset
        )
        get_meta_mock.return_value = {'driver': 'gpkg', 'is_raster': False}
        clip_dataset(boundary=geojson_file, in_dataset=in_dataset, out_dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

    @patch('eventkit_cloud.utils.gdalutils.get_meta')
    @patch('eventkit_cloud.utils.gdalutils.os.path.isfile')
    @patch('eventkit_cloud.utils.gdalutils.os.rename')
    def test_convert(self, rename, isfile, get_meta_mock):

        isfile.return_value = True

        with self.assertRaises(Exception):
            convert(dataset=None)

        # No-op (same source and destination format)
        dataset = "/path/to/dataset"
        in_dataset = "/path/to/old_dataset"
        fmt = "gpkg"
        get_meta_mock.return_value = {'driver': 'gpkg', 'is_raster': True}
        self.task_process.return_value = Mock(exitcode=0)
        convert(dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_not_called()

        # Raster geopackage from geotiff
        dataset = "/path/to/dataset"
        in_dataset = "/path/to/old_dataset"
        fmt = "gpkg"
        band_type = "-ot byte"
        expected_cmd = "gdalwarp -of {0} {1} {2} {3}".format(
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        get_meta_mock.return_value = {'driver': 'gtiff', 'is_raster': True, 'nodata': None}
        self.task_process.return_value = Mock(exitcode=0)
        convert(dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Geotiff from raster geopackage
        fmt = "gtiff"
        band_type = ""
        expected_cmd = "gdalwarp -of {0} {1} {2} {3}".format(
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        get_meta_mock.return_value = {'driver': 'gpkg', 'is_raster': True}
        convert(dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Vector
        fmt = "gpkg"
        expected_cmd = "ogr2ogr -f {0} {1} {2}".format(
            fmt,
            dataset,
            in_dataset
        )
        get_meta_mock.return_value = {'driver': 'geojson', 'is_raster': False}
        convert(dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)
<commit_msg>Fix gdalutils clip test setup<commit_after># -*- coding: utf-8 -*-
import logging
import os
from osgeo import gdal, ogr
from mock import Mock, patch, call
from uuid import uuid4
from django.test import TestCase

from ..gdalutils import open_ds, cleanup_ds, is_envelope, clip_dataset, convert, get_meta

logger = logging.getLogger(__name__)


class TestGdalUtils(TestCase):
    def setUp(self, ):
        self.path = os.path.dirname(os.path.realpath(__file__))
        self.task_process_patcher = patch('eventkit_cloud.utils.gdalutils.TaskProcess')
        self.task_process = self.task_process_patcher.start()
        self.addCleanup(self.task_process_patcher.stop)
        self.task_uid = uuid4()

    @patch('eventkit_cloud.utils.gdalutils.os.path.isfile')
    @patch('eventkit_cloud.utils.gdalutils.open_ds')
    def test_get_meta(self, open_ds_mock, isfile):

        dataset_path = "/path/to/dataset"
        isfile.return_value = True
        self.task_process.return_value = Mock(exitcode=0)

        ds_mock = Mock(spec=gdal.Dataset)
        ds_mock.RasterCount = 0
        open_ds_mock.return_value = ds_mock
        ds_mock.GetDriver.return_value.ShortName = 'gtiff'
        expected_meta = {'driver': 'gtiff', 'is_raster': True, 'nodata': None}
        returned_meta = get_meta(dataset_path)
        self.assertEqual(expected_meta, returned_meta)

        ds_mock.RasterCount = 2
        ds_mock.GetRasterBand.return_value.GetNoDataValue.return_value = -32768.0
        expected_meta = {'driver': 'gtiff', 'is_raster': True, 'nodata': -32768.0}
        returned_meta = get_meta(dataset_path)
        self.assertEqual(expected_meta, returned_meta)

        ds_mock = Mock(spec=ogr.DataSource)
        open_ds_mock.return_value = ds_mock
        ds_mock.GetDriver.return_value.GetName.return_value = 'gpkg'
        expected_meta = {'driver': 'gpkg', 'is_raster': False, 'nodata': None}
        returned_meta = get_meta(dataset_path)
        self.assertEqual(expected_meta, returned_meta)

        open_ds_mock.return_value = None
        expected_meta = {'driver': None, 'is_raster': None, 'nodata': None}
        returned_meta = get_meta(dataset_path)
        self.assertEqual(expected_meta, returned_meta)

    def test_is_envelope(self):
        envelope_gj = """{"type": "MultiPolygon",
            "coordinates": [ [
                [   [0,0],
                    [1,0],
                    [1,1],
                    [0,1],
                    [0,0]
                ]
            ] ]
        }"""
        triangle_gj = """{"type": "MultiPolygon",
            "coordinates": [ [
                [   [0,0],
                    [1,0],
                    [0,1],
                    [0,0]
                ]
            ] ]
        }"""
        non_env_gj = """{"type": "MultiPolygon",
            "coordinates": [ [
                [   [0,0],
                    [1.5,0],
                    [1,1],
                    [0,1],
                    [0,0]
                ]
            ] ]
        }"""
        empty_gj = ""

        self.assertTrue(is_envelope(envelope_gj))
        self.assertFalse(is_envelope(triangle_gj))
        self.assertFalse(is_envelope(non_env_gj))
        self.assertFalse(is_envelope(empty_gj))

    @patch('eventkit_cloud.utils.gdalutils.is_envelope')
    @patch('eventkit_cloud.utils.gdalutils.get_meta')
    @patch('eventkit_cloud.utils.gdalutils.os.path.isfile')
    def test_clip_dataset(self, isfile, get_meta_mock, is_envelope_mock):

        isfile.return_value = True

        with self.assertRaises(Exception):
            clip_dataset(boundary=None, in_dataset=None)

        # Raster geopackage
        geojson_file = "/path/to/geojson"
        dataset = "/path/to/dataset"
        in_dataset = "/path/to/old_dataset"
        fmt = "gpkg"
        band_type = "-ot byte"
        expected_cmd = "gdalwarp -cutline {} -crop_to_cutline {} -of {} {} {} {}".format(
            geojson_file,
            "-dstalpha",
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        get_meta_mock.return_value = {'driver': 'gpkg', 'is_raster': True, 'nodata': None}
        is_envelope_mock.return_value = False
        self.task_process.return_value = Mock(exitcode=0)
        clip_dataset(boundary=geojson_file, in_dataset=in_dataset, out_dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Geotiff
        fmt = "gtiff"
        band_type = ""
        expected_cmd = "gdalwarp -cutline {} -crop_to_cutline {} -of {} {} {} {}".format(
            geojson_file,
            "",
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        get_meta_mock.return_value = {'driver': 'gtiff', 'is_raster': True, 'nodata': None}
        is_envelope_mock.return_value = True  # So, no need for -dstalpha
        clip_dataset(boundary=geojson_file, in_dataset=in_dataset, out_dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Geotiff with non-envelope polygon cutline
        expected_cmd = "gdalwarp -cutline {} -crop_to_cutline {} -of {} {} {} {}".format(
            geojson_file,
            "-dstalpha",
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        is_envelope_mock.return_value = False
        clip_dataset(boundary=geojson_file, in_dataset=in_dataset, out_dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Vector
        fmt = "gpkg"
        expected_cmd = "ogr2ogr -f {0} -clipsrc {1} {2} {3}".format(
            fmt,
            geojson_file,
            dataset,
            in_dataset
        )
        get_meta_mock.return_value = {'driver': 'gpkg', 'is_raster': False}
        clip_dataset(boundary=geojson_file, in_dataset=in_dataset, out_dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

    @patch('eventkit_cloud.utils.gdalutils.get_meta')
    @patch('eventkit_cloud.utils.gdalutils.os.path.isfile')
    @patch('eventkit_cloud.utils.gdalutils.os.rename')
    def test_convert(self, rename, isfile, get_meta_mock):

        isfile.return_value = True

        with self.assertRaises(Exception):
            convert(dataset=None)

        # No-op (same source and destination format)
        dataset = "/path/to/dataset"
        in_dataset = "/path/to/old_dataset"
        fmt = "gpkg"
        get_meta_mock.return_value = {'driver': 'gpkg', 'is_raster': True}
        self.task_process.return_value = Mock(exitcode=0)
        convert(dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_not_called()

        # Raster geopackage from geotiff
        dataset = "/path/to/dataset"
        in_dataset = "/path/to/old_dataset"
        fmt = "gpkg"
        band_type = "-ot byte"
        expected_cmd = "gdalwarp -of {0} {1} {2} {3}".format(
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        get_meta_mock.return_value = {'driver': 'gtiff', 'is_raster': True, 'nodata': None}
        self.task_process.return_value = Mock(exitcode=0)
        convert(dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Geotiff from raster geopackage
        fmt = "gtiff"
        band_type = ""
        expected_cmd = "gdalwarp -of {0} {1} {2} {3}".format(
            fmt,
            band_type,
            in_dataset,
            dataset
        )
        get_meta_mock.return_value = {'driver': 'gpkg', 'is_raster': True}
        convert(dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

        # Vector
        fmt = "gpkg"
        expected_cmd = "ogr2ogr -f {0} {1} {2}".format(
            fmt,
            dataset,
            in_dataset
        )
        get_meta_mock.return_value = {'driver': 'geojson', 'is_raster': False}
        convert(dataset=dataset, fmt=fmt, task_uid=self.task_uid)
        self.task_process().start_process.assert_called_with(expected_cmd, executable='/bin/bash', shell=True,
                                                             stderr=-1, stdout=-1)

{'fraction_non_alphanumeric': 0.11847147706120661, 'fraction_numerical': 0.00886893682251177, 'mean_word_length': 2.0148539362931177, 'pattern_counts': {'":': 12, '<': 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': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3944328', 'n_tokens_mistral': 5751, 'n_tokens_neox': 5602, 'n_words': 902}