145 lines
4.8 KiB
Python
145 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gyxx_flow.core.layout import DataLayout
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
RUNTIME_ROOT = PROJECT_ROOT / "src" / "gyxx_flow" / "modules"
|
|
|
|
|
|
def _load_path_module(module: str, filename: str, monkeypatch, data_root: Path):
|
|
monkeypatch.setenv("GYXX_DATA_ROOT", str(data_root))
|
|
module_root = RUNTIME_ROOT / module
|
|
monkeypatch.setenv("GYXX_MODULE_ROOT", str(module_root))
|
|
path = module_root / filename
|
|
name = f"_gyxx_test_paths_{module}"
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec and spec.loader
|
|
loaded = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = loaded
|
|
try:
|
|
spec.loader.exec_module(loaded)
|
|
finally:
|
|
sys.modules.pop(name, None)
|
|
return loaded
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("module", "filename", "names"),
|
|
[
|
|
(
|
|
"content_marketing",
|
|
"runtime_paths.py",
|
|
("raw_root", "normalized_root", "curated_root", "exports_root"),
|
|
),
|
|
(
|
|
"product_commerce",
|
|
"runtime_paths.py",
|
|
(
|
|
"RAW_DATA_ROOT",
|
|
"NORMALIZED_DATA_ROOT",
|
|
"CURATED_DATA_ROOT",
|
|
"EXPORTS_DATA_ROOT",
|
|
),
|
|
),
|
|
(
|
|
"shop_intelligence",
|
|
"paths.py",
|
|
("RAW_ROOT", "NORMALIZED_ROOT", "CURATED_ROOT", "EXPORTS_ROOT"),
|
|
),
|
|
(
|
|
"supply_chain",
|
|
"paths.py",
|
|
("SHARED_ROOT", "NORMALIZED_ROOT", "CURATED_ROOT", "BACKUP_ROOT"),
|
|
),
|
|
],
|
|
)
|
|
def test_all_module_path_boundaries_use_the_canonical_data_tree(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
module: str,
|
|
filename: str,
|
|
names: tuple[str, str, str, str],
|
|
) -> None:
|
|
data_root = tmp_path / "relocatable-data"
|
|
loaded = _load_path_module(module, filename, monkeypatch, data_root)
|
|
|
|
layers = ("raw", "normalized", "curated", "exports")
|
|
for name, layer in zip(names, layers, strict=True):
|
|
assert getattr(loaded, name) == data_root / "data" / layer / module
|
|
assert not data_root.exists(), "path resolution must not mutate the filesystem"
|
|
|
|
|
|
def test_module_data_paths_cover_collected_and_exported_file_types(tmp_path: Path) -> None:
|
|
paths = DataLayout(tmp_path).for_module("content_marketing")
|
|
|
|
assert paths.raw_path("api", "response.json") == (
|
|
tmp_path / "data" / "raw" / "content_marketing" / "api" / "response.json"
|
|
)
|
|
assert paths.raw_path("downloads", "source.xlsx") == (
|
|
tmp_path / "data" / "raw" / "content_marketing" / "downloads" / "source.xlsx"
|
|
)
|
|
assert paths.raw_path("downloads", "source.xls") == (
|
|
tmp_path / "data" / "raw" / "content_marketing" / "downloads" / "source.xls"
|
|
)
|
|
assert paths.raw_path("downloads", "source.csv") == (
|
|
tmp_path / "data" / "raw" / "content_marketing" / "downloads" / "source.csv"
|
|
)
|
|
assert paths.export_path("reports", "weekly.md") == (
|
|
tmp_path / "data" / "exports" / "content_marketing" / "reports" / "weekly.md"
|
|
)
|
|
assert list(tmp_path.iterdir()) == []
|
|
|
|
|
|
@pytest.mark.parametrize("unsafe", ["", "..", "a/b", "a\\b", "C:\\data"])
|
|
def test_module_data_paths_reject_unsafe_module_names(tmp_path: Path, unsafe: str) -> None:
|
|
with pytest.raises(ValueError, match="path segment"):
|
|
DataLayout(tmp_path).for_module(unsafe)
|
|
|
|
|
|
@pytest.mark.parametrize("unsafe", [("..", "escape.json"), ("C:\\", "escape.json")])
|
|
def test_module_data_paths_reject_paths_outside_the_layer(
|
|
tmp_path: Path, unsafe: tuple[str, str]
|
|
) -> None:
|
|
with pytest.raises(ValueError, match="escapes module data root"):
|
|
DataLayout(tmp_path).for_module("product_commerce").raw_path(*unsafe)
|
|
|
|
|
|
@pytest.mark.parametrize("configured", ["", " "])
|
|
@pytest.mark.parametrize(
|
|
("module", "filename", "root_name"),
|
|
[
|
|
("content_marketing", "runtime_paths.py", "data_root"),
|
|
("product_commerce", "runtime_paths.py", "DATA_HOME"),
|
|
("shop_intelligence", "paths.py", "DATA_ROOT"),
|
|
("supply_chain", "paths.py", "DATA_ROOT"),
|
|
],
|
|
)
|
|
def test_blank_data_root_uses_the_project_var_default(
|
|
monkeypatch,
|
|
configured: str,
|
|
module: str,
|
|
filename: str,
|
|
root_name: str,
|
|
) -> None:
|
|
monkeypatch.setenv("GYXX_DATA_ROOT", configured)
|
|
module_root = RUNTIME_ROOT / module
|
|
monkeypatch.setenv("GYXX_MODULE_ROOT", str(module_root))
|
|
path = module_root / filename
|
|
name = f"_gyxx_test_blank_paths_{module}"
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
assert spec and spec.loader
|
|
loaded = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = loaded
|
|
try:
|
|
spec.loader.exec_module(loaded)
|
|
finally:
|
|
sys.modules.pop(name, None)
|
|
|
|
assert getattr(loaded, root_name) == PROJECT_ROOT / "var"
|