89 lines
3.0 KiB
Python
89 lines
3.0 KiB
Python
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
def _load_confirmation_snapshot_module():
|
|
module_path = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "gfa_core"
|
|
/ "mas"
|
|
/ "confirmation_snapshot.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"confirmation_snapshot_linked_job_test",
|
|
module_path,
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def _write_legacy_config_confirmation(source_root: Path, job_id: str) -> Path:
|
|
input_job = source_root / "input" / job_id
|
|
final_dir = source_root / "workspace" / job_id / "final"
|
|
input_job.mkdir(parents=True)
|
|
final_dir.mkdir(parents=True)
|
|
config_path = input_job / "config.xlsx"
|
|
config_path.write_bytes(b"confirmed config")
|
|
manifest_path = final_dir / "config_autofill.manifest.json"
|
|
manifest_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"manifest_version": "1.0",
|
|
"stage": "config_autofill",
|
|
"status": "success",
|
|
"artifacts": {
|
|
"confirmed_config": {
|
|
"path": str(config_path),
|
|
"type": "xlsx",
|
|
}
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return config_path
|
|
|
|
|
|
def test_snapshot_accepts_paired_job_links_to_one_project_root(tmp_path):
|
|
snapshot_module = _load_confirmation_snapshot_module()
|
|
job_id = "job_linked"
|
|
project_root = tmp_path / "sj-2"
|
|
source_root = tmp_path / "source"
|
|
config_path = _write_legacy_config_confirmation(source_root, job_id)
|
|
(project_root / "input").mkdir(parents=True)
|
|
(project_root / "workspace").mkdir(parents=True)
|
|
(project_root / "input" / job_id).symlink_to(source_root / "input" / job_id)
|
|
(project_root / "workspace" / job_id).symlink_to(
|
|
source_root / "workspace" / job_id
|
|
)
|
|
|
|
snapshot = snapshot_module.build_confirmation_snapshot(project_root, job_id)
|
|
|
|
assert snapshot["config_confirmed"] is True
|
|
assert snapshot["config"]["artifact_path"] == str(config_path.resolve())
|
|
|
|
|
|
def test_snapshot_rejects_job_links_to_different_project_roots(tmp_path):
|
|
snapshot_module = _load_confirmation_snapshot_module()
|
|
job_id = "job_mismatch"
|
|
project_root = tmp_path / "sj-2"
|
|
input_source = tmp_path / "input-source"
|
|
workspace_source = tmp_path / "workspace-source"
|
|
(input_source / "input" / job_id).mkdir(parents=True)
|
|
(workspace_source / "workspace" / job_id / "final").mkdir(parents=True)
|
|
(project_root / "input").mkdir(parents=True)
|
|
(project_root / "workspace").mkdir(parents=True)
|
|
(project_root / "input" / job_id).symlink_to(
|
|
input_source / "input" / job_id
|
|
)
|
|
(project_root / "workspace" / job_id).symlink_to(
|
|
workspace_source / "workspace" / job_id
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="linked job paths do not share one project root"):
|
|
snapshot_module.build_confirmation_snapshot(project_root, job_id)
|