tutor/tests/commands/test_compose.py

import typing as t
import unittest
from io import StringIO
from unittest.mock import patch

from click.exceptions import ClickException

from tutor import hooks
from tutor.commands import compose
from tutor.commands.local import LocalContext


class ComposeTests(unittest.TestCase):

    maxDiff = None  # Ensure we can see long diffs of YAML files.

    def test_mount_option_parsing(self) -> None:
        param = compose.MountParam()

        self.assertEqual(
            [("lms", "/path/to/edx-platform", "/openedx/edx-platform")],
            param("lms:/path/to/edx-platform:/openedx/edx-platform"),
        )
        self.assertEqual(
            [
                ("lms", "/path/to/edx-platform", "/openedx/edx-platform"),
                ("cms", "/path/to/edx-platform", "/openedx/edx-platform"),
            ],
            param("lms,cms:/path/to/edx-platform:/openedx/edx-platform"),
        )
        self.assertEqual(
            [
                ("lms", "/path/to/edx-platform", "/openedx/edx-platform"),
                ("cms", "/path/to/edx-platform", "/openedx/edx-platform"),
            ],
            param("lms, cms:/path/to/edx-platform:/openedx/edx-platform"),
        )
        self.assertEqual(
            [
                ("lms", "/path/to/edx-platform", "/openedx/edx-platform"),
                ("lms-worker", "/path/to/edx-platform", "/openedx/edx-platform"),
            ],
            param("lms,lms-worker:/path/to/edx-platform:/openedx/edx-platform"),
        )
        with self.assertRaises(ClickException):
            param("lms,:/path/to/edx-platform:/openedx/edx-platform")

    @patch("sys.stdout", new_callable=StringIO)
    def test_compose_local_tmp_generation(self, _mock_stdout: StringIO) -> None:
        """
        Ensure that docker-compose.tmp.yml is correctly generated.
        """
        param = compose.MountParam()
        mount_args = (
            # Auto-mounting of edx-platform to lms* and cms*
            param.convert_implicit_form("/path/to/edx-platform"),
            # Manual mounting of some other folder to mfe and lms
            param.convert_explicit_form(
                "mfe,lms:/path/to/something-else:/openedx/something-else"
            ),
        )
        # Mount volumes
        compose.mount_tmp_volumes(mount_args, LocalContext(""))

        compose_file: t.Dict[str, t.Any] = hooks.Filters.COMPOSE_LOCAL_TMP.apply({})
        actual_services: t.Dict[str, t.Any] = compose_file["services"]
        expected_services: t.Dict[str, t.Any] = {
            "cms": {"volumes": ["/path/to/edx-platform:/openedx/edx-platform"]},
            "cms-worker": {"volumes": ["/path/to/edx-platform:/openedx/edx-platform"]},
            "lms": {
                "volumes": [
                    "/path/to/edx-platform:/openedx/edx-platform",
                    "/path/to/something-else:/openedx/something-else",
                ]
            },
            "lms-worker": {"volumes": ["/path/to/edx-platform:/openedx/edx-platform"]},
            "mfe": {"volumes": ["/path/to/something-else:/openedx/something-else"]},
        }
        self.assertEqual(actual_services, expected_services)

        compose_jobs_file: t.Dict[
            str, t.Any
        ] = hooks.Filters.COMPOSE_LOCAL_JOBS_TMP.apply({})
        actual_jobs_services: t.Dict[str, t.Any] = compose_jobs_file["services"]
        expected_jobs_services: t.Dict[str, t.Any] = {
            "cms-job": {"volumes": ["/path/to/edx-platform:/openedx/edx-platform"]},
            "lms-job": {"volumes": ["/path/to/edx-platform:/openedx/edx-platform"]},
        }
        self.assertEqual(actual_jobs_services, expected_jobs_services)