diff --git a/tests/tests_integration/base.py b/tests/tests_integration/base.py index 6364c8e2..c26c814f 100644 --- a/tests/tests_integration/base.py +++ b/tests/tests_integration/base.py @@ -1,7 +1,9 @@ import os +import subprocess import warnings import pytest +import ssh_test_utils import test_utils from datashuttle.datashuttle import DataShuttle @@ -57,3 +59,14 @@ def clean_project_name(self): test_utils.delete_project_if_it_exists(project_name) yield project_name test_utils.delete_project_if_it_exists(project_name) + + @pytest.fixture( + scope="session", + ) + def setup_ssh_container(self): + # Annoying session scope does not seem to actually work + container_name = "running_ssh_tests" + ssh_test_utils.setup_ssh_container(container_name) + yield + subprocess.run(f"docker stop {container_name}", shell=True) + subprocess.run(f"docker rm {container_name}", shell=True) diff --git a/tests/tests_integration/test_ssh_setup.py b/tests/tests_integration/test_ssh_setup.py new file mode 100644 index 00000000..0bd4b760 --- /dev/null +++ b/tests/tests_integration/test_ssh_setup.py @@ -0,0 +1,95 @@ +import pytest +import ssh_test_utils +import test_utils +from base import BaseTest + +from datashuttle.utils import ssh + +TEST_SSH = ssh_test_utils.get_test_ssh() + + +@pytest.mark.skipif("not TEST_SSH", reason="TEST_SSH is false") +class TestSSHSetup(BaseTest): + + @pytest.fixture(scope="function") + def project(test, tmp_path, setup_ssh_container): + """ + Make a project as per usual, but now add + in test ssh configurations + """ + tmp_path = tmp_path / "test with space" + + test_project_name = "test_ssh" + project, cwd = test_utils.setup_project_fixture( + tmp_path, test_project_name + ) + ssh_test_utils.setup_project_for_ssh( + project, + central_path=f"/home/sshuser/datashuttle/{project.project_name}", + central_host_id="localhost", + central_host_username="sshuser", + ) + yield project + test_utils.teardown_project(cwd, project) + + # ----------------------------------------------------------------- + # Test Setup SSH Connection + # ----------------------------------------------------------------- + + @pytest.mark.parametrize("input_", ["n", "o", "@"]) + def test_verify_ssh_central_host_do_not_accept( + self, capsys, project, input_ + ): + """ + Use the main function to test this. Test the sub-function + when accepting, because this main function will also + call setup ssh key pairs which we don't want to do yet + + This should only accept for "y" so try some random strings + including "n" and check they all do not make the connection. + """ + orig_builtin = ssh_test_utils.setup_mock_input(input_) + + project.setup_ssh_connection() + + ssh_test_utils.restore_mock_input(orig_builtin) + + captured = capsys.readouterr() + + assert "Host not accepted. No connection made.\n" in captured.out + + def test_verify_ssh_central_host_accept(self, capsys, project): + """ + User is asked to accept the server hostkey. Mock this here + and check hostkey is successfully accepted and written to configs. + """ + test_utils.clear_capsys(capsys) + + verified = ssh_test_utils.setup_ssh_connection( + project, setup_ssh_key_pair=False + ) + + assert verified + captured = capsys.readouterr() + + assert captured.out == "Host accepted.\n" + + with open(project.cfg.hostkeys_path, "r") as file: + hostkey = file.readlines()[0] + + assert ( + f"[{project.cfg['central_host_id']}]:3306 ssh-ed25519 " in hostkey + ) + + def test_generate_and_write_ssh_key(self, project): + """ + Check ssh key for passwordless connection is written + to file + """ + path_to_save = project.cfg["local_path"] / "test" + ssh.generate_and_write_ssh_key(path_to_save) + + with open(path_to_save, "r") as file: + first_line = file.readlines()[0] + + assert first_line == "-----BEGIN RSA PRIVATE KEY-----\n" diff --git a/tests/tests_integration/test_ssh.py b/tests/tests_integration/test_ssh_transfer.py similarity index 57% rename from tests/tests_integration/test_ssh.py rename to tests/tests_integration/test_ssh_transfer.py index 89889030..da090714 100644 --- a/tests/tests_integration/test_ssh.py +++ b/tests/tests_integration/test_ssh_transfer.py @@ -1,10 +1,8 @@ import shutil -import subprocess import paramiko import pytest import ssh_test_utils -import test_utils from base_transfer import BaseTransfer # from pytest import ssh_config @@ -14,18 +12,7 @@ @pytest.mark.skipif("not TEST_SSH", reason="TEST_SSH is false") -class TestSSH(BaseTransfer): - - @pytest.fixture( - scope="session", - ) - def setup_ssh_container(self): - # Annoying session scope does not seem to actually work - container_name = "running_ssh_tests" - ssh_test_utils.setup_ssh_container(container_name) - yield - subprocess.run(f"docker stop {container_name}", shell=True) - subprocess.run(f"docker rm {container_name}", shell=True) +class TestSSHTransfer(BaseTransfer): @pytest.fixture( scope="class", @@ -51,89 +38,6 @@ def ssh_setup(self, pathtable_and_project, setup_ssh_container): return [pathtable, project] - @pytest.fixture(scope="function") - def project(test, tmp_path, setup_ssh_container): - """ - Make a project as per usual, but now add - in test ssh configurations - """ - tmp_path = tmp_path / "test with space" - - test_project_name = "test_ssh" - project, cwd = test_utils.setup_project_fixture( - tmp_path, test_project_name - ) - ssh_test_utils.setup_project_for_ssh( - project, - central_path=f"/home/sshuser/datashuttle/{project.project_name}", - central_host_id="localhost", - central_host_username="sshuser", - ) - yield project - test_utils.teardown_project(cwd, project) - - # ----------------------------------------------------------------- - # Test Setup SSH Connection - # ----------------------------------------------------------------- - - @pytest.mark.parametrize("input_", ["n", "o", "@"]) - def test_verify_ssh_central_host_do_not_accept( - self, capsys, project, input_ - ): - """ - Use the main function to test this. Test the sub-function - when accepting, because this main function will also - call setup ssh key pairs which we don't want to do yet - - This should only accept for "y" so try some random strings - including "n" and check they all do not make the connection. - """ - orig_builtin = ssh_test_utils.setup_mock_input(input_) - - project.setup_ssh_connection() - - ssh_test_utils.restore_mock_input(orig_builtin) - - captured = capsys.readouterr() - - assert "Host not accepted. No connection made.\n" in captured.out - - def test_verify_ssh_central_host_accept(self, capsys, project): - """ - User is asked to accept the server hostkey. Mock this here - and check hostkey is successfully accepted and written to configs. - """ - test_utils.clear_capsys(capsys) - - verified = ssh_test_utils.setup_ssh_connection( - project, setup_ssh_key_pair=False - ) - - assert verified - captured = capsys.readouterr() - - assert captured.out == "Host accepted.\n" - - with open(project.cfg.hostkeys_path, "r") as file: - hostkey = file.readlines()[0] - - assert ( - f"[{project.cfg['central_host_id']}]:3306 ssh-ed25519 " in hostkey - ) - - def test_generate_and_write_ssh_key(self, project): - """ - Check ssh key for passwordless connection is written - to file - """ - path_to_save = project.cfg["local_path"] / "test" - ssh.generate_and_write_ssh_key(path_to_save) - - with open(path_to_save, "r") as file: - first_line = file.readlines()[0] - - assert first_line == "-----BEGIN RSA PRIVATE KEY-----\n" - # ----------------------------------------------------------------- # Test Setup SSH Connection # ----------------------------------------------------------------- @@ -174,9 +78,7 @@ def test_combinations_ssh_transfer( tmp_central_path = ( project.cfg["central_path"] / "tmp" / project.project_name ) - project.get_logging_path().mkdir( - parents=True, exist_ok=True - ) # TODO: why is this necessary + self.remake_logging_path(project) project.update_config_file(central_path=tmp_central_path) @@ -237,10 +139,12 @@ def test_combinations_ssh_transfer( shutil.rmtree(tmp_local_path) - project.get_logging_path().mkdir( - parents=True, exist_ok=True - ) # TODO: why is this necessary + self.remake_logging_path(project) project.update_config_file(local_path=true_local_path) - project.get_logging_path().mkdir( - parents=True, exist_ok=True - ) # TODO: why is this necessary + + def remake_logging_path(self, project): + """ + Need to do this to compensate for switching + local_path location in the test environment. + """ + project.get_logging_path().mkdir(parents=True, exist_ok=True) # TOD diff --git a/tests/tests_integration/test_filesystem_transfer.py b/tests/tests_integration/test_transfer.py similarity index 100% rename from tests/tests_integration/test_filesystem_transfer.py rename to tests/tests_integration/test_transfer.py diff --git a/tests/tests_integration/test_ssh_file_transfer.py b/tests/tests_integration/test_transfer_special_arguments.py similarity index 100% rename from tests/tests_integration/test_ssh_file_transfer.py rename to tests/tests_integration/test_transfer_special_arguments.py