diff --git a/pyproject.toml b/pyproject.toml index 0c3a835..99ef4d9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,9 +54,6 @@ requires = [ packages = ["scuba", "tests"] warn_unused_configs = true warn_return_any = true - -[[tool.mypy.overrides]] -module = "scuba.*" disallow_untyped_calls = true disallow_untyped_defs = true disallow_incomplete_defs = true diff --git a/tests/conftest.py b/tests/conftest.py index dd4793e..3483ffd 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,8 +1,9 @@ +from pathlib import Path import pytest @pytest.fixture -def in_tmp_path(tmp_path, monkeypatch): +def in_tmp_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: """Runs a test in a temporary directory provided by the tmp_path fixture""" monkeypatch.chdir(tmp_path) return tmp_path diff --git a/tests/test_config.py b/tests/test_config.py index 65aa924..95b255e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,10 +1,8 @@ -# coding=utf-8 import logging import os -from os.path import join from pathlib import Path import pytest -from shutil import rmtree +from typing import Optional from unittest import mock from .utils import assert_paths_equal, assert_vol @@ -13,8 +11,23 @@ from scuba.config import ScubaVolume -def load_config() -> scuba.config.ScubaConfig: - return scuba.config.load_config(Path(".scuba.yml"), Path.cwd()) +SCUBA_YML = Path(".scuba.yml") +GITLAB_YML = Path(".gitlab.yml") + + +def load_config(*, config_text: Optional[str] = None) -> scuba.config.ScubaConfig: + if config_text is not None: + SCUBA_YML.write_text(config_text) + return scuba.config.load_config(SCUBA_YML, Path.cwd()) + + +def invalid_config( + *, + config_text: Optional[str] = None, + error_match: Optional[str] = None, +) -> None: + with pytest.raises(scuba.config.ConfigError, match=error_match) as e: + load_config(config_text=config_text) class TestCommonScriptSchema: @@ -57,98 +70,86 @@ def test_script_key_mapping_invalid(self) -> None: @pytest.mark.usefixtures("in_tmp_path") -class TestConfig: - ###################################################################### - # Find config +class ConfigTest: + pass + - def test_find_config_cur_dir(self, in_tmp_path) -> None: +class TestFindConfig(ConfigTest): + def test_find_config_cur_dir(self, in_tmp_path: Path) -> None: """find_config can find the config in the current directory""" - with open(".scuba.yml", "w") as f: - f.write("image: bosybux\n") + SCUBA_YML.write_text("image: bosybux") path, rel, _ = scuba.config.find_config() assert_paths_equal(path, in_tmp_path) assert_paths_equal(rel, "") - def test_find_config_parent_dir(self, in_tmp_path) -> None: + def test_find_config_parent_dir(self, in_tmp_path: Path) -> None: """find_config cuba can find the config in the parent directory""" - with open(".scuba.yml", "w") as f: - f.write("image: bosybux\n") + SCUBA_YML.write_text("image: bosybux") - os.mkdir("subdir") - os.chdir("subdir") + subdir = Path("subdir") + subdir.mkdir() + os.chdir(subdir) # Verify our current working dir - assert_paths_equal(os.getcwd(), in_tmp_path.joinpath("subdir")) + assert_paths_equal(Path.cwd(), in_tmp_path / subdir) path, rel, _ = scuba.config.find_config() assert_paths_equal(path, in_tmp_path) - assert_paths_equal(rel, "subdir") + assert_paths_equal(rel, subdir) - def test_find_config_way_up(self, in_tmp_path) -> None: + def test_find_config_way_up(self, in_tmp_path: Path) -> None: """find_config can find the config way up the directory hierarchy""" - with open(".scuba.yml", "w") as f: - f.write("image: bosybux\n") - - subdirs = ["foo", "bar", "snap", "crackle", "pop"] + SCUBA_YML.write_text("image: bosybux") - for sd in subdirs: - os.mkdir(sd) - os.chdir(sd) + subdir = Path("foo/bar/snap/crackle/pop") + subdir.mkdir(parents=True) + os.chdir(subdir) # Verify our current working dir - assert_paths_equal(os.getcwd(), in_tmp_path.joinpath(*subdirs)) + assert_paths_equal(Path.cwd(), in_tmp_path / subdir) path, rel, _ = scuba.config.find_config() assert_paths_equal(path, in_tmp_path) - assert_paths_equal(rel, join(*subdirs)) + assert_paths_equal(rel, subdir) def test_find_config_nonexist(self) -> None: """find_config raises ConfigError if the config cannot be found""" with pytest.raises(scuba.config.ConfigError): scuba.config.find_config() - ###################################################################### - # Load config - - def _invalid_config(self, match=None): - with pytest.raises(scuba.config.ConfigError, match=match) as e: - load_config() +class TestLoadConfig(ConfigTest): def test_load_config_no_image(self) -> None: """load_config raises ConfigError if the config is empty and image is referenced""" - with open(".scuba.yml", "w") as f: - pass - - config = load_config() + config = load_config(config_text="") with pytest.raises(scuba.config.ConfigError): img = config.image def test_load_unexpected_node(self) -> None: """load_config raises ConfigError on unexpected config node""" - with open(".scuba.yml", "w") as f: - f.write("image: bosybux\n") - f.write("unexpected_node_123456: value\n") - - self._invalid_config() + invalid_config( + config_text=""" + image: bosybux + unexpected_node_123456: value + """ + ) def test_load_config_minimal(self) -> None: """load_config loads a minimal config""" - with open(".scuba.yml", "w") as f: - f.write("image: bosybux\n") - - config = load_config() + config = load_config(config_text="image: bosybux") assert config.image == "bosybux" def test_load_config_with_aliases(self) -> None: """load_config loads a config with aliases""" - with open(".scuba.yml", "w") as f: - f.write("image: bosybux\n") - f.write("aliases:\n") - f.write(" foo: bar\n") - f.write(" snap: crackle pop\n") - - config = load_config() + config = load_config( + config_text=""" + image: bosybux + aliases: + foo: bar + snap: crackle pop + """ + ) assert config.image == "bosybux" assert len(config.aliases) == 2 assert config.aliases["foo"].script == ["bar"] @@ -156,158 +157,146 @@ def test_load_config_with_aliases(self) -> None: def test_load_config__no_spaces_in_aliases(self) -> None: """load_config refuses spaces in aliases""" - with open(".scuba.yml", "w") as f: - f.write("image: bosybux\n") - f.write("aliases:\n") - f.write(" this has spaces: whatever\n") - - self._invalid_config() + invalid_config( + config_text=""" + image: bosybux + aliases: + this has spaces: whatever + """ + ) def test_load_config_image_from_yaml(self) -> None: """load_config loads a config using !from_yaml""" - with open(".gitlab.yml", "w") as f: - f.write("image: dummian:8.2\n") - - with open(".scuba.yml", "w") as f: - f.write("image: !from_yaml .gitlab.yml image\n") - - config = load_config() + GITLAB_YML.write_text("image: dummian:8.2") + config = load_config(config_text=f"image: !from_yaml {GITLAB_YML} image") assert config.image == "dummian:8.2" def test_load_config_image_from_yaml_nested_keys(self) -> None: """load_config loads a config using !from_yaml with nested keys""" - with open(".gitlab.yml", "w") as f: - f.write("somewhere:\n") - f.write(" down:\n") - f.write(" here: dummian:8.2\n") - - with open(".scuba.yml", "w") as f: - f.write("image: !from_yaml .gitlab.yml somewhere.down.here\n") - - config = load_config() + GITLAB_YML.write_text( + """ + somewhere: + down: + here: dummian:8.2 + """ + ) + config = load_config( + config_text=f"image: !from_yaml {GITLAB_YML} somewhere.down.here" + ) assert config.image == "dummian:8.2" def test_load_config_image_from_yaml_nested_keys_with_escaped_characters( self, ) -> None: """load_config loads a config using !from_yaml with nested keys containing escaped '.' characters""" - with open(".gitlab.yml", "w") as f: - f.write(".its:\n") - f.write(" somewhere.down:\n") - f.write(" here: dummian:8.2\n") - - with open(".scuba.yml", "w") as f: - f.write('image: !from_yaml .gitlab.yml "\\.its.somewhere\\.down.here"\n') - - config = load_config() + GITLAB_YML.write_text( + """ + .its: + somewhere.down: + here: dummian:8.2 + """ + ) + config = load_config( + config_text=f'image: !from_yaml {GITLAB_YML} "\\.its.somewhere\\.down.here"\n' + ) assert config.image == "dummian:8.2" def test_load_config_from_yaml_cached_file(self) -> None: """load_config loads a config using !from_yaml from cached version""" - with open(".gitlab.yml", "w") as f: - f.write("one: dummian:8.2\n") - f.write("two: dummian:9.3\n") - f.write("three: dummian:10.1\n") - - with open(".scuba.yml", "w") as f: - f.write("image: !from_yaml .gitlab.yml one\n") - f.write("aliases:\n") - f.write(" two:\n") - f.write(" image: !from_yaml .gitlab.yml two\n") - f.write(" script: ugh\n") - f.write(" three:\n") - f.write(" image: !from_yaml .gitlab.yml three\n") - f.write(" script: ugh\n") + GITLAB_YML.write_text( + """ + one: dummian:8.2 + two: dummian:9.3 + three: dummian:10.1 + """ + ) + SCUBA_YML.write_text( + f""" + image: !from_yaml {GITLAB_YML} one + aliases: + two: + image: !from_yaml {GITLAB_YML} two + script: ugh + three: + image: !from_yaml {GITLAB_YML} three + script: ugh + """ + ) with mock.patch.object(Path, "open", autospec=True, side_effect=Path.open) as m: + # Write config separately so its open() call is not counted here. config = load_config() - # Assert that .gitlab.yml was only opened once + # Assert that GITLAB_YML was only opened once assert m.mock_calls == [ - mock.call(Path(".scuba.yml"), "r"), - mock.call(Path(".gitlab.yml"), "r"), + mock.call(SCUBA_YML, "r"), + mock.call(GITLAB_YML, "r"), ] def test_load_config_image_from_yaml_nested_key_missing(self) -> None: """load_config raises ConfigError when !from_yaml references nonexistant key""" - with open(".gitlab.yml", "w") as f: - f.write("somewhere:\n") - f.write(" down:\n") - - with open(".scuba.yml", "w") as f: - f.write("image: !from_yaml .gitlab.yml somewhere.NONEXISTANT\n") - - self._invalid_config() + GITLAB_YML.write_text( + """ + somewhere: + down: + """ + ) + invalid_config( + config_text=f"image: !from_yaml {GITLAB_YML} somewhere.NONEXISTANT" + ) def test_load_config_image_from_yaml_missing_file(self) -> None: """load_config raises ConfigError when !from_yaml references nonexistant file""" - with open(".scuba.yml", "w") as f: - f.write("image: !from_yaml .NONEXISTANT.yml image\n") - - self._invalid_config() + invalid_config(config_text="image: !from_yaml .NONEXISTANT.yml image") def test_load_config_image_from_yaml_unicode_args(self) -> None: """load_config !from_yaml works with unicode args""" - with open(".gitlab.yml", "w") as f: - f.write("𝕦𝕟𝕚𝕔𝕠𝕕𝕖: 𝕨𝕠𝕣𝕜𝕤:𝕠𝕜\n") - - with open(".scuba.yml", "w") as f: - f.write("image: !from_yaml .gitlab.yml 𝕦𝕟𝕚𝕔𝕠𝕕𝕖\n") - - config = load_config() + GITLAB_YML.write_text("𝕦𝕟𝕚𝕔𝕠𝕕𝕖: 𝕨𝕠𝕣𝕜𝕤:𝕠𝕜") + config = load_config(config_text=f"image: !from_yaml {GITLAB_YML} 𝕦𝕟𝕚𝕔𝕠𝕕𝕖") assert config.image == "𝕨𝕠𝕣𝕜𝕤:𝕠𝕜" def test_load_config_image_from_yaml_missing_arg(self) -> None: """load_config raises ConfigError when !from_yaml has missing args""" - with open(".gitlab.yml", "w") as f: - f.write("image: dummian:8.2\n") - - with open(".scuba.yml", "w") as f: - f.write("image: !from_yaml .gitlab.yml\n") - - self._invalid_config() - - def __test_load_config_safe(self, bad_yaml_path) -> None: - with open(bad_yaml_path, "w") as f: - f.write("danger:\n") - f.write(" - !!python/object/apply:print [Danger]\n") - f.write(" - !!python/object/apply:sys.exit [66]\n") - + GITLAB_YML.write_text("image: dummian:8.2") + invalid_config(config_text=f"image: !from_yaml {GITLAB_YML}") + + def __test_load_config_safe(self, bad_yaml_path: Path) -> None: + bad_yaml_path.write_text( + """ + danger: + - !!python/object/apply:print [Danger] + - !!python/object/apply:sys.exit [66] + """ + ) pat = "could not determine a constructor for the tag.*python/object/apply" with pytest.raises(scuba.config.ConfigError, match=pat) as ctx: load_config() def test_load_config_safe(self) -> None: """load_config safely loads yaml""" - self.__test_load_config_safe(".scuba.yml") + self.__test_load_config_safe(SCUBA_YML) def test_load_config_safe_external(self) -> None: """load_config safely loads yaml from external files""" - with open(".scuba.yml", "w") as f: - f.write("image: !from_yaml .external.yml danger\n") + external_yml = Path(".external.yml") + SCUBA_YML.write_text(f"image: !from_yaml {external_yml} danger") + self.__test_load_config_safe(external_yml) - self.__test_load_config_safe(".external.yml") - - ############################################################################ - # Hooks +class TestConfigHooks(ConfigTest): def test_hooks_mixed(self) -> None: """hooks of mixed forms are valid""" - with open(".scuba.yml", "w") as f: - f.write( - """ - image: na - hooks: - root: - script: - - echo "This runs before we switch users" - - id - user: id - """ - ) - - config = load_config() - + config = load_config( + config_text=""" + image: na + hooks: + root: + script: + - echo "This runs before we switch users" + - id + user: id + """ + ) assert config.hooks.get("root") == [ 'echo "This runs before we switch users"', "id", @@ -316,71 +305,58 @@ def test_hooks_mixed(self) -> None: def test_hooks_invalid_list(self) -> None: """hooks with list not under "script" key are invalid""" - with open(".scuba.yml", "w") as f: - f.write( - """ - image: na - hooks: - user: - - this list should be under - - a 'script' - """ - ) - - self._invalid_config() + invalid_config( + config_text=""" + image: na + hooks: + user: + - this list should be under + - a 'script' + """ + ) def test_hooks_missing_script(self) -> None: """hooks with dict, but missing "script" are invalid""" - with open(".scuba.yml", "w") as f: - f.write( - """ - image: na - hooks: - user: - not_script: missing "script" key - """ - ) - - self._invalid_config() + invalid_config( + config_text=""" + image: na + hooks: + user: + not_script: missing "script" key + """ + ) - ############################################################################ - # Env +class TestConfigEnv(ConfigTest): def test_env_invalid(self) -> None: """Environment must be dict or list of strings""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - environment: 666 - """ - ) - self._invalid_config("must be list or mapping") - - def test_env_top_dict(self, monkeypatch) -> None: - """Top-level environment can be loaded (dict)""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - environment: - FOO: This is foo - FOO_WITH_QUOTES: "\"Quoted foo\"" # Quotes included in value - BAR: "This is bar" - MAGIC: 42 - SWITCH_1: true # YAML boolean - SWITCH_2: "true" # YAML string - EMPTY: "" - EXTERNAL: # Comes from os env - EXTERNAL_NOTSET: # Missing in os env - """ - ) + invalid_config( + config_text=r""" + image: na + environment: 666 + """, + error_match="must be list or mapping", + ) + def test_env_top_dict(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Top-level environment can be loaded (dict)""" monkeypatch.setenv("EXTERNAL", "Outside world") monkeypatch.delenv("EXTERNAL_NOTSET", raising=False) - - config = load_config() - + config = load_config( + config_text=r""" + image: na + environment: + FOO: This is foo + FOO_WITH_QUOTES: "\"Quoted foo\"" # Quotes included in value + BAR: "This is bar" + MAGIC: 42 + SWITCH_1: true # YAML boolean + SWITCH_2: "true" # YAML string + EMPTY: "" + EXTERNAL: # Comes from os env + EXTERNAL_NOTSET: # Missing in os env + """ + ) expect = dict( FOO="This is foo", FOO_WITH_QUOTES='"Quoted foo"', @@ -394,29 +370,24 @@ def test_env_top_dict(self, monkeypatch) -> None: ) assert expect == config.environment - def test_env_top_list(self, monkeypatch) -> None: + def test_env_top_list(self, monkeypatch: pytest.MonkeyPatch) -> None: """Top-level environment can be loaded (list)""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - environment: - - FOO=This is foo # No quotes - - FOO_WITH_QUOTES="Quoted foo" # Quotes included in value - - BAR=This is bar - - MAGIC=42 - - SWITCH_2=true - - EMPTY= - - EXTERNAL # Comes from os env - - EXTERNAL_NOTSET # Missing in os env - """ - ) - monkeypatch.setenv("EXTERNAL", "Outside world") monkeypatch.delenv("EXTERNAL_NOTSET", raising=False) - - config = load_config() - + config = load_config( + config_text=r""" + image: na + environment: + - FOO=This is foo # No quotes + - FOO_WITH_QUOTES="Quoted foo" # Quotes included in value + - BAR=This is bar + - MAGIC=42 + - SWITCH_2=true + - EMPTY= + - EXTERNAL # Comes from os env + - EXTERNAL_NOTSET # Missing in os env + """ + ) expect = dict( FOO="This is foo", FOO_WITH_QUOTES='"Quoted foo"', @@ -431,295 +402,229 @@ def test_env_top_list(self, monkeypatch) -> None: def test_env_alias(self) -> None: """Alias can have environment""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - aliases: - al: - script: Don't care - environment: - FOO: Overridden - MORE: Hello world - """ - ) - - config = load_config() - + config = load_config( + config_text=r""" + image: na + aliases: + al: + script: Don't care + environment: + FOO: Overridden + MORE: Hello world + """ + ) assert config.aliases["al"].environment == dict( FOO="Overridden", MORE="Hello world", ) - ############################################################################ - # Entrypoint +class TestConfigEntrypoint(ConfigTest): def test_entrypoint_not_set(self) -> None: """Entrypoint can be missing""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - """ - ) - - config = load_config() + config = load_config(config_text="image: na") assert config.entrypoint is None def test_entrypoint_null(self) -> None: """Entrypoint can be set to null""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - entrypoint: - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + entrypoint: + """ + ) assert config.entrypoint == "" # Null => empty string def test_entrypoint_invalid(self) -> None: """Entrypoint of incorrect type raises ConfigError""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - entrypoint: 666 - """ - ) - - self._invalid_config("must be a string") + invalid_config( + config_text=r""" + image: na + entrypoint: 666 + """, + error_match="must be a string", + ) - def test_entrypoint_emptry_string(self) -> None: + def test_entrypoint_empty_string(self) -> None: """Entrypoint can be set to an empty string""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - entrypoint: "" - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + entrypoint: "" + """ + ) assert config.entrypoint == "" def test_entrypoint_set(self) -> None: """Entrypoint can be set""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - entrypoint: my_ep - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + entrypoint: my_ep + """ + ) assert config.entrypoint == "my_ep" def test_alias_entrypoint_null(self) -> None: """Entrypoint can be set to null via alias""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - entrypoint: na_ep - aliases: - testalias: - entrypoint: - script: - - ugh - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + entrypoint: na_ep + aliases: + testalias: + entrypoint: + script: + - ugh + """ + ) assert config.aliases["testalias"].entrypoint == "" # Null => empty string def test_alias_entrypoint_empty_string(self) -> None: """Entrypoint can be set to an empty string via alias""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - entrypoint: na_ep - aliases: - testalias: - entrypoint: "" - script: - - ugh - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + entrypoint: na_ep + aliases: + testalias: + entrypoint: "" + script: + - ugh + """ + ) assert config.aliases["testalias"].entrypoint == "" def test_alias_entrypoint(self) -> None: """Entrypoint can be set via alias""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - entrypoint: na_ep - aliases: - testalias: - entrypoint: use_this_ep - script: - - ugh - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + entrypoint: na_ep + aliases: + testalias: + entrypoint: use_this_ep + script: + - ugh + """ + ) assert config.aliases["testalias"].entrypoint == "use_this_ep" - ############################################################################ - # docker_args +class TestConfigDockerArgs(ConfigTest): def test_docker_args_not_set(self) -> None: """docker_args can be missing""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - """ - ) - - config = load_config() + config = load_config(config_text="image: na") assert config.docker_args is None def test_docker_args_invalid(self) -> None: """docker_args of incorrect type raises ConfigError""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: 666 - """ - ) - - self._invalid_config("must be a string") + invalid_config( + config_text=r""" + image: na + docker_args: 666 + """, + error_match="must be a string", + ) def test_docker_args_null(self) -> None: """docker_args can be set to null""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: + """ + ) assert config.docker_args == [] def test_docker_args_set_empty_string(self) -> None: """docker_args can be set to empty string""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: '' - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: '' + """ + ) assert config.docker_args == [] # '' -> [] after shlex.split() def test_docker_args_set(self) -> None: """docker_args can be set""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: --privileged + """ + ) assert config.docker_args == ["--privileged"] def test_docker_args_set_multi(self) -> None: """docker_args can be set to multiple args""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged -v /tmp/:/tmp/ - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: --privileged -v /tmp/:/tmp/ + """ + ) assert config.docker_args == ["--privileged", "-v", "/tmp/:/tmp/"] def test_alias_docker_args_null(self) -> None: """docker_args can be set to null via alias""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged - aliases: - testalias: - docker_args: - script: - - ugh - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: --privileged + aliases: + testalias: + docker_args: + script: + - ugh + """ + ) assert config.aliases["testalias"].docker_args == [] def test_alias_docker_args_empty_string(self) -> None: """docker_args can be set to empty string via alias""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged - aliases: - testalias: - docker_args: '' - script: - - ugh - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: --privileged + aliases: + testalias: + docker_args: '' + script: + - ugh + """ + ) assert config.aliases["testalias"].docker_args == [] def test_alias_docker_args_set(self) -> None: """docker_args can be set via alias""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged - aliases: - testalias: - docker_args: -v /tmp/:/tmp/ - script: - - ugh - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: --privileged + aliases: + testalias: + docker_args: -v /tmp/:/tmp/ + script: + - ugh + """ + ) assert config.aliases["testalias"].docker_args == ["-v", "/tmp/:/tmp/"] def test_alias_docker_args_override(self) -> None: """docker_args can be tagged for override""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged - aliases: - testalias: - docker_args: !override -v /tmp/:/tmp/ - script: - - ugh - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: --privileged + aliases: + testalias: + docker_args: !override -v /tmp/:/tmp/ + script: + - ugh + """ + ) assert config.aliases["testalias"].docker_args == ["-v", "/tmp/:/tmp/"] assert isinstance( config.aliases["testalias"].docker_args, scuba.config.OverrideMixin @@ -727,20 +632,17 @@ def test_alias_docker_args_override(self) -> None: def test_alias_docker_args_override_implicit_null(self) -> None: """docker_args can be overridden with an implicit null value""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged - aliases: - testalias: - docker_args: !override - script: - - ugh - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + docker_args: --privileged + aliases: + testalias: + docker_args: !override + script: + - ugh + """ + ) assert config.aliases["testalias"].docker_args == [] assert isinstance( config.aliases["testalias"].docker_args, scuba.config.OverrideMixin @@ -748,23 +650,20 @@ def test_alias_docker_args_override_implicit_null(self) -> None: def test_alias_docker_args_override_from_yaml(self) -> None: """!override tag can be applied before a !from_yaml tag""" - with open("args.yml", "w") as f: - f.write("args: -v /tmp/:/tmp/\n") - - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged - aliases: - testalias: - docker_args: !override '!from_yaml args.yml args' - script: - - ugh - """ - ) - - config = load_config() + args_yml = Path("args.yml") + args_yml.write_text("args: -v /tmp/:/tmp/") + + config = load_config( + config_text=rf""" + image: na + docker_args: --privileged + aliases: + testalias: + docker_args: !override '!from_yaml {args_yml} args' + script: + - ugh + """ + ) assert config.aliases["testalias"].docker_args == ["-v", "/tmp/:/tmp/"] assert isinstance( config.aliases["testalias"].docker_args, scuba.config.OverrideMixin @@ -772,147 +671,119 @@ def test_alias_docker_args_override_from_yaml(self) -> None: def test_alias_docker_args_from_yaml_override(self) -> None: """!override tag can be applied inside of a !from_yaml tag""" - with open("args.yml", "w") as f: - f.write("args: !override -v /tmp/:/tmp/\n") - - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - docker_args: --privileged - aliases: - testalias: - docker_args: !from_yaml args.yml args - script: - - ugh - """ - ) - - config = load_config() + args_yml = Path("args.yml") + args_yml.write_text("args: !override -v /tmp/:/tmp/") + + config = load_config( + config_text=rf""" + image: na + docker_args: --privileged + aliases: + testalias: + docker_args: !from_yaml {args_yml} args + script: + - ugh + """ + ) assert config.aliases["testalias"].docker_args == ["-v", "/tmp/:/tmp/"] assert isinstance( config.aliases["testalias"].docker_args, scuba.config.OverrideMixin ) - ############################################################################ - # volumes - def test_volumes_not_set(self) -> None: +class TestConfigVolumes(ConfigTest): + def test_not_set(self) -> None: """volumes can be missing""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - """ - ) - - config = load_config() + config = load_config(config_text="image: na") assert config.volumes is None - def test_volumes_null(self) -> None: + def test_null(self) -> None: """volumes can be set to null""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + """ + ) assert config.volumes == None - def test_volumes_invalid(self) -> None: - """volumes of incorrect type raises ConfigError""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: 666 - """ - ) - - self._invalid_config("must be a dict") + def test_invalid_int(self) -> None: + """volumes of incorrect type (int) raises ConfigError""" + invalid_config( + config_text=r""" + image: na + volumes: 666 + """, + error_match="must be a dict", + ) - def test_volumes_invalid_volume_type(self) -> None: + def test_invalid_list(self) -> None: """volume of incorrect type (list) raises ConfigError""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: - - a list makes no sense - """ - ) - - self._invalid_config("must be string or dict") + invalid_config( + config_text=r""" + image: na + volumes: + /foo: + - a list makes no sense + """, + error_match="must be string or dict", + ) - def test_volumes_null_volume_type(self) -> None: + def test_null_volume_type(self) -> None: """volume of None type raises ConfigError""" # NOTE: In the future, we might want to support this as a volume # (non-bindmount, e.g. '-v /somedata'), or as tmpfs - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /bar: - """ - ) - - self._invalid_config("hostpath") + invalid_config( + config_text=r""" + image: na + volumes: + /bar: + """, + error_match="hostpath", + ) - def test_volume_as_dict_missing_hostpath(self) -> None: + def test_complex_missing_hostpath(self) -> None: """volume of incorrect type raises ConfigError""" # NOTE: In the future, we might want to support this as a volume # (non-bindmount, e.g. '-v /somedata'), or as tmpfs - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /bar: - options: hostpath,is,missing - """ - ) - - self._invalid_config("hostpath") + invalid_config( + config_text=r""" + image: na + volumes: + /bar: + options: hostpath,is,missing + """, + error_match="hostpath", + ) - def test_volumes_simple_bind(self) -> None: + def test_simple_bind(self) -> None: """volumes can be set using the simple form""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /cpath: /hpath - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + /cpath: /hpath + """ + ) assert config.volumes is not None assert len(config.volumes) == 1 assert_vol(config.volumes, "/cpath", "/hpath") - def test_volumes_complex_bind(self) -> None: + def test_complex_bind(self) -> None: """volumes can be set using the complex form""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: /host/foo - /bar: - hostpath: /host/bar - /snap: - hostpath: /host/snap - options: z,ro - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + /foo: /host/foo + /bar: + hostpath: /host/bar + /snap: + hostpath: /host/snap + options: z,ro + """ + ) vols = config.volumes assert vols is not None assert len(vols) == 3 @@ -921,19 +792,16 @@ def test_volumes_complex_bind(self) -> None: assert_vol(vols, "/bar", "/host/bar") assert_vol(vols, "/snap", "/host/snap", ["z", "ro"]) - def test_volumes_complex_named_volume(self) -> None: + def test_complex_named_volume(self) -> None: """volumes complex form can specify a named volume""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: - name: foo-volume - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + /foo: + name: foo-volume + """ + ) assert config.volumes is not None assert len(config.volumes) == 1 vol = config.volumes[Path("/foo")] @@ -942,25 +810,22 @@ def test_volumes_complex_named_volume(self) -> None: assert_paths_equal(vol.container_path, "/foo") assert vol.volume_name == "foo-volume" - def test_alias_volumes_set(self) -> None: - """docker_args can be set via alias""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - aliases: - testalias: - script: - - ugh - volumes: - /foo: /host/foo - /bar: - hostpath: /host/bar - options: z,ro - """ - ) - - config = load_config() + def test_via_alias(self) -> None: + """volumes can be set via alias""" + config = load_config( + config_text=r""" + image: na + aliases: + testalias: + script: + - ugh + volumes: + /foo: /host/foo + /bar: + hostpath: /host/bar + options: z,ro + """ + ) vols = config.aliases["testalias"].volumes assert vols is not None assert len(vols) == 2 @@ -968,47 +833,41 @@ def test_alias_volumes_set(self) -> None: assert_vol(vols, "/foo", "/host/foo") assert_vol(vols, "/bar", "/host/bar", ["z", "ro"]) - def test_volumes_with_env_vars_simple(self, monkeypatch) -> None: + def test_with_env_vars_simple(self, monkeypatch: pytest.MonkeyPatch) -> None: """volume definitions can contain environment variables""" monkeypatch.setenv("TEST_VOL_PATH", "/bar/baz") monkeypatch.setenv("TEST_VOL_PATH2", "/moo/doo") - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - $TEST_VOL_PATH/foo: ${TEST_VOL_PATH2}/foo - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + $TEST_VOL_PATH/foo: ${TEST_VOL_PATH2}/foo + """ + ) vols = config.volumes assert vols is not None assert len(vols) == 1 assert_vol(vols, "/bar/baz/foo", "/moo/doo/foo") - def test_volumes_with_env_vars_complex(self, monkeypatch) -> None: + def test_with_env_vars_complex(self, monkeypatch: pytest.MonkeyPatch) -> None: """complex volume definitions can contain environment variables""" monkeypatch.setenv("TEST_HOME", "/home/testuser") monkeypatch.setenv("TEST_TMP", "/tmp") monkeypatch.setenv("TEST_MAIL", "/var/spool/mail/testuser") - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - $TEST_HOME/.config: ${TEST_HOME}/.config - $TEST_TMP/: - hostpath: $TEST_HOME/scuba/myproject/tmp - /var/spool/mail/container: - hostpath: $TEST_MAIL - options: z,ro - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + $TEST_HOME/.config: ${TEST_HOME}/.config + $TEST_TMP/: + hostpath: $TEST_HOME/scuba/myproject/tmp + /var/spool/mail/container: + hostpath: $TEST_MAIL + options: z,ro + """ + ) vols = config.volumes assert vols is not None assert len(vols) == 3 @@ -1019,35 +878,35 @@ def test_volumes_with_env_vars_complex(self, monkeypatch) -> None: vols, "/var/spool/mail/container", "/var/spool/mail/testuser", ["z", "ro"] ) - def test_volumes_with_invalid_env_vars(self, monkeypatch) -> None: + def test_with_invalid_env_vars(self, monkeypatch: pytest.MonkeyPatch) -> None: """Volume definitions cannot include unset env vars""" # Ensure that the entry does not exist in the environment monkeypatch.delenv("TEST_VAR1", raising=False) - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - $TEST_VAR1/foo: /host/foo - """ - ) - self._invalid_config("TEST_VAR1") + invalid_config( + config_text=r""" + image: na + volumes: + $TEST_VAR1/foo: /host/foo + """, + error_match="TEST_VAR1", + ) - def test_volumes_hostpath_rel(self, monkeypatch, in_tmp_path) -> None: + def test_hostpath_rel( + self, monkeypatch: pytest.MonkeyPatch, in_tmp_path: Path + ) -> None: """volume hostpath can be relative to scuba_root (top dir)""" monkeypatch.setenv("RELVAR", "./spam/eggs") - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /bar: ./foo/bar # simple form, dotted path - /scp: # complex form - hostpath: ./snap/crackle/pop - /relvar: $RELVAR # simple form, relative path in variable - """ - ) + SCUBA_YML.write_text( + r""" + image: na + volumes: + /bar: ./foo/bar # simple form, dotted path + /scp: # complex form + hostpath: ./snap/crackle/pop + /relvar: $RELVAR # simple form, relative path in variable + """ + ) # Make a subdirectory and cd into it subdir = Path("way/down/here") @@ -1064,7 +923,9 @@ def test_volumes_hostpath_rel(self, monkeypatch, in_tmp_path) -> None: assert_vol(config.volumes, "/scp", in_tmp_path / "snap" / "crackle" / "pop") assert_vol(config.volumes, "/relvar", in_tmp_path / "spam" / "eggs") - def test_volumes_hostpath_rel_above(self, monkeypatch, in_tmp_path) -> None: + def test_hostpath_rel_above( + self, monkeypatch: pytest.MonkeyPatch, in_tmp_path: Path + ) -> None: """volume hostpath can be relative, above scuba_root (top dir)""" # Directory structure: # @@ -1081,14 +942,13 @@ def test_volumes_hostpath_rel_above(self, monkeypatch, in_tmp_path) -> None: monkeypatch.chdir(project_dir) # Now put .scuba.yml here - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: ../../../foo_up_here - """ - ) + SCUBA_YML.write_text( + r""" + image: na + volumes: + /foo: ../../../foo_up_here + """ + ) # Locate the config found_topdir, found_rel, config = scuba.config.find_config() @@ -1098,63 +958,55 @@ def test_volumes_hostpath_rel_above(self, monkeypatch, in_tmp_path) -> None: assert config.volumes is not None assert_vol(config.volumes, "/foo", in_tmp_path / "foo_up_here") - def test_volumes_hostpath_rel_requires_dot_complex( - self, monkeypatch, in_tmp_path - ) -> None: + def test_hostpath_rel_requires_dot_complex(self) -> None: """relaitve volume hostpath (complex form) must start with ./ or ../""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /one: - hostpath: foo # Forbidden - """ - ) - self._invalid_config("Relative path must start with ./ or ../") + invalid_config( + config_text=r""" + image: na + volumes: + /one: + hostpath: foo # Forbidden + """, + error_match="Relative path must start with ./ or ../", + ) - def test_volumes_hostpath_rel_in_env(self, monkeypatch, in_tmp_path) -> None: + def test_hostpath_rel_in_env( + self, monkeypatch: pytest.MonkeyPatch, in_tmp_path: Path + ) -> None: """volume definitions can contain environment variables, including relative path portions""" monkeypatch.setenv("PREFIX", "./") - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: ${PREFIX}/foo - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + /foo: ${PREFIX}/foo + """ + ) vols = config.volumes assert vols is not None assert len(vols) == 1 assert_vol(vols, "/foo", in_tmp_path / "foo") - def test_volumes_contpath_rel(self, monkeypatch, in_tmp_path) -> None: - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - foo: /what/now - """ - ) - self._invalid_config("Relative path not allowed: foo") + def test_contpath_rel(self) -> None: + invalid_config( + config_text=r""" + image: na + volumes: + foo: /what/now + """, + error_match="Relative path not allowed: foo", + ) - def test_volumes_simple_named_volume(self) -> None: + def test_simple_named_volume(self) -> None: """volumes simple form can specify a named volume""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: foo-volume - """ - ) - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + /foo: foo-volume + """ + ) assert config.volumes is not None assert len(config.volumes) == 1 vol = config.volumes[Path("/foo")] @@ -1163,20 +1015,16 @@ def test_volumes_simple_named_volume(self) -> None: assert_paths_equal(vol.container_path, "/foo") assert vol.volume_name == "foo-volume" - def test_volumes_simple_named_volume_env(self, monkeypatch) -> None: + def test_simple_named_volume_env(self, monkeypatch: pytest.MonkeyPatch) -> None: """volumes simple form can specify a named volume via env var""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: $FOO_VOLUME - """ - ) - monkeypatch.setenv("FOO_VOLUME", "foo-volume") - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + /foo: $FOO_VOLUME + """ + ) assert config.volumes is not None assert len(config.volumes) == 1 vol = config.volumes[Path("/foo")] @@ -1185,21 +1033,17 @@ def test_volumes_simple_named_volume_env(self, monkeypatch) -> None: assert_paths_equal(vol.container_path, "/foo") assert vol.volume_name == "foo-volume" - def test_volumes_complex_named_volume_env(self, monkeypatch) -> None: + def test_complex_named_volume_env(self, monkeypatch: pytest.MonkeyPatch) -> None: """volumes complex form can specify a named volume via env var""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: - name: $FOO_VOLUME - """ - ) - monkeypatch.setenv("FOO_VOLUME", "foo-volume") - - config = load_config() + config = load_config( + config_text=r""" + image: na + volumes: + /foo: + name: $FOO_VOLUME + """ + ) assert config.volumes is not None assert len(config.volumes) == 1 vol = config.volumes[Path("/foo")] @@ -1208,58 +1052,50 @@ def test_volumes_complex_named_volume_env(self, monkeypatch) -> None: assert_paths_equal(vol.container_path, "/foo") assert vol.volume_name == "foo-volume" - def test_volumes_complex_named_volume_env_unset(self) -> None: + def test_complex_named_volume_env_unset(self) -> None: """volumes complex form fails on unset env var""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: - name: $FOO_VOLUME - """ - ) - self._invalid_config("Unset environment variable") + invalid_config( + config_text=r""" + image: na + volumes: + /foo: + name: $FOO_VOLUME + """, + error_match="Unset environment variable", + ) - def test_volumes_complex_invalid_hostpath(self) -> None: + def test_complex_invalid_hostpath(self) -> None: """volumes complex form cannot specify an invalid hostpath""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: - hostpath: foo-volume - """ - ) - self._invalid_config("Relative path must start with ./ or ../") + invalid_config( + config_text=r""" + image: na + volumes: + /foo: + hostpath: foo-volume + """, + error_match="Relative path must start with ./ or ../", + ) - def test_volumes_complex_hostpath_and_name(self) -> None: + def test_complex_hostpath_and_name(self) -> None: """volumes complex form cannot specify a named volume and hostpath""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: - hostpath: /bar - name: foo-volume - """ - ) - self._invalid_config( - "Volume /foo must have exactly one of 'hostpath' or 'name' subkey" + invalid_config( + config_text=r""" + image: na + volumes: + /foo: + hostpath: /bar + name: foo-volume + """, + error_match="Volume /foo must have exactly one of 'hostpath' or 'name' subkey", ) - def test_volumes_complex_empty(self) -> None: + def test_complex_empty(self) -> None: """volumes complex form cannot be empty""" - with open(".scuba.yml", "w") as f: - f.write( - r""" - image: na - volumes: - /foo: - """ - ) - self._invalid_config( - "Volume /foo must have exactly one of 'hostpath' or 'name' subkey" + invalid_config( + config_text=r""" + image: na + volumes: + /foo: + """, + error_match="Volume /foo must have exactly one of 'hostpath' or 'name' subkey", ) diff --git a/tests/test_dockerutil.py b/tests/test_dockerutil.py index dd418d4..2dac1e1 100644 --- a/tests/test_dockerutil.py +++ b/tests/test_dockerutil.py @@ -1,4 +1,3 @@ -# coding=utf-8 from pathlib import Path import pytest import subprocess @@ -24,7 +23,7 @@ def test_get_image_command_bad_image() -> None: def test_get_image_no_docker() -> None: """get_image_command raises an exception if docker is not installed""" - def mocked_run(args, real_run=subprocess.run, **kw): + def mocked_run(args, real_run=subprocess.run, **kw): # type: ignore[no-untyped-def] assert args[0] == "docker" args[0] = "dockerZZZZ" return real_run(args, **kw) @@ -34,8 +33,8 @@ def mocked_run(args, real_run=subprocess.run, **kw): uut.get_image_command("n/a") -def _test_get_images(stdout, returncode=0) -> Sequence[str]: - def mocked_run(*args, **kwargs): +def _test_get_images(stdout: str, returncode: int = 0) -> Sequence[str]: + def mocked_run(*args, **kwargs): # type: ignore[no-untyped-def] mock_obj = mock.MagicMock() mock_obj.returncode = returncode mock_obj.stdout = stdout diff --git a/tests/test_main.py b/tests/test_main.py index 7b242fb..455c52a 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -9,6 +9,8 @@ import subprocess import sys from tempfile import TemporaryFile, NamedTemporaryFile +from textwrap import dedent +from typing import cast, IO, List, Optional, Sequence, TextIO, Tuple from unittest import mock import warnings @@ -22,179 +24,177 @@ InTempDir, make_executable, skipUnlessTty, + PseudoTTY, ) +ScubaResult = Tuple[str, str] + +SCUBA_YML = Path(".scuba.yml") SCUBAINIT_EXIT_FAIL = 99 -@pytest.mark.usefixtures("in_tmp_path") -class TestMainBase: - def run_scuba(self, args, exp_retval=0, mock_isatty=False, stdin=None): - """Run scuba, checking its return value - - Returns scuba/docker stdout data. - """ - - # Capture both scuba and docker's stdout/stderr, - # just as the user would see it. - with TemporaryFile(prefix="scubatest-stdout", mode="w+t") as stdout: - with TemporaryFile(prefix="scubatest-stderr", mode="w+t") as stderr: - if mock_isatty: - stdout = PseudoTTY(stdout) - stderr = PseudoTTY(stderr) - - old_stdin = sys.stdin - old_stdout = sys.stdout - old_stderr = sys.stderr - - if stdin is None: - sys.stdin = open(os.devnull, "w") - else: - sys.stdin = stdin - sys.stdout = stdout - sys.stderr = stderr +def write_script(path: Path, text: str) -> None: + path.write_text(dedent(text) + "\n") + make_executable(path) + + +def run_scuba( + args: List[str], + *, + expect_return: int = 0, + mock_isatty: bool = False, + stdin: Optional[IO[str]] = None, +) -> ScubaResult: + """Run scuba, checking its return value + + Returns scuba/docker stdout data. + """ + # Capture both scuba and docker's stdout/stderr, + # just as the user would see it. + with TemporaryFile(prefix="scubatest-stdout", mode="w+t") as stdout: + with TemporaryFile(prefix="scubatest-stderr", mode="w+t") as stderr: + if mock_isatty: + stdout = PseudoTTY(stdout) # type: ignore[assignment] + stderr = PseudoTTY(stderr) # type: ignore[assignment] + + old_stdin = sys.stdin + old_stdout = sys.stdout + old_stderr = sys.stderr + + if stdin is None: + sys.stdin = open(os.devnull, "w") + else: + sys.stdin = cast(TextIO, stdin) + sys.stdout = cast(TextIO, stdout) + sys.stderr = cast(TextIO, stderr) + + try: + """ + Call scuba's main(), and expect it to either exit() + with a given return code, or return (implying an exit + status of 0). + """ try: - """ - Call scuba's main(), and expect it to either exit() - with a given return code, or return (implying an exit - status of 0). - """ - try: - main.main(argv=args) - except SystemExit as sysexit: - retcode = sysexit.code - else: - retcode = 0 + main.main(argv=args) + except SystemExit as sysexit: + retcode = sysexit.code + else: + retcode = 0 + + stdout.seek(0) + stderr.seek(0) - stdout.seek(0) - stderr.seek(0) + stdout_data = stdout.read() + stderr_data = stderr.read() - stdout_data = stdout.read() - stderr_data = stderr.read() + logging.info("scuba stdout:\n" + stdout_data) + logging.info("scuba stderr:\n" + stderr_data) - logging.info("scuba stdout:\n" + stdout_data) - logging.info("scuba stderr:\n" + stderr_data) + # Verify the return value was as expected + assert expect_return == retcode - # Verify the return value was as expected - assert exp_retval == retcode + return stdout_data, stderr_data - return stdout_data, stderr_data + finally: + sys.stdin = old_stdin + sys.stdout = old_stdout + sys.stderr = old_stderr - finally: - sys.stdin = old_stdin - sys.stdout = old_stdout - sys.stderr = old_stderr +@pytest.mark.usefixtures("in_tmp_path") +class MainTest: + pass -class TestMain(TestMainBase): + +class TestMainBasic(MainTest): def test_basic(self) -> None: """Verify basic scuba functionality""" - - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") args = ["/bin/echo", "-n", "my output"] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish("my output", out) def test_no_cmd(self) -> None: """Verify scuba works with no given command""" + SCUBA_YML.write_text("image: scuba/hello") - with open(".scuba.yml", "w") as f: - f.write("image: scuba/hello\n") - - out, _ = self.run_scuba([]) + out, _ = run_scuba([]) assert_str_equalish(out, "Hello world") def test_no_image_cmd(self) -> None: """Verify scuba gracefully handles an image with no Cmd and no user command""" - - with open(".scuba.yml", "w") as f: - f.write("image: scuba/scratch\n") + SCUBA_YML.write_text("image: scuba/scratch") # ScubaError -> exit(128) - out, _ = self.run_scuba([], 128) + out, _ = run_scuba([], expect_return=128) def test_handle_get_image_command_error(self) -> None: """Verify scuba handles a get_image_command error""" + SCUBA_YML.write_text("image: {DOCKER_IMAGE}") - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") - - def mocked_gic(*args, **kw): + def mocked_gic(image: str) -> Optional[Sequence[str]]: raise scuba.dockerutil.DockerError("mock error") # http://alexmarandon.com/articles/python_mock_gotchas/#patching-in-the-wrong-place # http://www.voidspace.org.uk/python/mock/patch.html#where-to-patch with mock.patch("scuba.scuba.get_image_command", side_effect=mocked_gic): # DockerError -> exit(128) - self.run_scuba([], 128) + run_scuba([], expect_return=128) def test_config_error(self) -> None: """Verify config errors are handled gracefully""" - - with open(".scuba.yml", "w") as f: - f.write("invalid_key: is no good\n") + SCUBA_YML.write_text("invalid_key: is no good") # ConfigError -> exit(128) - self.run_scuba([], 128) + run_scuba([], expect_return=128) def test_multiline_alias_no_args_error(self) -> None: """Verify config errors from passing arguments to multi-line alias are caught""" - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - aliases: - multi: - script: - - echo multi - - echo line - - echo alias - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + aliases: + multi: + script: + - echo multi + - echo line + - echo alias + """ + ) # ConfigError -> exit(128) - self.run_scuba(["multi", "with", "args"], 128) + run_scuba(["multi", "with", "args"], expect_return=128) def test_version(self) -> None: """Verify scuba prints its version for -v""" - out, err = self.run_scuba(["-v"]) + out, err = run_scuba(["-v"]) name, ver = out.split() assert name == "scuba" assert ver == scuba.__version__ - def test_no_docker(self) -> None: + def test_no_docker(self, monkeypatch: pytest.MonkeyPatch) -> None: """Verify scuba gracefully handles docker not being installed""" - - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") args = ["/bin/echo", "-n", "my output"] - old_PATH = os.environ["PATH"] - os.environ["PATH"] = "" - - try: - _, err = self.run_scuba(args, 2) - finally: - os.environ["PATH"] = old_PATH + monkeypatch.setenv("PATH", "") + _, err = run_scuba(args, expect_return=2) @mock.patch("subprocess.call") - def test_dry_run(self, subproc_call_mock): + def test_dry_run(self, subproc_call_mock: mock.Mock) -> None: + print(f"subproc_call_mock is a {type(subproc_call_mock)}") """Verify scuba handles --dry-run and --verbose""" - - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") args = ["--dry-run", "--verbose", "/bin/false"] - - _, err = self.run_scuba(args) + _, err = run_scuba(args) assert not subproc_call_mock.called @@ -202,50 +202,57 @@ def test_dry_run(self, subproc_call_mock): def test_args(self) -> None: """Verify scuba handles cmdline args""" - - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") - - with open("test.sh", "w") as f: - f.write("#!/bin/sh\n") - f.write('for a in "$@"; do echo $a; done\n') - make_executable("test.sh") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") + test_script = Path("test.sh") + + write_script( + test_script, + """\ + #!/bin/sh + for a in "$@"; do echo $a; done + """, + ) lines = ["here", "are", "some args"] - out, _ = self.run_scuba(["./test.sh"] + lines) + out, _ = run_scuba([f"./{test_script}"] + lines) assert_seq_equal(out.splitlines(), lines) def test_created_file_ownership(self) -> None: """Verify files created under scuba have correct ownership""" - - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") - + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") filename = "newfile.txt" - self.run_scuba(["/bin/touch", filename]) + run_scuba(["/bin/touch", filename]) st = os.stat(filename) assert st.st_uid == os.getuid() assert st.st_gid == os.getgid() + +class TestMainStdinStdout(MainTest): + CHECK_TTY_SCRIPT = Path("check_tty.sh") + def _setup_test_tty(self) -> None: - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + assert sys.stdin.isatty() + + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") - with open("check_tty.sh", "w") as f: - f.write("#!/bin/sh\n") - f.write('if [ -t 1 ]; then echo "isatty"; else echo "notatty"; fi\n') - make_executable("check_tty.sh") + write_script( + self.CHECK_TTY_SCRIPT, + """\ + #!/bin/sh + if [ -t 1 ]; then echo "isatty"; else echo "notatty"; fi + """, + ) @skipUnlessTty() def test_with_tty(self) -> None: """Verify docker allocates tty if stdout is a tty.""" self._setup_test_tty() - out, _ = self.run_scuba(["./check_tty.sh"], mock_isatty=True) + out, _ = run_scuba([f"./{self.CHECK_TTY_SCRIPT}"], mock_isatty=True) assert_str_equalish(out, "isatty") @@ -254,50 +261,59 @@ def test_without_tty(self) -> None: """Verify docker doesn't allocate tty if stdout is not a tty.""" self._setup_test_tty() - out, _ = self.run_scuba(["./check_tty.sh"]) + out, _ = run_scuba([f"./{self.CHECK_TTY_SCRIPT}"]) assert_str_equalish(out, "notatty") def test_redirect_stdin(self) -> None: """Verify stdin redirection works""" - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") test_str = "hello world" with TemporaryFile(prefix="scubatest-stdin", mode="w+t") as stdin: stdin.write(test_str) stdin.seek(0) - out, _ = self.run_scuba(["cat"], stdin=stdin) + out, _ = run_scuba(["cat"], stdin=stdin) assert_str_equalish(out, test_str) + +class TestMainUser(MainTest): def _test_user( self, - expected_uid, - expected_username, - expected_gid, - expected_groupname, - scuba_args=[], - ): - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + expected_uid: int, + expected_username: str, + expected_gid: int, + expected_groupname: str, + scuba_args: List[str] = [], + ) -> None: + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") args = scuba_args + [ "/bin/sh", "-c", "echo $(id -u) $(id -un) $(id -g) $(id -gn)", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) - uid, username, gid, groupname = out.split() - uid = int(uid) - gid = int(gid) + uid_str, username, gid_str, groupname = out.split() + uid = int(uid_str) + gid = int(gid_str) assert uid == expected_uid assert username == expected_username assert gid == expected_gid assert groupname == expected_groupname + def _test_user_expect_root(self, scuba_args: List[str] = []) -> None: + return self._test_user( + expected_uid=0, + expected_username="root", + expected_gid=0, + expected_groupname="root", + scuba_args=scuba_args, + ) + def test_user_scubauser(self) -> None: """Verify scuba runs container as the current (host) uid/gid""" self._test_user( @@ -307,19 +323,9 @@ def test_user_scubauser(self) -> None: expected_groupname=getgrgid(os.getgid()).gr_name, ) - EXPECT_ROOT = dict( - expected_uid=0, - expected_username="root", - expected_gid=0, - expected_groupname="root", - ) - def test_user_root(self) -> None: """Verify scuba -r runs container as root""" - self._test_user( - **self.EXPECT_ROOT, - scuba_args=["-r"], - ) + self._test_user_expect_root(scuba_args=["-r"]) def test_user_run_as_root(self) -> None: '''Verify running scuba as root is identical to "scuba -r"''' @@ -327,25 +333,24 @@ def test_user_run_as_root(self) -> None: with mock.patch("os.getuid", return_value=0) as getuid_mock, mock.patch( "os.getgid", return_value=0 ) as getgid_mock: - self._test_user(**self.EXPECT_ROOT) + self._test_user_expect_root() assert getuid_mock.called assert getgid_mock.called def test_user_root_alias(self) -> None: """Verify that aliases can set whether the container is run as root""" - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - aliases: - root_test: - root: true - script: - - echo $(id -u) $(id -un) $(id -g) $(id -gn) - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + aliases: + root_test: + root: true + script: + - echo $(id -u) $(id -un) $(id -g) $(id -gn) + """ + ) - out, _ = self.run_scuba(["root_test"]) + out, _ = run_scuba(["root_test"]) uid, username, gid, groupname = out.split() assert int(uid) == 0 @@ -355,19 +360,18 @@ def test_user_root_alias(self) -> None: # No one should ever specify 'root: false' in an alias, but Scuba should behave # correctly if they do - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - aliases: - no_root_test: - root: false - script: - - echo $(id -u) $(id -un) $(id -g) $(id -gn) - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + aliases: + no_root_test: + root: false + script: + - echo $(id -u) $(id -un) $(id -g) $(id -gn) + """ + ) - out, _ = self.run_scuba(["no_root_test"]) + out, _ = run_scuba(["no_root_test"]) uid, username, gid, groupname = out.split() assert int(uid) == os.getuid() @@ -375,16 +379,17 @@ def test_user_root_alias(self) -> None: assert int(gid) == os.getgid() assert groupname == getgrgid(os.getgid()).gr_name - def _test_home_writable(self, scuba_args=[]): - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + +class TestMainHomedir(MainTest): + def _test_home_writable(self, scuba_args: List[str] = []) -> None: + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") args = scuba_args + [ "/bin/sh", "-c", "echo success >> ~/testfile; cat ~/testfile", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish(out, "success") @@ -409,11 +414,11 @@ def test_home_writable_root(self) -> None: """Verify root has a writable homedir""" self._test_home_writable(["-r"]) + +class TestMainDockerArgs(MainTest): def test_arbitrary_docker_args(self) -> None: """Verify -d successfully passes arbitrary docker arguments""" - - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") data = "Lorem ipsum dolor sit amet" data_path = "/lorem/ipsum" @@ -427,7 +432,7 @@ def test_arbitrary_docker_args(self) -> None: "cat", data_path, ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish(out, data) @@ -438,121 +443,149 @@ def test_arbitrary_docker_args_merge_config(self) -> None: expfiles = set() tgtdir = "/tgtdir" - def mount_dummy(name): + def mount_dummy(name: str) -> str: assert name not in expfiles expfiles.add(name) return f'-v "{dummy.absolute()}:{tgtdir}/{name}"\n' - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") - f.write("docker_args: " + mount_dummy("one")) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + docker_args: {mount_dummy('one')} + """ + ) args = [ "-d=" + mount_dummy("two"), "ls", tgtdir, ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) files = set(out.splitlines()) assert files == expfiles + +class TestMainAliasScripts(MainTest): + def test_complex_commands_in_alias(self) -> None: + """Verify complex commands can be used in alias scripts""" + test_dir = Path("foo") + test_file = test_dir / "bar.txt" + test_string = "Hello world" + + test_dir.mkdir() + test_file.write_text(test_string) + + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + aliases: + alias1: + script: + - cd {test_dir} && cat {test_file.name} + """ + ) + + out, _ = run_scuba(["alias1"]) + assert_str_equalish(test_string, out) + def test_nested_sript(self) -> None: """Verify nested scripts works""" - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") - f.write("aliases:\n") - f.write(" foo:\n") - f.write(" script:\n") - f.write(' - echo "This"\n') - f.write(' - - echo "list"\n') - f.write(' - echo "is"\n') - f.write(' - echo "nested"\n') - f.write(' - - echo "kinda"\n') - f.write(' - echo "crazy"\n') + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + aliases: + foo: + script: + - echo "This" + - - echo "list" + - echo "is" + - echo "nested" + - - echo "kinda" + - echo "crazy" + """ + ) test_str = "This list is nested kinda crazy" - out, _ = self.run_scuba(["foo"]) + out, _ = run_scuba(["foo"]) out = out.replace("\n", " ") assert_str_equalish(out, test_str) - ############################################################################ - # Entrypoint +class TestMainEntrypoint(MainTest): def test_image_entrypoint(self) -> None: """Verify scuba doesn't interfere with the configured image ENTRYPOINT""" + SCUBA_YML.write_text("image: scuba/entrypoint-test") - with open(".scuba.yml", "w") as f: - f.write("image: scuba/entrypoint-test") - - out, _ = self.run_scuba(["cat", "entrypoint_works.txt"]) + out, _ = run_scuba(["cat", "entrypoint_works.txt"]) assert_str_equalish("success", out) def test_image_entrypoint_multiline(self) -> None: """Verify entrypoints are handled correctly with multi-line scripts""" - with open(".scuba.yml", "w") as f: - f.write( - """ - image: scuba/entrypoint-test - aliases: - testalias: - script: - - cat entrypoint_works.txt - - echo $ENTRYPOINT_WORKS - """ - ) + SCUBA_YML.write_text( + """ + image: scuba/entrypoint-test + aliases: + testalias: + script: + - cat entrypoint_works.txt + - echo $ENTRYPOINT_WORKS + """ + ) - out, _ = self.run_scuba(["testalias"]) + out, _ = run_scuba(["testalias"]) assert_str_equalish("\n".join(["success"] * 2), out) def test_entrypoint_override(self) -> None: """Verify --entrypoint override works""" - with open(".scuba.yml", "w") as f: - f.write( - """ - image: scuba/entrypoint-test - aliases: - testalias: - script: - - echo $ENTRYPOINT_WORKS - """ - ) + SCUBA_YML.write_text( + """ + image: scuba/entrypoint-test + aliases: + testalias: + script: + - echo $ENTRYPOINT_WORKS + """ + ) + test_script = Path("new.sh") test_str = "This is output from the overridden entrypoint" - with open("new.sh", "w") as f: - f.write("#!/bin/sh\n") - f.write(f'echo "{test_str}"\n') - make_executable("new.sh") + write_script( + test_script, + f"""\ + #!/bin/sh + echo "{test_str}" + """, + ) args = [ "--entrypoint", - os.path.abspath("new.sh"), + str(test_script.absolute()), "true", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish(test_str, out) def test_entrypoint_override_none(self) -> None: """Verify --entrypoint override (to nothing) works""" - with open(".scuba.yml", "w") as f: - f.write( - """ - image: scuba/entrypoint-test - aliases: - testalias: - script: - - echo $ENTRYPOINT_WORKS - """ - ) + SCUBA_YML.write_text( + """ + image: scuba/entrypoint-test + aliases: + testalias: + script: + - echo $ENTRYPOINT_WORKS + """ + ) args = [ "--entrypoint", "", "testalias", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) # Verify that ENTRYPOINT_WORKS was not set by the entrypoint # (because it didn't run) @@ -560,59 +593,57 @@ def test_entrypoint_override_none(self) -> None: def test_yaml_entrypoint_override(self) -> None: """Verify entrypoint in .scuba.yml works""" - with open(".scuba.yml", "w") as f: - f.write( - """ - image: scuba/entrypoint-test - entrypoint: "./new.sh" - """ - ) - + test_script = Path("new.sh") test_str = "This is output from the overridden entrypoint" - with open("new.sh", "w") as f: - f.write("#!/bin/sh\n") - f.write(f'echo "{test_str}"\n') - make_executable("new.sh") + write_script( + test_script, + f"""\ + #!/bin/sh + echo "{test_str}" + """, + ) - args = [ - "true", - ] - out, _ = self.run_scuba(args) + SCUBA_YML.write_text( + f""" + image: scuba/entrypoint-test + entrypoint: "./{test_script}" + """ + ) + + out, _ = run_scuba(["true"]) assert_str_equalish(test_str, out) def test_yaml_entrypoint_override_none(self) -> None: """Verify "none" entrypoint in .scuba.yml works""" - with open(".scuba.yml", "w") as f: - f.write( - """ - image: scuba/entrypoint-test - entrypoint: - aliases: - testalias: - script: - - echo $ENTRYPOINT_WORKS - """ - ) + SCUBA_YML.write_text( + """ + image: scuba/entrypoint-test + entrypoint: + aliases: + testalias: + script: + - echo $ENTRYPOINT_WORKS + """ + ) args = [ "testalias", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) # Verify that ENTRYPOINT_WORKS was not set by the entrypoint # (because it didn't run) assert_str_equalish("", out) - ############################################################################ - # Image override +class TestMainImageOverride(MainTest): def test_image_override(self) -> None: """Verify --image works""" - - with open(".scuba.yml", "w") as f: + SCUBA_YML.write_text( # This image does not exist - f.write("image: scuba/notheredoesnotexistbb7e344f9722\n") + "image: scuba/notheredoesnotexistbb7e344f9722" + ) args = [ "--image", @@ -620,33 +651,31 @@ def test_image_override(self) -> None: "echo", "success", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish("success", out) def test_image_override_with_alias(self) -> None: """Verify --image works with aliases""" - - with open(".scuba.yml", "w") as f: + SCUBA_YML.write_text( # These images do not exist - f.write( - """ - image: scuba/notheredoesnotexistbb7e344f9722 - aliases: - testalias: - image: scuba/notheredoesnotexist765205d09dea - script: - - echo multi - - echo line - - echo alias - """ - ) + """ + image: scuba/notheredoesnotexistbb7e344f9722 + aliases: + testalias: + image: scuba/notheredoesnotexist765205d09dea + script: + - echo multi + - echo line + - echo alias + """ + ) args = [ "--image", DOCKER_IMAGE, "testalias", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish("multi\nline\nalias", out) def test_yml_not_needed_with_image_override(self) -> None: @@ -660,50 +689,42 @@ def test_yml_not_needed_with_image_override(self) -> None: "echo", "success", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish("success", out) - def test_complex_commands_in_alias(self) -> None: - """Verify complex commands can be used in alias scripts""" - test_string = "Hello world" - os.mkdir("foo") - with open("foo/bar.txt", "w") as f: - f.write(test_string) - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") - f.write("aliases:\n") - f.write(" alias1:\n") - f.write(" script:\n") - f.write(" - cd foo && cat bar.txt\n") - - out, _ = self.run_scuba(["alias1"]) - assert_str_equalish(test_string, out) - ############################################################################ - # Hooks - - def _test_one_hook(self, hookname, hookcmd, cmd, exp_retval=0): - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") - f.write("hooks:\n") - f.write(f" {hookname}: {hookcmd}\n") +class TestMainHooks(MainTest): + def _test_one_hook( + self, + hookname: str, + hookcmd: str, + cmd: str, + expect_return: int = 0, + ) -> ScubaResult: + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + hooks: + {hookname}: {hookcmd} + """ + ) args = ["/bin/sh", "-c", cmd] - return self.run_scuba(args, exp_retval=exp_retval) + return run_scuba(args, expect_return=expect_return) - def _test_hook_runs_as(self, hookname, exp_uid, exp_gid) -> None: + def _test_hook_runs_as(self, hookname: str, exp_uid: int, exp_gid: int) -> None: out, _ = self._test_one_hook( - hookname, - "echo $(id -u) $(id -g)", - "echo success", + hookname=hookname, + hookcmd="echo $(id -u) $(id -g)", + cmd="echo success", ) - out = out.splitlines() + out_lines = out.splitlines() - uid, gid = map(int, out[0].split()) + uid, gid = map(int, out_lines[0].split()) assert exp_uid == uid assert exp_gid == gid - assert_str_equalish(out[1], "success") + assert_str_equalish(out_lines[1], "success") def test_user_hook_runs_as_user(self) -> None: """Verify user hook executes as user""" @@ -719,17 +740,15 @@ def test_hook_failure_shows_correct_status(self) -> None: "root", f"exit {testval}", "dont care", - exp_retval=SCUBAINIT_EXIT_FAIL, + expect_return=SCUBAINIT_EXIT_FAIL, ) assert re.match(f"^scubainit: .* exited with status {testval}$", err) - ############################################################################ - # Environment +class TestMainEnvironment(MainTest): def test_env_var_keyval(self) -> None: """Verify -e KEY=VAL works""" - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") args = [ "-e", "KEY=VAL", @@ -737,13 +756,12 @@ def test_env_var_keyval(self) -> None: "-c", "echo $KEY", ] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish(out, "VAL") - def test_env_var_key_only(self, monkeypatch): + def test_env_var_key_only(self, monkeypatch: pytest.MonkeyPatch) -> None: """Verify -e KEY works""" - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") args = [ "-e", "KEY", @@ -752,35 +770,34 @@ def test_env_var_key_only(self, monkeypatch): "echo $KEY", ] monkeypatch.setenv("KEY", "mockedvalue") - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish(out, "mockedvalue") - def test_env_var_sources(self, monkeypatch): + def test_env_var_sources(self, monkeypatch: pytest.MonkeyPatch) -> None: """Verify scuba handles all possible environment variable sources""" - with open(".scuba.yml", "w") as f: - f.write( - rf""" - image: {DOCKER_IMAGE} + SCUBA_YML.write_text( + rf""" + image: {DOCKER_IMAGE} + environment: + FOO: Top-level + BAR: 42 + EXTERNAL_2: + aliases: + al: + script: + - echo "FOO=\"$FOO\"" + - echo "BAR=\"$BAR\"" + - echo "MORE=\"$MORE\"" + - echo "EXTERNAL_1=\"$EXTERNAL_1\"" + - echo "EXTERNAL_2=\"$EXTERNAL_2\"" + - echo "EXTERNAL_3=\"$EXTERNAL_3\"" + - echo "BAZ=\"$BAZ\"" environment: - FOO: Top-level - BAR: 42 - EXTERNAL_2: - aliases: - al: - script: - - echo "FOO=\"$FOO\"" - - echo "BAR=\"$BAR\"" - - echo "MORE=\"$MORE\"" - - echo "EXTERNAL_1=\"$EXTERNAL_1\"" - - echo "EXTERNAL_2=\"$EXTERNAL_2\"" - - echo "EXTERNAL_3=\"$EXTERNAL_3\"" - - echo "BAZ=\"$BAZ\"" - environment: - FOO: Overridden - MORE: Hello world - EXTERNAL_3: - """ - ) + FOO: Overridden + MORE: Hello world + EXTERNAL_3: + """ + ) args = [ "-e", @@ -794,7 +811,7 @@ def test_env_var_sources(self, monkeypatch): monkeypatch.setenv("EXTERNAL_2", "External value 2") monkeypatch.setenv("EXTERNAL_3", "External value 3") - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) # Convert key/pair output to dictionary result = dict(pair.split("=", 1) for pair in shlex.split(out)) @@ -809,71 +826,65 @@ def test_env_var_sources(self, monkeypatch): BAZ="From the command line", ) - def test_builtin_env__SCUBA_ROOT(self, in_tmp_path): + def test_builtin_env__SCUBA_ROOT(self, in_tmp_path: Path) -> None: """Verify SCUBA_ROOT is set in container""" - with open(".scuba.yml", "w") as f: - f.write(f"image: {DOCKER_IMAGE}\n") + SCUBA_YML.write_text(f"image: {DOCKER_IMAGE}") args = ["/bin/sh", "-c", "echo $SCUBA_ROOT"] - out, _ = self.run_scuba(args) + out, _ = run_scuba(args) assert_str_equalish(in_tmp_path, out) - ############################################################################ - # Shell Override +class TestMainShellOverride(MainTest): def test_use_top_level_shell_override(self) -> None: """Verify that the shell can be overriden at the top level""" - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - shell: /bin/bash - aliases: - check_shell: - script: readlink -f /proc/$$/exe - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + shell: /bin/bash + aliases: + check_shell: + script: readlink -f /proc/$$/exe + """ + ) - out, _ = self.run_scuba(["check_shell"]) + out, _ = run_scuba(["check_shell"]) # If we failed to override, the shebang would be #!/bin/sh assert_str_equalish("/bin/bash", out) def test_alias_level_shell_override(self) -> None: """Verify that the shell can be overriden at the alias level without affecting other aliases""" - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - aliases: - shell_override: - shell: /bin/bash - script: readlink -f /proc/$$/exe - default_shell: - script: readlink -f /proc/$$/exe - """ - ) - out, _ = self.run_scuba(["shell_override"]) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + aliases: + shell_override: + shell: /bin/bash + script: readlink -f /proc/$$/exe + default_shell: + script: readlink -f /proc/$$/exe + """ + ) + out, _ = run_scuba(["shell_override"]) assert_str_equalish("/bin/bash", out) - out, _ = self.run_scuba(["default_shell"]) + out, _ = run_scuba(["default_shell"]) # The way that we check the shell uses the resolved symlink of /bin/sh, # which is /bin/dash on Debian assert out.strip() in ["/bin/sh", "/bin/dash"] def test_cli_shell_override(self) -> None: """Verify that the shell can be overriden by the CLI""" - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - aliases: - default_shell: - script: readlink -f /proc/$$/exe - """ - ) - - out, _ = self.run_scuba(["--shell", "/bin/bash", "default_shell"]) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + aliases: + default_shell: + script: readlink -f /proc/$$/exe + """ + ) + out, _ = run_scuba(["--shell", "/bin/bash", "default_shell"]) assert_str_equalish("/bin/bash", out) def test_shell_override_precedence(self) -> None: @@ -882,50 +893,46 @@ def test_shell_override_precedence(self) -> None: # Top-level SCUBA_YML shell << alias-level SCUBA_YML shell << CLI-specified shell # Test top-level << alias-level - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - shell: /bin/this_does_not_exist - aliases: - shell_override: - shell: /bin/bash - script: readlink -f /proc/$$/exe - """ - ) - out, _ = self.run_scuba(["shell_override"]) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + shell: /bin/this_does_not_exist + aliases: + shell_override: + shell: /bin/bash + script: readlink -f /proc/$$/exe + """ + ) + out, _ = run_scuba(["shell_override"]) assert_str_equalish("/bin/bash", out) # Test alias-level << CLI - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - aliases: - shell_overridden: - shell: /bin/this_is_not_a_real_shell - script: readlink -f /proc/$$/exe - """ - ) - out, _ = self.run_scuba(["--shell", "/bin/bash", "shell_overridden"]) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + aliases: + shell_overridden: + shell: /bin/this_is_not_a_real_shell + script: readlink -f /proc/$$/exe + """ + ) + out, _ = run_scuba(["--shell", "/bin/bash", "shell_overridden"]) assert_str_equalish("/bin/bash", out) # Test top-level << CLI - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - shell: /bin/this_is_not_a_real_shell - aliases: - shell_check: readlink -f /proc/$$/exe - """ - ) - out, _ = self.run_scuba(["--shell", "/bin/bash", "shell_check"]) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + shell: /bin/this_is_not_a_real_shell + aliases: + shell_check: readlink -f /proc/$$/exe + """ + ) + out, _ = run_scuba(["--shell", "/bin/bash", "shell_check"]) assert_str_equalish("/bin/bash", out) - ############################################################################ - # Volumes +class TestMainVolumes(MainTest): def test_volumes_basic(self) -> None: """Verify volumes can be added at top-level and alias""" @@ -938,23 +945,21 @@ def test_volumes_basic(self) -> None: aliasdata.mkdir() (aliasdata / "thing").write_text("from the alias\n") - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + volumes: + /topdata: {topdata.absolute()} + aliases: + doit: volumes: - /topdata: {topdata.absolute()} - aliases: - doit: - volumes: - /aliasdata: {aliasdata.absolute()} - script: "cat /topdata/thing /aliasdata/thing" - """ - ) + /aliasdata: {aliasdata.absolute()} + script: "cat /topdata/thing /aliasdata/thing" + """ + ) - out, _ = self.run_scuba(["doit"]) - out = out.splitlines() - assert out == ["from the top", "from the alias"] + out, _ = run_scuba(["doit"]) + assert out.splitlines() == ["from the top", "from the alias"] def test_volumes_alias_override(self) -> None: """Verify volumes can be overridden by an alias""" @@ -968,29 +973,26 @@ def test_volumes_alias_override(self) -> None: aliasdata.mkdir() (aliasdata / "thing").write_text("from the alias\n") - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + volumes: + /data: {topdata.absolute()} + aliases: + doit: volumes: - /data: {topdata.absolute()} - aliases: - doit: - volumes: - /data: {aliasdata.absolute()} - script: "cat /data/thing" - """ - ) + /data: {aliasdata.absolute()} + script: "cat /data/thing" + """ + ) # Run a non-alias command - out, _ = self.run_scuba(["cat", "/data/thing"]) - out = out.splitlines() - assert out == ["from the top"] + out, _ = run_scuba(["cat", "/data/thing"]) + assert out.splitlines() == ["from the top"] # Run the alias - out, _ = self.run_scuba(["doit"]) - out = out.splitlines() - assert out == ["from the alias"] + out, _ = run_scuba(["doit"]) + assert out.splitlines() == ["from the alias"] def test_volumes_host_path_create(self) -> None: """Missing host paths should be created before starting Docker""" @@ -998,16 +1000,15 @@ def test_volumes_host_path_create(self) -> None: userdir = Path("./user") testfile = userdir / "test.txt" - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - volumes: - /userdir: {userdir.absolute()} - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + volumes: + /userdir: {userdir.absolute()} + """ + ) - self.run_scuba(["touch", "/userdir/test.txt"]) + run_scuba(["touch", "/userdir/test.txt"]) assert testfile.exists(), "Test file was not created" @@ -1024,23 +1025,22 @@ def test_volumes_host_path_permissions(self) -> None: rootdir.mkdir() - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - volumes: - /userdir: {userdir.absolute()} - aliases: - doit: - root: true - script: "touch /userdir/test.txt" - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + volumes: + /userdir: {userdir.absolute()} + aliases: + doit: + root: true + script: "touch /userdir/test.txt" + """ + ) try: # Prevent current user from creating directory rootdir.chmod(0o555) - self.run_scuba(["doit"]) + run_scuba(["doit"]) finally: # Restore permissions to allow deletion rootdir.chmod(0o755) @@ -1061,16 +1061,15 @@ def test_volumes_host_path_failure(self) -> None: # rootdir is not a dir, it's a file rootdir.write_text("lied about the dir") - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - volumes: - /userdir: {userdir.absolute()} - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + volumes: + /userdir: {userdir.absolute()} + """ + ) - self.run_scuba(["touch", "/userdir/test.txt"], 128) + run_scuba(["touch", "/userdir/test.txt"], expect_return=128) def test_volumes_host_path_rel(self) -> None: """Volume host paths can be relative""" @@ -1082,21 +1081,20 @@ def test_volumes_host_path_rel(self) -> None: test_message = "Relative paths work" (userdir / "test.txt").write_text(test_message) - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - volumes: - /userdir: ./{userdir} - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + volumes: + /userdir: ./{userdir} + """ + ) # Invoke scuba from a different subdir, for good measure. otherdir = Path("way/down/here") otherdir.mkdir(parents=True) os.chdir(otherdir) - out, _ = self.run_scuba(["cat", "/userdir/test.txt"]) + out, _ = run_scuba(["cat", "/userdir/test.txt"]) assert out == test_message def test_volumes_hostpath_rel_above(self) -> None: @@ -1124,20 +1122,19 @@ def test_volumes_hostpath_rel_above(self) -> None: # Change to the project subdir and write the .scuba.yml file there. os.chdir(projdir) - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - volumes: - /userdir: ../../../{userdir} - """ - ) + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + volumes: + /userdir: ../../../{userdir} + """ + ) - out, _ = self.run_scuba(["cat", "/userdir/test.txt"]) + out, _ = run_scuba(["cat", "/userdir/test.txt"]) assert out == test_message -class TestMainNamedVolumes(TestMainBase): +class TestMainNamedVolumes(MainTest): VOLUME_NAME = "foo-volume" def _rm_volume(self) -> None: @@ -1150,32 +1147,31 @@ def _rm_volume(self) -> None: return result.check_returncode() - def setup_method(self, method) -> None: + def setup_method(self) -> None: self._rm_volume() - def teardown_method(self, method) -> None: + def teardown_method(self) -> None: self._rm_volume() def test_volumes_named(self) -> None: """Verify named volumes can be used""" - VOL_PATH = Path("/foo") - TEST_PATH = VOL_PATH / "test.txt" - TEST_STR = "it works!" - - with open(".scuba.yml", "w") as f: - f.write( - f""" - image: {DOCKER_IMAGE} - hooks: - root: chmod 777 {VOL_PATH} - volumes: - {VOL_PATH}: {self.VOLUME_NAME} - """ - ) + vol_path = Path("/foo") + test_path = vol_path / "test.txt" + test_str = "it works!" + + SCUBA_YML.write_text( + f""" + image: {DOCKER_IMAGE} + hooks: + root: chmod 777 {vol_path} + volumes: + {vol_path}: {self.VOLUME_NAME} + """ + ) # Inoke scuba once: Write a file to the named volume - self.run_scuba(["/bin/sh", "-c", f"echo {TEST_STR} > {TEST_PATH}"]) + run_scuba(["/bin/sh", "-c", f"echo {test_str} > {test_path}"]) # Invoke scuba again: Verify the file is still there - out, _ = self.run_scuba(["/bin/sh", "-c", f"cat {TEST_PATH}"]) - assert_str_equalish(out, TEST_STR) + out, _ = run_scuba(["/bin/sh", "-c", f"cat {test_path}"]) + assert_str_equalish(out, test_str) diff --git a/tests/test_utils.py b/tests/test_utils.py index eef34f0..f9222b5 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -54,7 +54,7 @@ def test_format_cmdline() -> None: ) -def test_shell_quote_cmd(): +def test_shell_quote_cmd() -> None: args = ["foo", "bar pop", '"tee ball"'] result = scuba.utils.shell_quote_cmd(args) @@ -64,44 +64,44 @@ def test_shell_quote_cmd(): assert_seq_equal(out_args, args) -def test_parse_env_var(): +def test_parse_env_var() -> None: """parse_env_var returns a key, value pair""" result = scuba.utils.parse_env_var("KEY=value") assert result == ("KEY", "value") -def test_parse_env_var_more_equals(): +def test_parse_env_var_more_equals() -> None: """parse_env_var handles multiple equals signs""" result = scuba.utils.parse_env_var("KEY=anotherkey=value") assert result == ("KEY", "anotherkey=value") -def test_parse_env_var_no_equals(monkeypatch): +def test_parse_env_var_no_equals(monkeypatch: pytest.MonkeyPatch) -> None: """parse_env_var handles no equals and gets value from environment""" monkeypatch.setenv("KEY", "mockedvalue") result = scuba.utils.parse_env_var("KEY") assert result == ("KEY", "mockedvalue") -def test_parse_env_var_not_set(monkeypatch): +def test_parse_env_var_not_set(monkeypatch: pytest.MonkeyPatch) -> None: """parse_env_var returns an empty string if not set""" monkeypatch.delenv("NOTSET", raising=False) result = scuba.utils.parse_env_var("NOTSET") assert result == ("NOTSET", "") -def test_flatten_list__not_list(): +def test_flatten_list__not_list() -> None: with pytest.raises(ValueError): - scuba.utils.flatten_list("abc") + scuba.utils.flatten_list("abc") # type: ignore[arg-type] -def test_flatten_list__not_nested(): +def test_flatten_list__not_nested() -> None: sample = [1, 2, 3, 4] result = scuba.utils.flatten_list(sample) assert result == sample -def test_flatten_list__nested_1(): +def test_flatten_list__nested_1() -> None: sample = [ 1, [2, 3], @@ -113,7 +113,7 @@ def test_flatten_list__nested_1(): assert_seq_equal(result, exp) -def test_flatten_list__nested_many(): +def test_flatten_list__nested_many() -> None: sample = [ 1, [2, 3], @@ -127,7 +127,7 @@ def test_flatten_list__nested_many(): assert_seq_equal(result, exp) -def test_get_umask(): +def test_get_umask() -> None: testval = 0o123 # unlikely default orig = os.umask(testval) try: @@ -143,14 +143,14 @@ def test_get_umask(): os.umask(orig) -def test_writeln(): +def test_writeln() -> None: with io.StringIO() as s: scuba.utils.writeln(s, "hello") scuba.utils.writeln(s, "goodbye") assert s.getvalue() == "hello\ngoodbye\n" -def test_expand_env_vars(monkeypatch): +def test_expand_env_vars(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("MY_VAR", "my favorite variable") assert ( scuba.utils.expand_env_vars("This is $MY_VAR") == "This is my favorite variable" @@ -161,7 +161,7 @@ def test_expand_env_vars(monkeypatch): ) -def test_expand_missing_env_vars(monkeypatch): +def test_expand_missing_env_vars(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.delenv("MY_VAR", raising=False) # Verify that a KeyError is raised for unset env variables with pytest.raises(KeyError) as kerr: @@ -169,7 +169,7 @@ def test_expand_missing_env_vars(monkeypatch): assert kerr.value.args[0] == "MY_VAR" -def test_expand_env_vars_dollars(): +def test_expand_env_vars_dollars() -> None: # Verify that a ValueError is raised for bare, unescaped '$' characters with pytest.raises(ValueError): scuba.utils.expand_env_vars("Just a lonely $") diff --git a/tests/utils.py b/tests/utils.py index 8e78fea..cfb7b91 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -1,3 +1,4 @@ +from __future__ import annotations import os import sys from os.path import normpath @@ -6,13 +7,15 @@ import unittest import logging from pathlib import Path -from typing import Any, Dict, List, Sequence, Union +from typing import Any, Callable, Dict, List, Sequence, TypeVar, Optional, Union from unittest import mock from scuba.config import ScubaVolume PathStr = Union[Path, str] +_FT = TypeVar("_FT", bound=Callable[..., Any]) + def assert_seq_equal(a: Sequence, b: Sequence) -> None: assert list(a) == list(b) @@ -39,8 +42,8 @@ def assert_str_equalish(exp: Any, act: Any) -> None: def assert_vol( vols: Dict[Path, ScubaVolume], - cpath_str: str, - hpath_str: str, + cpath_str: PathStr, + hpath_str: PathStr, options: List[str] = [], ) -> None: cpath = Path(cpath_str) @@ -62,33 +65,39 @@ def make_executable(path: PathStr) -> None: # http://stackoverflow.com/a/8389373/119527 class PseudoTTY: - def __init__(self, underlying): + def __init__(self, underlying: object): self.__underlying = underlying - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return getattr(self.__underlying, name) - def isatty(self): + def isatty(self) -> bool: return True -def skipUnlessTty(): +def skipUnlessTty() -> Callable[[_FT], _FT]: return unittest.skipUnless( sys.stdin.isatty(), "Can't test docker tty if not connected to a terminal" ) class InTempDir: - def __init__(self, suffix="", prefix="tmp", dir=None, delete=True): + def __init__( + self, + suffix: str = "", + prefix: str = "tmp", + dir: Optional[PathStr] = None, + delete: bool = True, + ): self.delete = delete self.temp_path = tempfile.mkdtemp(suffix=suffix, prefix=prefix, dir=dir) - def __enter__(self): + def __enter__(self) -> InTempDir: self.orig_path = os.getcwd() os.chdir(self.temp_path) return self - def __exit__(self, *exc_info): + def __exit__(self, *exc_info: Any) -> None: # Restore the working dir and cleanup the temp one os.chdir(self.orig_path) if self.delete: