tutor/tests/test_env.py

import os
import tempfile
import unittest
from unittest.mock import Mock, patch

from tests.helpers import PluginsTestCase, temporary_root
from tutor import config as tutor_config
from tutor import env, exceptions, fmt, plugins
from tutor.__about__ import __version__
from tutor.plugins.v0 import DictPlugin
from tutor.types import Config


class EnvTests(PluginsTestCase):
    def test_walk_templates(self) -> None:
        renderer = env.Renderer()
        templates = list(renderer.walk_templates("local"))
        self.assertIn("local/docker-compose.yml", templates)

    def test_walk_templates_partials_are_ignored(self) -> None:
        template_name = "apps/openedx/settings/partials/common_all.py"
        renderer = env.Renderer()
        templates = list(renderer.walk_templates("apps"))
        self.assertIn(template_name, renderer.environment.loader.list_templates())
        self.assertNotIn(template_name, templates)

    def test_files_are_rendered(self) -> None:
        self.assertTrue(env.is_rendered("some/file"))
        self.assertFalse(env.is_rendered(".git"))
        self.assertFalse(env.is_rendered(".git/subdir"))
        self.assertFalse(env.is_rendered("directory/.git"))
        self.assertFalse(env.is_rendered("directory/.git/somefile"))
        self.assertFalse(env.is_rendered("directory/somefile.pyc"))
        self.assertTrue(env.is_rendered("directory/somedir.pyc/somefile"))
        self.assertFalse(env.is_rendered("directory/__pycache__"))
        self.assertFalse(env.is_rendered("directory/__pycache__/somefile"))
        self.assertFalse(env.is_rendered("directory/partials/extra.scss"))
        self.assertFalse(env.is_rendered("directory/partials"))
        self.assertFalse(env.is_rendered("partials/somefile"))

    def test_is_binary_file(self) -> None:
        self.assertTrue(env.is_binary_file("/home/somefile.ico"))

    def test_find_os_path(self) -> None:
        renderer = env.Renderer()
        path = renderer.find_os_path("local/docker-compose.yml")
        self.assertTrue(os.path.exists(path))

    def test_pathjoin(self) -> None:
        with temporary_root() as root:
            self.assertEqual(
                os.path.join(env.base_dir(root), "dummy"), env.pathjoin(root, "dummy")
            )

    def test_render_str(self) -> None:
        self.assertEqual(
            "hello world", env.render_str({"name": "world"}, "hello {{ name }}")
        )

    def test_render_unknown(self) -> None:
        config: Config = {
            "var1": "a",
        }
        self.assertEqual("ab", env.render_unknown(config, "{{ var1 }}b"))
        self.assertEqual({"x": "ac"}, env.render_unknown(config, {"x": "{{ var1 }}c"}))

    def test_common_domain(self) -> None:
        self.assertEqual(
            "mydomain.com",
            env.render_str(
                {"d1": "d1.mydomain.com", "d2": "d2.mydomain.com"},
                "{{ d1|common_domain(d2) }}",
            ),
        )

    def test_render_str_missing_configuration(self) -> None:
        self.assertRaises(exceptions.TutorError, env.render_str, {}, "hello {{ name }}")

    def test_render_file(self) -> None:
        config: Config = {}
        tutor_config.update_with_base(config)
        tutor_config.update_with_defaults(config)
        tutor_config.render_full(config)

        config["MYSQL_ROOT_PASSWORD"] = "testpassword"
        rendered = env.render_file(config, "jobs", "init", "mysql.sh")
        self.assertIn("testpassword", rendered)

    @patch.object(fmt, "echo")
    def test_render_file_missing_configuration(self, _: Mock) -> None:
        self.assertRaises(
            exceptions.TutorError, env.render_file, {}, "local", "docker-compose.yml"
        )

    def test_save_full(self) -> None:
        with temporary_root() as root:
            config = tutor_config.load_full(root)
            with patch.object(fmt, "STDOUT"):
                env.save(root, config)
            self.assertTrue(
                os.path.exists(
                    os.path.join(env.base_dir(root), "local", "docker-compose.yml")
                )
            )

    def test_save_full_with_https(self) -> None:
        with temporary_root() as root:
            config = tutor_config.load_full(root)
            config["ENABLE_HTTPS"] = True
            with patch.object(fmt, "STDOUT"):
                env.save(root, config)
                with open(
                    os.path.join(env.base_dir(root), "apps", "caddy", "Caddyfile"),
                    encoding="utf-8",
                ) as f:
                    self.assertIn("www.myopenedx.com{$default_site_port}", f.read())

    def test_patch(self) -> None:
        patches = {"plugin1": "abcd", "plugin2": "efgh"}
        with patch.object(
            plugins, "iter_patches", return_value=patches.values()
        ) as mock_iter_patches:
            rendered = env.render_str({}, '{{ patch("location") }}')
            mock_iter_patches.assert_called_once_with("location")
        self.assertEqual("abcd\nefgh", rendered)

    def test_patch_separator_suffix(self) -> None:
        patches = {"plugin1": "abcd", "plugin2": "efgh"}
        with patch.object(plugins, "iter_patches", return_value=patches.values()):
            rendered = env.render_str(
                {}, '{{ patch("location", separator=",\n", suffix=",") }}'
            )
        self.assertEqual("abcd,\nefgh,", rendered)

    def test_plugin_templates(self) -> None:
        with tempfile.TemporaryDirectory() as plugin_templates:
            DictPlugin(
                {"name": "plugin1", "version": "0", "templates": plugin_templates}
            )
            # Create two templates
            os.makedirs(os.path.join(plugin_templates, "plugin1", "apps"))
            with open(
                os.path.join(plugin_templates, "plugin1", "unrendered.txt"),
                "w",
                encoding="utf-8",
            ) as f:
                f.write("This file should not be rendered")
            with open(
                os.path.join(plugin_templates, "plugin1", "apps", "rendered.txt"),
                "w",
                encoding="utf-8",
            ) as f:
                f.write("Hello my ID is {{ ID }}")

            # Render templates
            with temporary_root() as root:
                # Create configuration
                config: Config = tutor_config.load_full(root)
                config["ID"] = "Hector Rumblethorpe"
                plugins.load("plugin1")
                tutor_config.save_enabled_plugins(config)

                # Render environment
                with patch.object(fmt, "STDOUT"):
                    env.save(root, config)

                # Check that plugin template was rendered
                root_env = os.path.join(root, "env")
                dst_unrendered = os.path.join(
                    root_env, "plugins", "plugin1", "unrendered.txt"
                )
                dst_rendered = os.path.join(
                    root_env, "plugins", "plugin1", "apps", "rendered.txt"
                )
                self.assertFalse(os.path.exists(dst_unrendered))
                self.assertTrue(os.path.exists(dst_rendered))
                with open(dst_rendered, encoding="utf-8") as f:
                    self.assertEqual("Hello my ID is Hector Rumblethorpe", f.read())

    def test_renderer_is_reset_on_config_change(self) -> None:
        with tempfile.TemporaryDirectory() as plugin_templates:
            plugin1 = DictPlugin(
                {"name": "plugin1", "version": "0", "templates": plugin_templates}
            )

            # Create one template
            os.makedirs(os.path.join(plugin_templates, plugin1.name))
            with open(
                os.path.join(plugin_templates, plugin1.name, "myplugin.txt"),
                "w",
                encoding="utf-8",
            ) as f:
                f.write("some content")

            # Load env once
            config: Config = {"PLUGINS": []}
            env1 = env.Renderer(config).environment

            # Enable plugins
            plugins.load("plugin1")

            # Load env a second time
            config["PLUGINS"] = ["myplugin"]
            env2 = env.Renderer(config).environment

            self.assertNotIn("plugin1/myplugin.txt", env1.loader.list_templates())
            self.assertIn("plugin1/myplugin.txt", env2.loader.list_templates())

    def test_iter_values_named(self) -> None:
        config: Config = {
            "something0_test_app": 0,
            "something1_test_not_app": 1,
            "notsomething_test_app": 2,
            "something3_test_app": 3,
        }
        renderer = env.Renderer(config)
        self.assertEqual([2, 3], list(renderer.iter_values_named(suffix="test_app")))
        self.assertEqual([1, 3], list(renderer.iter_values_named(prefix="something")))
        self.assertEqual(
            [0, 3],
            list(
                renderer.iter_values_named(
                    prefix="something", suffix="test_app", allow_empty=True
                )
            ),
        )


class CurrentVersionTests(unittest.TestCase):
    def test_current_version_in_empty_env(self) -> None:
        with temporary_root() as root:
            self.assertIsNone(env.current_version(root))
            self.assertIsNone(env.get_env_release(root))
            self.assertIsNone(env.should_upgrade_from_release(root))
            self.assertTrue(env.is_up_to_date(root))

    def test_current_version_in_lilac_env(self) -> None:
        with temporary_root() as root:
            os.makedirs(env.base_dir(root))
            with open(
                os.path.join(env.base_dir(root), env.VERSION_FILENAME),
                "w",
                encoding="utf-8",
            ) as f:
                f.write("12.0.46")
            self.assertEqual("12.0.46", env.current_version(root))
            self.assertEqual("lilac", env.get_env_release(root))
            self.assertEqual("lilac", env.should_upgrade_from_release(root))
            self.assertFalse(env.is_up_to_date(root))

    def test_current_version_in_latest_env(self) -> None:
        with temporary_root() as root:
            os.makedirs(env.base_dir(root))
            with open(
                os.path.join(env.base_dir(root), env.VERSION_FILENAME),
                "w",
                encoding="utf-8",
            ) as f:
                f.write(__version__)
            self.assertEqual(__version__, env.current_version(root))
            self.assertEqual("nutmeg", env.get_env_release(root))
            self.assertIsNone(env.should_upgrade_from_release(root))
            self.assertTrue(env.is_up_to_date(root))