<commit_before>#!/usr/bin/env python

"""Tests for the dbupdater.py module."""

import sys
import os
import subprocess
import unittest
import dbutil
import dbcompare
from osgeo import gdal
import numpy
import re

#
# Test cases
#

# pylint: disable=too-many-public-methods
#
# Disabled to avoid complaints about the unittest.TestCase class (which has too
# many public methods according to pylint).
# Call this program from the ga-datacube source directory.
# There are three call methods
# 0. python test_landsat_tiler.py 0
# This will put all results into the expected directory.
# 1. python test_landsat_tiler.py 1
# Put all results into the output directory, and compare resulting database,
# with expected database, stopping if databases disagree. If databases agree,
# will proceed to calculating valid pixel counts and difference histograms.
# 2. python test_landsat_tiler.py 2
# As for 1. but continue to histogram stage regardless of database differences.

class TestLandsatTiler(unittest.TestCase):
    """Unit tests for the dbupdater.py script"""

    MODULE = 'landsat_tiler'
    SUITE = 'old_then_new'

    INPUT_DIR = dbutil.input_directory(MODULE, SUITE)
    OUTPUT_DIR = dbutil.output_directory(MODULE, SUITE)
    EXPECTED_DIR = dbutil.expected_directory(MODULE, SUITE)
    #define three modes of execution
    mode_desc_dict = {0: 'Initialise benchmark data in the expected directory',
                      1: 'Do not do ingestion. Compare existing ingestion ' \
                          'in\n %s\n with benchmark\n %s\n' \
                          %(OUTPUT_DIR, EXPECTED_DIR),
                      2: 'Compare  from this run with ' \
                          'expected benchmark database exiting if ' \
                          'they are different',
                      3: 'Compare databases and also compare tiles, even if ' \
                          'the databases are different'}
    if len(sys.argv) < 2:
        mode = -1
    else:
        try:
            mode = int(sys.argv[1])
        except ValueError:
            mode = -1
    if mode not in [0, 1, 2, 3]:
        print 'Please specify a mode as follows:'
        for mode, desc in mode_desc_dict.items():
            print 'python test_landsat_tiler.py %d:\t%s\n' %(mode, desc)
        sys.exit()

    print 'mode=', mode
    if mode == 0:
        OUTPUT_DIR = EXPECTED_DIR
    print 'OUTPUT_DIR =%s' %OUTPUT_DIR

    PQA_CONTIGUITY_BIT = 8

    def setUp(self):
        self.test_dbname = None
        self.expected_dbname = None

        self.test_conn = None
        self.expected_conn = None

        self.logfile = None
        self.bands_expected = None
        self.bands_output = None

        print 'OUTPUT_DIR =%s' %self.OUTPUT_DIR

    def test_landsat_tiler(self):
        """Test the cataloging and tiling of Landsat scences and compare
        resulting database and tile contents with an ingestion benchmark"""
        # This test is intended as an example, and so is extensively
        # commented.
        # Open a log file
        logfile_path = os.path.join(self.OUTPUT_DIR, "test_landsat_tiler.log")
        self.logfile = open(logfile_path, "w")

        #
        # Create the initial database
        #

        # Randomise the name to avoid collisions with other users.
        self.test_dbname = dbutil.random_name("test_tiler")

        # Create the database.
        print 'About to create dbase from %s' \
            %(os.path.join(self.INPUT_DIR, "hypercube_empty.sql"))
        if self.mode != 1:
            dbutil.TESTSERVER.create(self.test_dbname,
                                     self.INPUT_DIR, "hypercube_empty.sql")

        #
        # Run dbupdater on the test database and save the result
        #

        # Create updated datacube_conf file with the new dbname and tile_root
        tile_root = os.path.join(self.OUTPUT_DIR, "tiles")
        configuration_dict = {'dbname': self.test_dbname,
                              'tile_root': tile_root}
        config_file_path = dbutil.update_config_file2(configuration_dict,
                                                     self.INPUT_DIR,
                                                     self.OUTPUT_DIR,
                                                     "test_datacube.conf")

        # Run dbupdater

        ingest_dir = os.path.join(self.INPUT_DIR, 'tiler_testing')
        dbupdater_cmd = ["python",
                         "dbupdater.py",
                         "--debug",
                         "--config=%s" % config_file_path,
                         "--source=%s" % ingest_dir,
                         "--removedblist",
                         "--followsymlinks"]
        if self.mode != 1:
            subprocess.check_call(dbupdater_cmd, stdout=self.logfile,
                                  stderr=subprocess.STDOUT)

        # Run landsat_tiler
        landsat_tiler_cmd = ["python",
                             "landsat_tiler.py",
                             "--config=%s" % config_file_path]
        if self.mode != 1:
            subprocess.check_call(landsat_tiler_cmd, stdout=self.logfile,
                                  stderr=subprocess.STDOUT)
        # Save the updated database<commit_msg>Put argument parsing for test_landsat_tiler into method to enable reuse of TestLandsatTiler.load_and_check method<commit_after>#!/usr/bin/env python

"""Tests for the dbupdater.py module."""

import sys
import os
import subprocess
import unittest
import dbutil
import dbcompare
from osgeo import gdal
import numpy
import re

#
# Test cases
#

