from unittest.mock import Mock, patch from tests.helpers import PluginsTestCase, temporary_root from tutor import images, plugins, utils from tutor.__about__ import __version__ from tutor.commands.images import ImageNotFoundError from .base import TestCommandMixin class ImagesTests(PluginsTestCase, TestCommandMixin): def test_images_help(self) -> None: result = self.invoke(["images", "--help"]) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code) def test_images_pull_image(self) -> None: result = self.invoke(["images", "pull"]) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code) def test_images_pull_plugin_invalid_plugin_should_throw_error(self) -> None: result = self.invoke(["images", "pull", "plugin"]) self.assertEqual(1, result.exit_code) self.assertEqual(ImageNotFoundError, type(result.exception)) @patch.object(images, "pull", return_value=None) def test_images_pull_plugin(self, image_pull: Mock) -> None: plugins.v0.DictPlugin( { "name": "plugin1", "hooks": { "remote-image": { "service1": "service1:1.0.0", "service2": "service2:2.0.0", } }, } ) plugins.load("plugin1") result = self.invoke(["images", "pull", "service1"]) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code) image_pull.assert_called_once_with("service1:1.0.0") @patch.object(images, "pull", return_value=None) def test_images_pull_all_vendor_images(self, image_pull: Mock) -> None: result = self.invoke(["images", "pull", "mysql"]) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code) # Note: we should update this tag whenever the mysql image is updated image_pull.assert_called_once_with("docker.io/mysql:8.0.33") def test_images_printtag_image(self) -> None: result = self.invoke(["images", "printtag", "openedx"]) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code) self.assertRegex( result.output, rf"docker.io/overhangio/openedx:{__version__}\n" ) def test_images_printtag_plugin(self) -> None: plugins.v0.DictPlugin( { "name": "plugin1", "hooks": { "build-image": { "service1": "service1:1.0.0", "service2": "service2:2.0.0", } }, } ) plugins.load("plugin1") result = self.invoke(["images", "printtag", "service1"]) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code, result) self.assertEqual(result.output, "service1:1.0.0\n") @patch.object(images, "build", return_value=None) def test_images_build_plugin(self, mock_image_build: Mock) -> None: plugins.v0.DictPlugin( { "name": "plugin1", "hooks": { "build-image": { "service1": "service1:1.0.0", "service2": "service2:2.0.0", } }, } ) plugins.load("plugin1") with temporary_root() as root: self.invoke_in_root(root, ["config", "save"]) result = self.invoke_in_root(root, ["images", "build", "service1"]) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code) mock_image_build.assert_called() self.assertIn("service1:1.0.0", mock_image_build.call_args[0]) @patch.object(images, "build", return_value=None) def test_images_build_plugin_with_args(self, image_build: Mock) -> None: plugins.v0.DictPlugin( { "name": "plugin1", "hooks": { "build-image": { "service1": "service1:1.0.0", "service2": "service2:2.0.0", } }, } ) plugins.load("plugin1") build_args = [ "images", "build", "--no-cache", "-a", "myarg=value", "--add-host", "host", "--target", "target", "-d", "docker_args", "service1", ] with temporary_root() as root: utils.is_buildkit_enabled.cache_clear() with patch.object(utils, "is_buildkit_enabled", return_value=False): self.invoke_in_root(root, ["config", "save"]) result = self.invoke_in_root(root, build_args) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code) image_build.assert_called() self.assertIn("service1:1.0.0", image_build.call_args[0]) self.assertEqual( [ "service1:1.0.0", "--no-cache", "--build-arg", "myarg=value", "--add-host", "host", "--target", "target", "docker_args", "--cache-from=type=registry,ref=service1:1.0.0-cache", ], list(image_build.call_args[0][1:]), ) def test_images_push(self) -> None: result = self.invoke(["images", "push"]) self.assertIsNone(result.exception) self.assertEqual(0, result.exit_code)