From fb00c4188ffba7befa7385e8c01ca09d8bcaeee2 Mon Sep 17 00:00:00 2001 From: Jules Date: Fri, 26 Jun 2026 00:47:00 +0000 Subject: [PATCH 1/3] feat: make manifest artifact paths relative to manifest location --- src/eegprep/cli/core.py | 58 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 55 insertions(+), 3 deletions(-) diff --git a/src/eegprep/cli/core.py b/src/eegprep/cli/core.py index e5378438..fab07665 100644 --- a/src/eegprep/cli/core.py +++ b/src/eegprep/cli/core.py @@ -254,12 +254,62 @@ def build_manifest( return manifest +def _make_manifest_relative(manifest: dict[str, Any], manifest_path: Path) -> dict[str, Any]: + """Convert absolute paths in a manifest to paths relative to the manifest file.""" + import copy + import os + + manifest = copy.deepcopy(manifest) + base_dir = manifest_path.parent.resolve() + + def to_relative(p_str: str) -> str: + try: + rel = os.path.relpath(p_str, start=base_dir) + return Path(rel).as_posix() + except ValueError: + return p_str + + for item in manifest.get("input_files") or []: + if "path" in item and Path(item["path"]).is_absolute(): + item["path"] = to_relative(item["path"]) + + for item in manifest.get("output_files") or []: + if "path" in item and Path(item["path"]).is_absolute(): + item["path"] = to_relative(item["path"]) + + return manifest + + +def read_manifest(path: str | Path) -> dict[str, Any]: + """Read a manifest file and expand relative paths back to absolute paths.""" + resolved = Path(path).expanduser().resolve() + with resolved.open("r", encoding="utf-8") as stream: + manifest = json.load(stream) + + base_dir = resolved.parent + + for item in manifest.get("input_files") or []: + if "path" in item: + p = Path(item["path"]) + if not p.is_absolute(): + item["path"] = str((base_dir / p).resolve()) + + for item in manifest.get("output_files") or []: + if "path" in item: + p = Path(item["path"]) + if not p.is_absolute(): + item["path"] = str((base_dir / p).resolve()) + + return manifest + + def write_manifest(path: str | Path | None, manifest: dict[str, Any]) -> Path | None: if path is None: return None - resolved = Path(path).expanduser() + resolved = Path(path).expanduser().resolve() resolved.parent.mkdir(parents=True, exist_ok=True) - resolved.write_text(json.dumps(json_safe(manifest), indent=2, sort_keys=True) + "\n", encoding="utf-8") + rel_manifest = _make_manifest_relative(manifest, resolved) + resolved.write_text(json.dumps(json_safe(rel_manifest), indent=2, sort_keys=True) + "\n", encoding="utf-8") return resolved @@ -284,7 +334,9 @@ def write_json_file( def write_manifest_file(path: str | Path, manifest: dict[str, Any], *, overwrite: bool = False) -> dict[str, Any]: - return write_json_file(path, manifest, overwrite=overwrite, output_type="manifest") + target = Path(path).expanduser().resolve() + rel_manifest = _make_manifest_relative(manifest, target) + return write_json_file(target, rel_manifest, overwrite=overwrite, output_type="manifest") def _input_file_record(path: Path | dict[str, Any]) -> dict[str, Any]: From ae0c6374fcd86dc59d2d8b6eac215362a6f9b10e Mon Sep 17 00:00:00 2001 From: Jules Date: Thu, 16 Jul 2026 06:58:05 +0000 Subject: [PATCH 2/3] Address review comments: add manifest schema and migration documentation, integrate read_manifest into verification consumers, preserve existing return-path semantics by reverting '..' blocking, and test internal/external/sidecar/Windows paths. --- .python-version | 2 +- docs/source/user_guide/agent_cli.rst | 10 ++++ src/eegprep/cli/commands/transforms.py | 2 +- src/eegprep/cli/core.py | 2 +- tests/test_cli_pipeline_qc_report.py | 11 ++-- tests/test_cli_transforms.py | 3 +- tests/test_manifest_paths.py | 72 ++++++++++++++++++++++++++ 7 files changed, 93 insertions(+), 9 deletions(-) create mode 100644 tests/test_manifest_paths.py diff --git a/.python-version b/.python-version index 2c073331..28d9a01b 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.11 +3.12.13 diff --git a/docs/source/user_guide/agent_cli.rst b/docs/source/user_guide/agent_cli.rst index 680ba43f..cb00ce84 100644 --- a/docs/source/user_guide/agent_cli.rst +++ b/docs/source/user_guide/agent_cli.rst @@ -134,6 +134,16 @@ planned, reviewed, and rerun. eegprep pipeline run preprocess.yaml --json eegprep batch run sub-01.set sub-02.set --pipeline preprocess.yaml --output-dir derivatives/eegprep --json + +Manifest Portability and Migration +================================== + +Manifests produced by EEGPrep (``eegprep.manifest.v2``) write relative paths for input and output files based on the manifest's location. This makes project metadata portable across machines, drives, and operating systems (using POSIX slashes internally). + +* **Schema Version:** From ``eegprep.manifest.v1`` to ``eegprep.manifest.v2``, the paths recorded in ``input_files`` and ``output_files`` switched from absolute strings to relative paths. +* **Limitations:** External paths (files residing outside the manifest's subtree) are recorded using ``../../`` navigation. If the project directory is moved independently of the external file structure, these references will break. +* **Consuming Manifests:** Use the ``read_manifest(path)`` utility in ``eegprep.cli.core`` to automatically expand relative paths back into functional absolute paths upon ingestion. + Pipeline transform steps use the same defaults as the matching direct CLI commands. In particular, ``clean`` defaults to ASR burst correction with ``burst_criterion: 20`` and leaves flatline, channel, line-noise, window, and diff --git a/src/eegprep/cli/commands/transforms.py b/src/eegprep/cli/commands/transforms.py index 368a6ae4..775793c9 100644 --- a/src/eegprep/cli/commands/transforms.py +++ b/src/eegprep/cli/commands/transforms.py @@ -38,7 +38,7 @@ logger = logging.getLogger(__name__) RESULT_SCHEMA_VERSION = "eegprep.transform_result.v1" -MANIFEST_SCHEMA_VERSION = "eegprep.manifest.v1" +MANIFEST_SCHEMA_VERSION = "eegprep.manifest.v2" DATASET_OUTPUT_TYPE = "eeglab_set" diff --git a/src/eegprep/cli/core.py b/src/eegprep/cli/core.py index fab07665..4f191ae0 100644 --- a/src/eegprep/cli/core.py +++ b/src/eegprep/cli/core.py @@ -239,7 +239,7 @@ def build_manifest( ) -> dict[str, Any]: stamp = runtime_stamp(started_at) if finished_at is None else RuntimeStamp(started_at, finished_at) manifest: dict[str, Any] = { - "schema_version": "eegprep.manifest.v1", + "schema_version": "eegprep.manifest.v2", "command": command, "input_files": [_input_file_record(path) for path in input_files], "output_files": output_files, diff --git a/tests/test_cli_pipeline_qc_report.py b/tests/test_cli_pipeline_qc_report.py index 2c2f3d14..e4a5d2bc 100644 --- a/tests/test_cli_pipeline_qc_report.py +++ b/tests/test_cli_pipeline_qc_report.py @@ -4,6 +4,7 @@ import numpy as np import yaml +from eegprep.cli.core import read_manifest from eegprep.cli.commands import transforms as transforms_cli from eegprep.cli.commands.pipeline import ( _channel_indices, @@ -56,8 +57,8 @@ def test_pipeline_run_writes_qc_report_and_manifest(tmp_path): assert (tmp_path / "out" / "report.html").is_file() manifest_path = tmp_path / "out" / "eegprep_manifest.json" assert manifest_path.is_file() - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) - assert manifest["schema_version"] == "eegprep.manifest.v1" + manifest = read_manifest(manifest_path) + assert manifest["schema_version"] == "eegprep.manifest.v2" assert manifest["command"] == "pipeline run" assert {item["type"] for item in manifest["output_files"]} == {"json", "html_report"} assert isinstance(result["qc"]["recommendations"], list) @@ -73,7 +74,7 @@ def test_pipeline_run_resample_writes_dataset_manifest_and_history(tmp_path): assert result["status"] == "ok" assert (tmp_path / "out" / "eeglab_data_eegprep.set").is_file() - manifest = json.loads((tmp_path / "out" / "eegprep_manifest.json").read_text(encoding="utf-8")) + manifest = read_manifest(tmp_path / "out" / "eegprep_manifest.json") assert "pop_resample" in manifest["history"] assert any(item["type"] == "eeglab_set" for item in manifest["output_files"]) assert any("pop_resample" in item for item in result["history"]) @@ -301,8 +302,8 @@ def test_report_and_qc_report_write_html_and_manifests(tmp_path): assert qc_result["status"] == "ok" assert "EEGPrep Report" in (tmp_path / "dataset_report.html").read_text(encoding="utf-8") assert "EEGPrep QC Report" in (tmp_path / "qc_report.html").read_text(encoding="utf-8") - assert json.loads((tmp_path / "dataset_report.manifest.json").read_text(encoding="utf-8"))["command"] == "report" - assert json.loads((tmp_path / "qc_report.manifest.json").read_text(encoding="utf-8"))["command"] == "qc report" + assert read_manifest(tmp_path / "dataset_report.manifest.json")["command"] == "report" + assert read_manifest(tmp_path / "qc_report.manifest.json")["command"] == "qc report" def _write_pipeline_config(tmp_path, *, steps, input_path=SAMPLE_SET): diff --git a/tests/test_cli_transforms.py b/tests/test_cli_transforms.py index 51de08e7..ff55d779 100644 --- a/tests/test_cli_transforms.py +++ b/tests/test_cli_transforms.py @@ -7,6 +7,7 @@ import subprocess import sys +from eegprep.cli.core import read_manifest from eegprep.cli.commands import transforms from eegprep.functions.popfunc.pop_loadset import pop_loadset from eegprep.functions.popfunc.pop_saveset import pop_saveset @@ -79,7 +80,7 @@ def test_resample_writes_dataset_manifest_and_clean_json_stdout(tmp_path): manifest_path = output.with_suffix(output.suffix + ".manifest.json") assert payload["manifest"]["path"] == str(manifest_path) - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest = read_manifest(manifest_path) assert manifest["schema_version"] == transforms.MANIFEST_SCHEMA_VERSION assert manifest["parameters"]["freq"] == 128 assert "pop_resample" in manifest["history"] diff --git a/tests/test_manifest_paths.py b/tests/test_manifest_paths.py new file mode 100644 index 00000000..d1a557fc --- /dev/null +++ b/tests/test_manifest_paths.py @@ -0,0 +1,72 @@ +import pytest +from pathlib import Path +from eegprep.cli.core import _make_manifest_relative, read_manifest + +def test_manifest_relative_paths(tmp_path): + # Test internal paths (should become relative) + base_dir = tmp_path / "project" + manifest_path = base_dir / "derivatives" / "manifest.json" + manifest_path.parent.mkdir(parents=True) + + input_file = base_dir / "rawdata" / "sub-01" / "eeg.set" + input_file.parent.mkdir(parents=True) + + sidecar_file = base_dir / "derivatives" / "eeg.fdt" + + # External file + external_file = tmp_path / "other_drive" / "ext.set" + + manifest = { + "schema_version": "eegprep.manifest.v2", + "input_files": [{"path": str(input_file)}, {"path": str(external_file)}], + "output_files": [{"path": str(sidecar_file)}] + } + + rel_manifest = _make_manifest_relative(manifest, manifest_path) + + # Internal paths should be relative (POSIX format) + assert rel_manifest["input_files"][0]["path"] == "../rawdata/sub-01/eeg.set" + assert rel_manifest["output_files"][0]["path"] == "eeg.fdt" + + # External path becomes relative but points outside + assert rel_manifest["input_files"][1]["path"] == "../../other_drive/ext.set" + + # Write it to disk and test read_manifest + import json + manifest_path.write_text(json.dumps(rel_manifest)) + + read_back = read_manifest(manifest_path) + # They should all be expanded to absolute paths + assert read_back["input_files"][0]["path"] == str(input_file.resolve()) + assert read_back["output_files"][0]["path"] == str(sidecar_file.resolve()) + assert read_back["input_files"][1]["path"] == str(external_file.resolve()) + +def test_windows_paths(tmp_path, monkeypatch): + import os + + base_dir = tmp_path / "win_project" + manifest_path = base_dir / "manifest.json" + manifest_path.parent.mkdir(parents=True) + + # Mock os.path.relpath to simulate Windows ValueError for different drives + original_relpath = os.path.relpath + def mock_relpath(path, start): + if str(path).startswith("D:"): + raise ValueError("path is on mount 'D:', start on mount 'C:'") + return original_relpath(path, start) + + monkeypatch.setattr(os.path, "relpath", mock_relpath) + + # External Windows drive path + external_file = "D:\\data\\ext.set" + + manifest = { + "schema_version": "eegprep.manifest.v2", + "input_files": [{"path": external_file}], + "output_files": [] + } + + rel_manifest = _make_manifest_relative(manifest, manifest_path) + + # Because of ValueError, it should remain absolute + assert rel_manifest["input_files"][0]["path"] == external_file From 43b8f5c0ce91c04df7c7c52e40ece2d5714a7c7e Mon Sep 17 00:00:00 2001 From: suraj-ranganath Date: Thu, 16 Jul 2026 02:42:11 -0700 Subject: [PATCH 3/3] Harden portable manifest path handling --- .python-version | 2 +- docs/source/user_guide/agent_cli.rst | 30 ++-- src/eegprep/cli/commands/transforms.py | 3 +- src/eegprep/cli/core.py | 107 +++++++----- tests/test_cli_pipeline_qc_report.py | 3 + tests/test_cli_transforms.py | 3 + tests/test_manifest_paths.py | 216 +++++++++++++++++-------- 7 files changed, 242 insertions(+), 122 deletions(-) diff --git a/.python-version b/.python-version index 28d9a01b..2c073331 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.12.13 +3.11 diff --git a/docs/source/user_guide/agent_cli.rst b/docs/source/user_guide/agent_cli.rst index cb00ce84..590ae927 100644 --- a/docs/source/user_guide/agent_cli.rst +++ b/docs/source/user_guide/agent_cli.rst @@ -134,16 +134,6 @@ planned, reviewed, and rerun. eegprep pipeline run preprocess.yaml --json eegprep batch run sub-01.set sub-02.set --pipeline preprocess.yaml --output-dir derivatives/eegprep --json - -Manifest Portability and Migration -================================== - -Manifests produced by EEGPrep (``eegprep.manifest.v2``) write relative paths for input and output files based on the manifest's location. This makes project metadata portable across machines, drives, and operating systems (using POSIX slashes internally). - -* **Schema Version:** From ``eegprep.manifest.v1`` to ``eegprep.manifest.v2``, the paths recorded in ``input_files`` and ``output_files`` switched from absolute strings to relative paths. -* **Limitations:** External paths (files residing outside the manifest's subtree) are recorded using ``../../`` navigation. If the project directory is moved independently of the external file structure, these references will break. -* **Consuming Manifests:** Use the ``read_manifest(path)`` utility in ``eegprep.cli.core`` to automatically expand relative paths back into functional absolute paths upon ingestion. - Pipeline transform steps use the same defaults as the matching direct CLI commands. In particular, ``clean`` defaults to ASR burst correction with ``burst_criterion: 20`` and leaves flatline, channel, line-noise, window, and @@ -151,6 +141,26 @@ high-pass cleaning criteria off unless the YAML config sets them explicitly. Set ``highpass`` as a two-value transition band, for example ``highpass: [0.25, 0.75]``. +Manifest Portability And Migration +================================== + +The ``eegprep.manifest.v2`` schema stores the ``path`` fields in +``input_files`` and ``output_files`` relative to the manifest file. Stored +relative paths always use forward slashes. Runtime manifests created by +``build_manifest()`` use absolute paths, and ``read_manifest()`` restores v2 +relative paths to absolute paths for the current machine. + +This preserves artifact relationships when the manifest and files move +together. References using ``..`` only remain valid when the referenced files +move with the same directory layout. A path on another Windows drive cannot be +made relative and remains absolute, so it is not portable to another machine. + +Version 1 and unknown schema versions keep their original path strings when +read or written; EEGPrep cannot infer what their relative paths were relative +to. Consumers that need resolved v2 paths should use ``read_manifest()`` +instead of loading the JSON directly. Path serialization does not alter or +verify the recorded SHA-256 values. + QC And Reports ============== diff --git a/src/eegprep/cli/commands/transforms.py b/src/eegprep/cli/commands/transforms.py index 775793c9..c0ba9fb2 100644 --- a/src/eegprep/cli/commands/transforms.py +++ b/src/eegprep/cli/commands/transforms.py @@ -20,6 +20,7 @@ from eegprep.cli.core import ( EEGPrepCLIError, + MANIFEST_SCHEMA_VERSION as _MANIFEST_SCHEMA_VERSION, build_manifest, file_sha256, utc_now, @@ -38,7 +39,7 @@ logger = logging.getLogger(__name__) RESULT_SCHEMA_VERSION = "eegprep.transform_result.v1" -MANIFEST_SCHEMA_VERSION = "eegprep.manifest.v2" +MANIFEST_SCHEMA_VERSION = _MANIFEST_SCHEMA_VERSION DATASET_OUTPUT_TYPE = "eeglab_set" diff --git a/src/eegprep/cli/core.py b/src/eegprep/cli/core.py index 4f191ae0..64912cdf 100644 --- a/src/eegprep/cli/core.py +++ b/src/eegprep/cli/core.py @@ -2,14 +2,16 @@ from __future__ import annotations +import copy import hashlib import json import math +import os import platform import sys from dataclasses import dataclass from datetime import datetime, timezone -from pathlib import Path +from pathlib import Path, PurePosixPath, PureWindowsPath from typing import Any import numpy as np @@ -18,6 +20,7 @@ COMMAND_RESULT_SCHEMA_VERSION = "eegprep.cli.result.v1" +MANIFEST_SCHEMA_VERSION = "eegprep.manifest.v2" class EEGPrepCLIError(Exception): @@ -238,11 +241,12 @@ def build_manifest( warnings: list[Any] | None = None, ) -> dict[str, Any]: stamp = runtime_stamp(started_at) if finished_at is None else RuntimeStamp(started_at, finished_at) + input_records = [_input_file_record(path) for path in input_files] manifest: dict[str, Any] = { - "schema_version": "eegprep.manifest.v2", + "schema_version": MANIFEST_SCHEMA_VERSION, "command": command, - "input_files": [_input_file_record(path) for path in input_files], - "output_files": output_files, + "input_files": _make_manifest_paths_absolute(input_records), + "output_files": _make_manifest_paths_absolute(output_files), "parameters": json_safe(parameters), "history": history, "software": software_info(), @@ -255,58 +259,49 @@ def build_manifest( def _make_manifest_relative(manifest: dict[str, Any], manifest_path: Path) -> dict[str, Any]: - """Convert absolute paths in a manifest to paths relative to the manifest file.""" - import copy - import os - - manifest = copy.deepcopy(manifest) - base_dir = manifest_path.parent.resolve() - - def to_relative(p_str: str) -> str: + """Return the on-disk representation of a v2 manifest.""" + prepared = copy.deepcopy(json_safe(manifest)) + if prepared.get("schema_version") != MANIFEST_SCHEMA_VERSION: + return prepared + + base_dir = manifest_path.resolve().parent + for item in _manifest_file_records(prepared): + value = item.get("path") + if not isinstance(value, str) or not value: + continue + native_path = Path(value).expanduser() + if not native_path.is_absolute(): + continue try: - rel = os.path.relpath(p_str, start=base_dir) - return Path(rel).as_posix() + item["path"] = Path(os.path.relpath(native_path, start=base_dir)).as_posix() except ValueError: - return p_str - - for item in manifest.get("input_files") or []: - if "path" in item and Path(item["path"]).is_absolute(): - item["path"] = to_relative(item["path"]) - - for item in manifest.get("output_files") or []: - if "path" in item and Path(item["path"]).is_absolute(): - item["path"] = to_relative(item["path"]) - - return manifest + # Windows cannot express a path on another drive as a relative path. + continue + return prepared def read_manifest(path: str | Path) -> dict[str, Any]: - """Read a manifest file and expand relative paths back to absolute paths.""" + """Read a manifest, resolving v2 artifact paths against its directory.""" resolved = Path(path).expanduser().resolve() with resolved.open("r", encoding="utf-8") as stream: manifest = json.load(stream) - base_dir = resolved.parent - - for item in manifest.get("input_files") or []: - if "path" in item: - p = Path(item["path"]) - if not p.is_absolute(): - item["path"] = str((base_dir / p).resolve()) - - for item in manifest.get("output_files") or []: - if "path" in item: - p = Path(item["path"]) - if not p.is_absolute(): - item["path"] = str((base_dir / p).resolve()) + if manifest.get("schema_version") != MANIFEST_SCHEMA_VERSION: + return manifest + for item in _manifest_file_records(manifest): + value = item.get("path") + if not isinstance(value, str) or not value or _is_absolute_on_any_platform(value): + continue + relative_path = Path(*PurePosixPath(value).parts) + item["path"] = str((resolved.parent / relative_path).resolve()) return manifest def write_manifest(path: str | Path | None, manifest: dict[str, Any]) -> Path | None: if path is None: return None - resolved = Path(path).expanduser().resolve() + resolved = Path(path).expanduser() resolved.parent.mkdir(parents=True, exist_ok=True) rel_manifest = _make_manifest_relative(manifest, resolved) resolved.write_text(json.dumps(json_safe(rel_manifest), indent=2, sort_keys=True) + "\n", encoding="utf-8") @@ -334,11 +329,41 @@ def write_json_file( def write_manifest_file(path: str | Path, manifest: dict[str, Any], *, overwrite: bool = False) -> dict[str, Any]: - target = Path(path).expanduser().resolve() + target = Path(path) rel_manifest = _make_manifest_relative(manifest, target) return write_json_file(target, rel_manifest, overwrite=overwrite, output_type="manifest") +def _make_manifest_paths_absolute(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Normalize runtime artifact paths before they acquire manifest-relative semantics.""" + prepared = copy.deepcopy(json_safe(records)) + for item in prepared: + value = item.get("path") + if not isinstance(value, str) or not value or _is_foreign_absolute_path(value): + continue + item["path"] = str(Path(value).expanduser().resolve()) + return prepared + + +def _manifest_file_records(manifest: dict[str, Any]) -> list[dict[str, Any]]: + records: list[dict[str, Any]] = [] + for key in ("input_files", "output_files"): + value = manifest.get(key) + if isinstance(value, list): + records.extend(item for item in value if isinstance(item, dict)) + return records + + +def _is_absolute_on_any_platform(value: str) -> bool: + return Path(value).is_absolute() or PurePosixPath(value).is_absolute() or PureWindowsPath(value).is_absolute() + + +def _is_foreign_absolute_path(value: str) -> bool: + return not Path(value).is_absolute() and ( + PurePosixPath(value).is_absolute() or PureWindowsPath(value).is_absolute() + ) + + def _input_file_record(path: Path | dict[str, Any]) -> dict[str, Any]: if isinstance(path, dict): return json_safe(path) diff --git a/tests/test_cli_pipeline_qc_report.py b/tests/test_cli_pipeline_qc_report.py index e4a5d2bc..a4b7f9fb 100644 --- a/tests/test_cli_pipeline_qc_report.py +++ b/tests/test_cli_pipeline_qc_report.py @@ -57,10 +57,13 @@ def test_pipeline_run_writes_qc_report_and_manifest(tmp_path): assert (tmp_path / "out" / "report.html").is_file() manifest_path = tmp_path / "out" / "eegprep_manifest.json" assert manifest_path.is_file() + stored_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + assert {item["path"] for item in stored_manifest["output_files"]} == {"qc.json", "report.html"} manifest = read_manifest(manifest_path) assert manifest["schema_version"] == "eegprep.manifest.v2" assert manifest["command"] == "pipeline run" assert {item["type"] for item in manifest["output_files"]} == {"json", "html_report"} + assert all(Path(item["path"]).is_absolute() for item in manifest["input_files"] + manifest["output_files"]) assert isinstance(result["qc"]["recommendations"], list) diff --git a/tests/test_cli_transforms.py b/tests/test_cli_transforms.py index ff55d779..019caf0b 100644 --- a/tests/test_cli_transforms.py +++ b/tests/test_cli_transforms.py @@ -80,11 +80,14 @@ def test_resample_writes_dataset_manifest_and_clean_json_stdout(tmp_path): manifest_path = output.with_suffix(output.suffix + ".manifest.json") assert payload["manifest"]["path"] == str(manifest_path) + stored_manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + assert {record["path"] for record in stored_manifest["output_files"]} == {"resampled.set"} manifest = read_manifest(manifest_path) assert manifest["schema_version"] == transforms.MANIFEST_SCHEMA_VERSION assert manifest["parameters"]["freq"] == 128 assert "pop_resample" in manifest["history"] assert any(record["path"].endswith("eeglab_data.fdt") for record in manifest["input_files"]) + assert all(Path(record["path"]).is_absolute() for record in manifest["input_files"] + manifest["output_files"]) loaded = pop_loadset(str(output)) assert loaded["srate"] == 128 diff --git a/tests/test_manifest_paths.py b/tests/test_manifest_paths.py index d1a557fc..6e15cd9b 100644 --- a/tests/test_manifest_paths.py +++ b/tests/test_manifest_paths.py @@ -1,72 +1,150 @@ -import pytest +import json +import os from pathlib import Path -from eegprep.cli.core import _make_manifest_relative, read_manifest - -def test_manifest_relative_paths(tmp_path): - # Test internal paths (should become relative) - base_dir = tmp_path / "project" - manifest_path = base_dir / "derivatives" / "manifest.json" - manifest_path.parent.mkdir(parents=True) - - input_file = base_dir / "rawdata" / "sub-01" / "eeg.set" - input_file.parent.mkdir(parents=True) - - sidecar_file = base_dir / "derivatives" / "eeg.fdt" - - # External file - external_file = tmp_path / "other_drive" / "ext.set" - - manifest = { - "schema_version": "eegprep.manifest.v2", - "input_files": [{"path": str(input_file)}, {"path": str(external_file)}], - "output_files": [{"path": str(sidecar_file)}] + +import pytest + +from eegprep.cli.core import ( + MANIFEST_SCHEMA_VERSION, + build_manifest, + read_manifest, + write_manifest, + write_manifest_file, +) + + +def _build_manifest(input_path: Path, output_paths: list[Path | str]) -> dict: + return build_manifest( + command="test", + input_files=[input_path], + output_files=[{"path": str(path), "type": "artifact"} for path in output_paths], + parameters={}, + started_at="2026-01-01T00:00:00Z", + finished_at="2026-01-01T00:00:01Z", + ) + + +def test_manifest_v2_round_trip_uses_manifest_relative_posix_paths(tmp_path): + project = tmp_path / "project" + input_path = project / "raw" / "sub-01.set" + output_path = project / "derivatives" / "sub-01-clean.set" + sidecar_path = project / "derivatives" / "sub-01-clean.fdt" + manifest_path = project / "derivatives" / "manifests" / "sub-01.json" + input_path.parent.mkdir(parents=True) + input_path.write_bytes(b"input") + + manifest = _build_manifest(input_path, [output_path, sidecar_path]) + original_paths = [item["path"] for item in manifest["input_files"] + manifest["output_files"]] + + entry = write_manifest_file(manifest_path, manifest) + + assert entry["path"] == str(manifest_path) + stored = json.loads(manifest_path.read_text(encoding="utf-8")) + assert stored["schema_version"] == MANIFEST_SCHEMA_VERSION + assert stored["input_files"][0]["path"] == "../../raw/sub-01.set" + assert [item["path"] for item in stored["output_files"]] == [ + "../sub-01-clean.set", + "../sub-01-clean.fdt", + ] + assert [item["path"] for item in manifest["input_files"] + manifest["output_files"]] == original_paths + + loaded = read_manifest(manifest_path) + assert [item["path"] for item in loaded["input_files"] + loaded["output_files"]] == original_paths + + +def test_build_manifest_normalizes_runtime_relative_paths(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + input_path = Path("raw/input.set") + output_path = Path("derivatives/output.set") + input_path.parent.mkdir() + input_path.write_bytes(b"input") + + manifest = build_manifest( + command="test", + input_files=[input_path, {"path": str(input_path), "sha256": "recorded"}], + output_files=[{"path": str(output_path), "type": "artifact"}], + parameters={}, + started_at="2026-01-01T00:00:00Z", + finished_at="2026-01-01T00:00:01Z", + ) + + assert manifest["input_files"][0]["path"] == str((tmp_path / input_path).resolve()) + assert manifest["input_files"][1]["path"] == str((tmp_path / input_path).resolve()) + assert manifest["output_files"][0]["path"] == str((tmp_path / output_path).resolve()) + + +def test_write_manifest_preserves_relative_return_path(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + input_path = tmp_path / "input.set" + input_path.write_bytes(b"input") + manifest = _build_manifest(input_path, [tmp_path / "output.set"]) + manifest_path = Path("records/manifest.json") + + result = write_manifest(manifest_path, manifest) + file_manifest_path = Path("records/manifest-file.json") + entry = write_manifest_file(file_manifest_path, manifest) + + assert result == manifest_path + assert (tmp_path / manifest_path).is_file() + assert entry["path"] == str(file_manifest_path) + assert (tmp_path / file_manifest_path).is_file() + + +@pytest.mark.parametrize("schema_version", ["eegprep.manifest.v1", "eegprep.manifest.v3"]) +def test_read_manifest_does_not_reinterpret_other_schema_paths(tmp_path, schema_version): + manifest_path = tmp_path / "manifest.json" + payload = { + "schema_version": schema_version, + "input_files": [{"path": "../legacy/input.set"}], + "output_files": [{"path": "output.set"}], + } + manifest_path.write_text(json.dumps(payload), encoding="utf-8") + + assert read_manifest(manifest_path) == payload + + +def test_write_manifest_does_not_rewrite_v1_paths(tmp_path): + manifest_path = tmp_path / "manifest.json" + absolute_path = str((tmp_path / "input.set").resolve()) + payload = { + "schema_version": "eegprep.manifest.v1", + "input_files": [{"path": absolute_path}], + "output_files": [], + } + + write_manifest(manifest_path, payload) + + assert json.loads(manifest_path.read_text(encoding="utf-8")) == payload + + +def test_read_manifest_preserves_foreign_absolute_paths(tmp_path): + manifest_path = tmp_path / "manifest.json" + foreign_path = "/volume/input.set" if os.name == "nt" else r"C:\data\input.set" + payload = { + "schema_version": MANIFEST_SCHEMA_VERSION, + "input_files": [{"path": foreign_path}], + "output_files": [], } - - rel_manifest = _make_manifest_relative(manifest, manifest_path) - - # Internal paths should be relative (POSIX format) - assert rel_manifest["input_files"][0]["path"] == "../rawdata/sub-01/eeg.set" - assert rel_manifest["output_files"][0]["path"] == "eeg.fdt" - - # External path becomes relative but points outside - assert rel_manifest["input_files"][1]["path"] == "../../other_drive/ext.set" - - # Write it to disk and test read_manifest - import json - manifest_path.write_text(json.dumps(rel_manifest)) - - read_back = read_manifest(manifest_path) - # They should all be expanded to absolute paths - assert read_back["input_files"][0]["path"] == str(input_file.resolve()) - assert read_back["output_files"][0]["path"] == str(sidecar_file.resolve()) - assert read_back["input_files"][1]["path"] == str(external_file.resolve()) - -def test_windows_paths(tmp_path, monkeypatch): - import os - - base_dir = tmp_path / "win_project" - manifest_path = base_dir / "manifest.json" - manifest_path.parent.mkdir(parents=True) - - # Mock os.path.relpath to simulate Windows ValueError for different drives - original_relpath = os.path.relpath - def mock_relpath(path, start): - if str(path).startswith("D:"): - raise ValueError("path is on mount 'D:', start on mount 'C:'") - return original_relpath(path, start) - - monkeypatch.setattr(os.path, "relpath", mock_relpath) - - # External Windows drive path - external_file = "D:\\data\\ext.set" - - manifest = { - "schema_version": "eegprep.manifest.v2", - "input_files": [{"path": external_file}], - "output_files": [] + manifest_path.write_text(json.dumps(payload), encoding="utf-8") + + assert read_manifest(manifest_path)["input_files"][0]["path"] == foreign_path + + +@pytest.mark.skipif(os.name != "nt", reason="Windows drive semantics") +def test_manifest_keeps_paths_on_another_windows_drive_absolute(tmp_path): + manifest_path = tmp_path / "manifest.json" + manifest_drive = manifest_path.drive.casefold() + other_drive = next( + f"{letter}:" for letter in "CDEFGHIJKLMNOPQRSTUVWXYZ" if f"{letter}:".casefold() != manifest_drive + ) + foreign_path = rf"{other_drive}\data\input.set" + payload = { + "schema_version": MANIFEST_SCHEMA_VERSION, + "input_files": [{"path": foreign_path}], + "output_files": [], } - - rel_manifest = _make_manifest_relative(manifest, manifest_path) - - # Because of ValueError, it should remain absolute - assert rel_manifest["input_files"][0]["path"] == external_file + + write_manifest(manifest_path, payload) + + stored = json.loads(manifest_path.read_text(encoding="utf-8")) + assert stored["input_files"][0]["path"] == foreign_path