Source code for ewoksid22.tests.test_dirutils

import socket

import pytest

from .. import dirutils




def test_prepare_outdirs():
    primary_outdir = None
    outdirs = dict()

    result = dirutils.prepare_outdirs(outdirs, primary_outdir)
    assert result == dict()

    primary_outdir = "/users/opid22/inhouse/id222207/id22/20220701/processed"
    result = dirutils.prepare_outdirs(outdirs, primary_outdir)
    expected = {
        "primary": (
            None,
            "/users/opid22/inhouse/id222207/id22/20220701/processed",
        )
    }
    assert result == expected

    outdirs = {
        "primary": "opid22@diffract22new:/users/opid22/data1/",
        "secondary": "opid22@diffract22new:/users/opid22/data1/",
    }
    with pytest.raises(ValueError):
        dirutils.prepare_outdirs(outdirs, primary_outdir)

    outdirs = {
        "primary": f"opid22@{socket.gethostname()}:/users/opid22/data1/",
        "secondary": "opid22@diffract22new:/users/opid22/data1/",
    }
    result = dirutils.prepare_outdirs(outdirs, primary_outdir)
    expected = {
        "primary": (None, "/users/opid22/data1/"),
        "secondary": ("opid22@diffract22new", "/users/opid22/data1/"),
        "processed": (
            None,
            "/users/opid22/inhouse/id222207/id22/20220701/processed",
        ),
    }
    assert result == expected





def test_copy_local_file_creates_file(tmp_path):
    src = tmp_path / "src.txt"
    dst = tmp_path / "dst.txt"

    src.write_text("hello")

    dirutils.copy_local_file(str(src), str(dst), overwrite=False)

    assert dst.exists()
    assert dst.read_text() == "hello"





def test_copy_local_file_no_overwrite(tmp_path):
    src = tmp_path / "src.txt"
    dst = tmp_path / "dst.txt"

    src.write_text("original")
    dirutils.copy_local_file(str(src), str(dst), overwrite=False)

    src.write_text("new content")
    dirutils.copy_local_file(str(src), str(dst), overwrite=False)

    assert dst.read_text() == "original"





def test_copy_local_file_with_overwrite(tmp_path):
    src = tmp_path / "src.txt"
    dst = tmp_path / "dst.txt"

    src.write_text("original")
    dirutils.copy_local_file(str(src), str(dst), overwrite=False)

    src.write_text("new content")
    dirutils.copy_local_file(str(src), str(dst), overwrite=True)

    assert dst.read_text() == "new content"





def test_copy_file_to_primary(tmp_path):
    primary_dir = tmp_path / "primary"
    primary_dir.mkdir()

    src = tmp_path / "file.txt"
    src.write_text("data")

    outdirs = {"primary": (None, str(primary_dir))}

    dirutils.copy_file_to_primary(str(src), outdirs, overwrite=False)

    dst = primary_dir / "file.txt"
    assert dst.exists()
    assert dst.read_text() == "data"





def test_copy_file_to_primary_overwrite(tmp_path):
    primary_dir = tmp_path / "primary"
    primary_dir.mkdir()

    src = tmp_path / "file.txt"
    src.write_text("v1")

    outdirs = {"primary": (None, str(primary_dir))}

    dirutils.copy_file_to_primary(str(src), outdirs, overwrite=False)
    dst = primary_dir / "file.txt"

    src.write_text("v2")
    dirutils.copy_file_to_primary(str(src), outdirs, overwrite=True)

    assert dst.read_text() == "v2"





def test_copy_primary_file(tmp_path):
    primary_dir = tmp_path / "primary"
    other_dir = tmp_path / "other"

    primary_dir.mkdir()
    other_dir.mkdir()

    file_path = primary_dir / "data.txt"
    file_path.write_text("content")

    outdirs = {
        "primary": (None, str(primary_dir)),
        "processed": (None, str(other_dir)),
    }

    dirutils.copy_primary_file("data.txt", outdirs, overwrite=False)

    dst = other_dir / "data.txt"
    assert dst.exists()
    assert dst.read_text() == "content"





def test_copy_primary_file_no_overwrite(tmp_path):
    primary_dir = tmp_path / "primary"
    other_dir = tmp_path / "other"

    primary_dir.mkdir()
    other_dir.mkdir()

    file_path = primary_dir / "data.txt"
    file_path.write_text("v1")

    outdirs = {
        "primary": (None, str(primary_dir)),
        "processed": (None, str(other_dir)),
    }

    dirutils.copy_primary_file("data.txt", outdirs, overwrite=False)
    dst = other_dir / "data.txt"

    file_path.write_text("v2")
    dirutils.copy_primary_file("data.txt", outdirs, overwrite=False)

    assert dst.read_text() == "v1"





def test_copy_primary_file_with_overwrite(tmp_path):
    primary_dir = tmp_path / "primary"
    other_dir = tmp_path / "other"

    primary_dir.mkdir()
    other_dir.mkdir()

    file_path = primary_dir / "data.txt"
    file_path.write_text("v1")

    outdirs = {
        "primary": (None, str(primary_dir)),
        "processed": (None, str(other_dir)),
    }

    dirutils.copy_primary_file("data.txt", outdirs, overwrite=False)
    dst = other_dir / "data.txt"

    file_path.write_text("v2")
    dirutils.copy_primary_file("data.txt", outdirs, overwrite=True)

    assert dst.read_text() == "v2"