Unification of Galaxy configuration parsing for scripts/*

test/integration/test_scripts.py

@@ -0,0 +1,168 @@
+"""Integration tests for various scripts in scripts/.
+"""
+
+import json
+import os
+import subprocess
+import tempfile
+import unittest
+
+import yaml
+from base import integration_util
+from base.populators import DatasetPopulator
+
+from galaxy.util import galaxy_directory
+
+
+def skip_unless_module(module):
+    available = True
+    try:
+        __import__(module)
+    except ImportError:
+        available = False
+    if available:
+        return lambda func: func
+    template = "Module %s could not be loaded, dependent test skipped."
+    return unittest.skip(template % module)
+
+
+class ScriptsIntegrationTestCase(integration_util.IntegrationTestCase):
+
+    def setUp(self):
+        super(ScriptsIntegrationTestCase, self).setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+        self.config_dir = tempfile.mkdtemp()
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config):
+        cls._raw_config = config
+
+    def test_helper(self):
+        history_id = self.dataset_populator.new_history()
+        dataset = self.dataset_populator.new_dataset(history_id, wait=True)
+        dataset_id = dataset["id"]
+        config_file = self.write_config_file()
+        output = self._scripts_check_output("helper.py", ["-c", config_file, "--decode-id", dataset_id])
+        assert "Decoded " in output
+
+    def test_cleanup(self):
+        history_id = self.dataset_populator.new_history()
+        delete_response = self.dataset_populator._delete("histories/%s" % history_id)
+        assert delete_response.status_code == 200
+        assert delete_response.json()["purged"] is False
+        config_file = self.write_config_file()
+        output = self._scripts_check_output("cleanup_datasets/cleanup_datasets.py", ["-c", config_file, "--days", "0", "--purge_histories"])
+        print(output)
+        history_response = self.dataset_populator._get("histories/%s" % history_id)
+        assert history_response.status_code == 200
+        assert history_response.json()["purged"] is True, history_response.json()
+
+    def test_pgcleanup(self):
+        self._skip_if_not_postgres()
+
+        history_id = self.dataset_populator.new_history()
+        delete_response = self.dataset_populator._delete("histories/%s" % history_id)
+        assert delete_response.status_code == 200
+        assert delete_response.json()["purged"] is False
+        config_file = self.write_config_file()
+        output = self._scripts_check_output("cleanup_datasets/pgcleanup.py", ["-c", config_file, "--older-than", "0", "--sequence", "purge_deleted_histories"])
+        print(output)
+        history_response = self.dataset_populator._get("histories/%s" % history_id)
+        assert history_response.status_code == 200
+        assert history_response.json()["purged"] is True, history_response.json()
+
+    def test_set_user_disk_usage(self):
+        history_id = self.dataset_populator.new_history()
+        self.dataset_populator.new_dataset(history_id, wait=True)
+        config_file = self.write_config_file()
+        output = self._scripts_check_output("set_user_disk_usage.py", ["-c", config_file])
+        # verify the script runs to completion without crashing
+        assert "100% complete" in output, output
+
+    def test_set_dataset_sizes(self):
+        # TODO: change the size of the dataset and verify this works.
+        history_id = self.dataset_populator.new_history()
+        self.dataset_populator.new_dataset(history_id, wait=True)
+        config_file = self.write_config_file()
+        output = self._scripts_check_output("set_dataset_sizes.py", ["-c", config_file])
+        # verify the script runs to completion without crashing
+        assert "Completed 100%" in output, output
+
+    def test_populate_uuid(self):
+        history_id = self.dataset_populator.new_history()
+        self.dataset_populator.new_dataset(history_id, wait=True)
+        config_file = self.write_config_file()
+        output = self._scripts_check_output("cleanup_datasets/populate_uuid.py", ["-c", config_file])
+        assert "Complete" in output
+
+    def test_grt_export(self):
+        self._scripts_check_argparse_help("grt/export.py")
+
+        history_id = self.dataset_populator.new_history()
+        self.dataset_populator.new_dataset(history_id, wait=True)
+        config_file = self.write_config_file()
+        grt_config_file = os.path.join(self.config_dir, "grt.yml")
+        with open(grt_config_file, "w") as f:
+            yaml.dump({"grt": {"shared_toolbox": True}, "sanitization": {"tools": []}, "tool_params": {}}, f)
+        self._scripts_check_output("grt/export.py", ["-c", config_file, "-g", grt_config_file, "-r", self.config_dir])
+        report_files = os.listdir(self.config_dir)
+        json_files = [j for j in report_files if j.endswith(".json")]
+        assert len(json_files) == 1, "Expected one json report file in [%s]" % json_files
+        json_file = os.path.join(self.config_dir, json_files[0])
+        with open(json_file, "r") as f:
+            export = json.load(f)
+        assert export["version"] == 1
+
+    def test_admin_cleanup_datasets(self):
+        self._scripts_check_argparse_help("cleanup_datasets/admin_cleanup_datasets.py")
+
+    @skip_unless_module("flask_socketio")
+    def test_communication_server(self):
+        self._scripts_check_argparse_help("communication/communication_server.py")
+
+    def test_secret_decoder_ring(self):
+        self._scripts_check_argparse_help("secret_decoder_ring.py")
+
+        config_file = self.write_config_file()
+        output = self._scripts_check_output("secret_decoder_ring.py", ["-c", config_file, "encode", "1"])
+        encoded_id = output.strip()
+
+        output = self._scripts_check_output("secret_decoder_ring.py", ["-c", config_file, "decode", encoded_id])
+        assert output.strip() == "1"
+
+    def test_database_scripts(self):
+        self._scripts_check_argparse_help("create_db.py")
+        self._scripts_check_argparse_help("manage_db.py")
+        # TODO: test creating a smaller database - e.g. tool install database based on fresh
+        # config file.
+
+    def test_runtime_stats(self):
+        self._skip_if_not_postgres()
+        self._scripts_check_argparse_help("runtime_stats.py")
+
+    def _skip_if_not_postgres(self):
+        if not self._app.config.database_connection.startswith("post"):
+            raise unittest.SkipTest("Test only valid for postgres")
+
+    def _scripts_check_argparse_help(self, script):
+        # Test imports and argparse repsonse to --help with 0 exit code.
+        output = self._scripts_check_output(script, ["--help"])
+        # Test -h, --help in printed output message.
+        assert "-h, --help" in output
+
+    def _scripts_check_output(self, script, args):
+        cwd = galaxy_directory()
+        cmd = ["python", os.path.join(cwd, "scripts", script)] + args
+        clean_env = {
+            "PATH": os.environ.get("PATH", None),
+        }  # Don't let testing environment variables interfere with config.
+        return subprocess.check_output(cmd, cwd=cwd, env=clean_env)
+
+    def write_config_file(self):
+        config_dir = self.config_dir
+        path = os.path.join(config_dir, "galaxy.yml")
+        self._test_driver.temp_directories.extend([config_dir])
+        with open(path, "w") as f:
+            yaml.dump({"galaxy": self._raw_config}, f)
+
+        return path