Skip to content

Commit

Permalink
Add test and tweak logic to permit symlinks elsewhere for flow.cylc file
Browse files Browse the repository at this point in the history
  • Loading branch information
datamel committed Nov 29, 2021
1 parent d70b3de commit e366a38
Show file tree
Hide file tree
Showing 2 changed files with 10 additions and 32 deletions.
8 changes: 6 additions & 2 deletions cylc/flow/workflow_files.py
Original file line number Diff line number Diff line change
Expand Up @@ -1792,9 +1792,13 @@ def is_forbidden(flow_file: Path) -> bool:
return True
return False
link = flow_file.resolve()
if Path(link).parent == flow_file.parent: # link points within dir
if link.parent == flow_file.parent:
# link points within dir (permitted)
return False
return True
# link points elsewhere, check that suite.rc does not also exist in dir
if flow_file.parent.joinpath(WorkflowFiles.SUITE_RC).exists():
return True
return False


def detect_flow_exists(
Expand Down
34 changes: 4 additions & 30 deletions tests/unit/test_workflow_files.py
Original file line number Diff line number Diff line change
Expand Up @@ -1846,14 +1846,6 @@ def test_detect_both_flow_and_suite(tmp_path):
WorkflowFiles.SUITE_RC, False, WorkflowFilesError, None,
id="flow.cylc symlinked to non-existent suite.rc"
),
pytest.param(
'other-path', True, None, WorkflowFiles.SUITE_RC,
id="flow.cylc symlinked to other file, suite.rc exists"
),
pytest.param(
'other-path', False, WorkflowFilesError, None,
id="flow.cylc symlinked to other file, no suite.rc"
),
pytest.param(
None, True, None, WorkflowFiles.SUITE_RC,
id="No flow.cylc, suite.rc exists"
Expand All @@ -1864,17 +1856,11 @@ def test_detect_both_flow_and_suite(tmp_path):
),
]
)
@pytest.mark.parametrize(
'symlink_suiterc_arg',
[pytest.param(False, id="symlink_suiterc=False "),
pytest.param(True, id="symlink_suiterc=True ")]
)
def test_check_flow_file_symlink(
flow_file_target: Optional[str],
suiterc_exists: bool,
err: Optional[Type[Exception]],
expected_file: Optional[str],
symlink_suiterc_arg: bool,
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Test check_flow_file() when flow.cylc is a symlink or doesn't exist.
Expand All @@ -1887,8 +1873,6 @@ def test_check_flow_file_symlink(
expected_file: Which file's path should get returned, when
symlink_suiterc_arg is FALSE (otherwise it will always be
flow.cylc, assuming no exception occurred).
symlink_suiterc_arg: Value of the symlink_suiterc arg passed to
check_flow_file().
"""
flow_file = tmp_path.joinpath(WorkflowFiles.FLOW_FILE)
suiterc = tmp_path.joinpath(WorkflowFiles.SUITE_RC)
Expand All @@ -1898,25 +1882,14 @@ def test_check_flow_file_symlink(
if flow_file_target:
flow_file.symlink_to(flow_file_target)

caplog.set_level(logging.INFO, CYLC_LOG)
log_msg = ""

if err:
with pytest.raises(err):
check_flow_file(tmp_path, symlink_suiterc_arg)
check_flow_file(tmp_path)
else:
assert expected_file is not None # otherwise test is wrong
result = check_flow_file(tmp_path, symlink_suiterc_arg)
if symlink_suiterc_arg is True:
assert flow_file.samefile(suiterc)
expected_file = WorkflowFiles.FLOW_FILE
if flow_file_target != WorkflowFiles.SUITE_RC:
log_msg = "Symlink created: flow.cylc -> suite.rc"
result = check_flow_file(tmp_path)
assert result == tmp_path.joinpath(expected_file)

if log_msg:
assert log_msg in caplog.messages


def test_detect_both_flow_and_suite_symlinked(tmp_path):
"""Test flow.cylc symlinked to suite.rc together in dir is permitted."""
Expand All @@ -1935,6 +1908,8 @@ def test_flow_symlinked_elsewhere_and_suite_present(tmp_path: Path):
run_dir.mkdir(exist_ok=True)
flow_file = (run_dir / WorkflowFiles.FLOW_FILE)
flow_file.symlink_to(suite_file)
forbidden_external = is_forbidden(flow_file)
assert forbidden_external is False
(run_dir / WorkflowFiles.SUITE_RC).touch()
forbidden = is_forbidden(flow_file)
assert forbidden is True
Expand All @@ -1951,7 +1926,6 @@ def test_flow_symlinked_elsewhere_and_suite_present(tmp_path: Path):

def test_is_forbidden_symlink_returns_false_for_non_symlink(tmp_path):
"""Test sending a non symlink path is not marked as forbidden"""
Path(tmp_path)
flow_file = (tmp_path / WorkflowFiles.FLOW_FILE)
flow_file.touch()
forbidden = is_forbidden(Path(flow_file))
Expand Down

0 comments on commit e366a38

Please sign in to comment.