# pylint: disable=too-many-public-methods
#
# Disabled to avoid complaints about the unittest.TestCase class (which has too
# many public methods according to pylint).
# Call this program from the ga-datacube source directory.
# There are three call methods
# 0. python test_landsat_tiler.py 0
# This will put all results into the expected directory.
# 1. python test_landsat_tiler.py 1
# Put all results into the output directory, and compare resulting database,
# with expected database, stopping if databases disagree. If databases agree,
# will proceed to calculating valid pixel counts and difference histograms.
# 2. python test_landsat_tiler.py 2
# As for 1. but continue to histogram stage regardless of database differences.

class TestLandsatTiler(unittest.TestCase):
    """Unit tests for the dbupdater.py script"""
    def process_args(self):
        MODULE = 'landsat_tiler'
        SUITE = 'old_then_new'

        self.INPUT_DIR = dbutil.input_directory(MODULE, SUITE)
        self.OUTPUT_DIR = dbutil.output_directory(MODULE, SUITE)
        self.EXPECTED_DIR = dbutil.expected_directory(MODULE, SUITE)
        #define three modes of execution
        mode_desc_dict = {0: 'Initialise benchmark data in the expected directory',
                          1: 'Do not do ingestion. Compare existing ingestion ' \
                          'in\n %s\n with benchmark\n %s\n' \
                          %(self.OUTPUT_DIR, self.EXPECTED_DIR),
                          2: 'Compare  from this run with ' \
                          'expected benchmark database exiting if ' \
                          'they are different',
                          3: 'Compare databases and also compare tiles, even if ' \
                          'the databases are different'}
        if len(sys.argv) < 2:
            mode = -1
        else:
            try:
                mode = int(sys.argv[1])
            except ValueError:
                mode = -1
        msg = ''
        if mode not in [0, 1, 2, 3]:
            msg =  'Please specify a mode as follows:'
            for mode_num, desc in mode_desc_dict.items():
                msg = msg + 'python test_landsat_tiler.py %d:\t%s\n' %(mode_num, desc)
        return mode, msg


        print 'mode=', mode
        if mode == 0:
            self.OUTPUT_DIR = self.EXPECTED_DIR
        print 'OUTPUT_DIR =%s' %self.OUTPUT_DIR

        self.PQA_CONTIGUITY_BIT = 8

    def setUp(self):
        self.test_dbname = None
        self.expected_dbname = None

        self.test_conn = None
        self.expected_conn = None

        self.logfile = None
        self.bands_expected = None
        self.bands_output = None
        self.mode, self.msg = self.process_args()
        print self.mode
        print self.msg

    def test_landsat_tiler(self):
        """Test the cataloging and tiling of Landsat scences and compare
        resulting database and tile contents with an ingestion benchmark"""
        # This test is intended as an example, and so is extensively
        # commented.
        # Open a log file
        if self.mode not in [0, 1, 2, 3]:
            self.skipTest('Skipping test_landsat_tiler since flag is not in [0, 1, 2, 3]')
        logfile_path = os.path.join(self.OUTPUT_DIR, "test_landsat_tiler.log")
        self.logfile = open(logfile_path, "w")

        #
        # Create the initial database
        #

        # Randomise the name to avoid collisions with other users.
        self.test_dbname = dbutil.random_name("test_tiler")

        # Create the database.
        print 'About to create dbase from %s' \
            %(os.path.join(self.INPUT_DIR, "hypercube_empty.sql"))
        if self.mode != 1:
            dbutil.TESTSERVER.create(self.test_dbname,
                                     self.INPUT_DIR, "hypercube_empty.sql")

        #
        # Run dbupdater on the test database and save the result
        #
        # Create updated datacube_conf file with the new dbname and tile_root
        tile_root = os.path.join(self.OUTPUT_DIR, "tiles")
        configuration_dict = {'dbname': self.test_dbname,
                              'tile_root': tile_root}
        config_file_path = dbutil.update_config_file2(configuration_dict,
                                                     self.INPUT_DIR,
                                                     self.OUTPUT_DIR,
                                                     "test_datacube.conf")
        # Run dbupdater

        ingest_dir = os.path.join(self.INPUT_DIR, 'tiler_testing')
        dbupdater_cmd = ["python",
                         "dbupdater.py",
                         "--debug",
                         "--config=%s" % config_file_path,
                         "--source=%s" % ingest_dir,
                         "--removedblist",
                         "--followsymlinks"]
        if self.mode != 1:
            subprocess.check_call(dbupdater_cmd, stdout=self.logfile,
                                  stderr=subprocess.STDOUT)

        # Run landsat_tiler
        landsat_tiler_cmd = ["python",
                             "landsat_tiler.py",
                             "--config=%s" % config_file_path]
        if self.mode != 1:
            subprocess.check_call(landsat_tiler_cmd, stdout=self.logfile,
                                  stderr=subprocess.STDOUT)
        # Save the updated database
{'fraction_non_alphanumeric': 0.08123172514619884, 'fraction_numerical': 0.005299707602339181, 'mean_word_length': 1.810734463276836, 'pattern_counts': {'":': 0, '<': 5, '<?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': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1165695', 'n_tokens_mistral': 3095, 'n_tokens_neox': 2969, 'n_words': 885}