277 lines
9.0 KiB
Python
277 lines
9.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gyxx_flow.baseline import (
|
|
DataRootSpec,
|
|
PowerShellScheduledTaskProvider,
|
|
ProjectSpec,
|
|
ScheduledTask,
|
|
TaskCountMismatch,
|
|
build_baseline_manifest,
|
|
collect_code_inventory,
|
|
collect_data_summary,
|
|
collect_scheduled_task_inventory,
|
|
project_specs_from_env,
|
|
write_manifest_atomic,
|
|
)
|
|
|
|
|
|
class FakeTaskProvider:
|
|
def __init__(self, tasks: list[ScheduledTask]) -> None:
|
|
self._tasks = tasks
|
|
|
|
def scheduled_tasks(self) -> list[ScheduledTask]:
|
|
return self._tasks
|
|
|
|
|
|
def test_powershell_provider_parses_tasks_without_using_a_shell(monkeypatch) -> None:
|
|
captured: dict[str, object] = {}
|
|
|
|
def fake_run(command, **kwargs):
|
|
captured["command"] = command
|
|
captured["kwargs"] = kwargs
|
|
return subprocess.CompletedProcess(
|
|
command,
|
|
0,
|
|
stdout='[{"task_id":"\\\\Daily","command":"python.exe",'
|
|
'"arguments":"D:\\\\legacy\\\\run.py","working_directory":"D:\\\\legacy"}]',
|
|
stderr="",
|
|
)
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
|
|
tasks = PowerShellScheduledTaskProvider().scheduled_tasks()
|
|
|
|
assert tasks == [
|
|
ScheduledTask(
|
|
task_id=r"\Daily",
|
|
command="python.exe",
|
|
arguments=r"D:\legacy\run.py",
|
|
working_directory=r"D:\legacy",
|
|
)
|
|
]
|
|
assert captured["command"][0] == "powershell.exe"
|
|
assert captured["kwargs"]["shell"] is False
|
|
assert captured["kwargs"]["check"] is True
|
|
|
|
|
|
def test_project_specs_are_built_only_from_injected_legacy_roots(tmp_path: Path) -> None:
|
|
roots = {
|
|
"GYXX_LEGACY_CONTENT_ROOT": tmp_path / "content",
|
|
"GYXX_LEGACY_SHOP_ROOT": tmp_path / "shop",
|
|
"GYXX_LEGACY_PRODUCT_ROOT": tmp_path / "product",
|
|
"GYXX_LEGACY_SUPPLY_ROOT": tmp_path / "supply",
|
|
}
|
|
specs = project_specs_from_env({key: str(value) for key, value in roots.items()})
|
|
|
|
assert [spec.project_id for spec in specs] == [
|
|
"content_marketing",
|
|
"shop_intelligence",
|
|
"product_commerce",
|
|
"supply_chain",
|
|
]
|
|
assert [spec.root for spec in specs] == list(roots.values())
|
|
assert [item.label for item in specs[0].data_roots] == ["data", "reports"]
|
|
assert specs[3].data_roots[0].path == roots["GYXX_LEGACY_SUPPLY_ROOT"] / "shared-data"
|
|
|
|
|
|
def test_project_specs_report_all_missing_root_variables() -> None:
|
|
with pytest.raises(ValueError, match="GYXX_LEGACY_SHOP_ROOT.*GYXX_LEGACY_PRODUCT_ROOT"):
|
|
project_specs_from_env({"GYXX_LEGACY_CONTENT_ROOT": "configured"})
|
|
|
|
|
|
def test_code_inventory_hashes_allowed_files_and_excludes_runtime_and_secrets(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
root = tmp_path / "legacy"
|
|
(root / "src").mkdir(parents=True)
|
|
(root / "src" / "worker.py").write_text("print('safe')\n", encoding="utf-8")
|
|
(root / "README.md").write_text("docs\n", encoding="utf-8")
|
|
(root / "payload.csv").write_text("private,data\n", encoding="utf-8")
|
|
|
|
excluded_files = [
|
|
root / ".git" / "tracked.py",
|
|
root / ".venv" / "site.py",
|
|
root / "var" / "runtime.py",
|
|
root / "data" / "raw.py",
|
|
root / "logs" / "run.py",
|
|
root / ".chrome_profile" / "state.py",
|
|
root / "cookies" / "session.py",
|
|
root / "config" / ".env",
|
|
root / "config" / "credentials.json",
|
|
]
|
|
for path in excluded_files:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("TOP_SECRET=must-not-leak", encoding="utf-8")
|
|
|
|
inventory = collect_code_inventory(root)
|
|
|
|
assert [item["relative_path"] for item in inventory] == [
|
|
"README.md",
|
|
"src/worker.py",
|
|
]
|
|
worker = inventory[1]
|
|
worker_bytes = (root / "src" / "worker.py").read_bytes()
|
|
assert set(worker) == {"relative_path", "size_bytes", "mtime_ns", "sha256"}
|
|
assert worker["size_bytes"] == len(worker_bytes)
|
|
assert worker["sha256"] == hashlib.sha256(worker_bytes).hexdigest()
|
|
serialized = json.dumps(inventory)
|
|
assert "TOP_SECRET" not in serialized
|
|
assert "private,data" not in serialized
|
|
|
|
|
|
def test_data_summary_contains_only_aggregate_counts_bytes_extensions_and_top_level(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
root = tmp_path / "data"
|
|
root.mkdir()
|
|
(root / "root.csv").write_bytes(b"123")
|
|
(root / "orders").mkdir()
|
|
(root / "orders" / "one.json").write_bytes(b"12")
|
|
(root / "orders" / "archive").mkdir()
|
|
(root / "orders" / "archive" / "two.JSON").write_bytes(b"1234")
|
|
|
|
summary = collect_data_summary(root)
|
|
|
|
assert summary == {
|
|
"file_count": 3,
|
|
"total_bytes": 9,
|
|
"by_extension": {
|
|
".csv": {"file_count": 1, "total_bytes": 3},
|
|
".json": {"file_count": 2, "total_bytes": 6},
|
|
},
|
|
"by_top_level": {
|
|
".": {"file_count": 1, "total_bytes": 3},
|
|
"orders": {"file_count": 2, "total_bytes": 6},
|
|
},
|
|
}
|
|
serialized = json.dumps(summary)
|
|
assert "root.csv" not in serialized
|
|
assert "one.json" not in serialized
|
|
assert "archive" not in serialized
|
|
|
|
|
|
def _task_specs(tmp_path: Path) -> tuple[ProjectSpec, ...]:
|
|
roots = [tmp_path / name for name in ("one", "two", "three", "four")]
|
|
for root in roots:
|
|
root.mkdir()
|
|
return tuple(
|
|
ProjectSpec(project_id=f"project_{index}", root=root)
|
|
for index, root in enumerate(roots, start=1)
|
|
)
|
|
|
|
|
|
def test_task_inventory_filters_by_legacy_roots_and_exposes_no_command_values(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
specs = _task_specs(tmp_path)
|
|
tasks = [
|
|
ScheduledTask(
|
|
task_id=f"legacy-{index:02d}",
|
|
command="python.exe",
|
|
arguments=f'"{specs[index % 4].root / "run.py"}" --token SECRET-{index}',
|
|
)
|
|
for index in range(21)
|
|
]
|
|
tasks.append(
|
|
ScheduledTask(
|
|
task_id="unrelated",
|
|
command="python.exe",
|
|
arguments=r"C:\unrelated\run.py --token OUTSIDE_SECRET",
|
|
)
|
|
)
|
|
tasks.append(
|
|
ScheduledTask(
|
|
task_id="lookalike-root",
|
|
command="python.exe",
|
|
arguments=f'"{specs[0].root}-backup\\run.py"',
|
|
)
|
|
)
|
|
|
|
result = collect_scheduled_task_inventory(
|
|
specs, FakeTaskProvider(tasks), expected_count=21
|
|
)
|
|
|
|
assert result["expected_count"] == 21
|
|
assert result["actual_count"] == 21
|
|
assert result["is_complete"] is True
|
|
assert len(result["tasks"]) == 21
|
|
assert set(result["tasks"][0]) == {"task_id", "project_id"}
|
|
serialized = json.dumps(result)
|
|
assert "SECRET" not in serialized
|
|
assert "unrelated" not in serialized
|
|
assert "lookalike-root" not in serialized
|
|
|
|
|
|
def test_task_inventory_rejects_any_count_other_than_exactly_21(tmp_path: Path) -> None:
|
|
specs = _task_specs(tmp_path)
|
|
tasks = [
|
|
ScheduledTask(task_id=f"task-{index}", command=str(specs[0].root / "run.py"))
|
|
for index in range(20)
|
|
]
|
|
|
|
with pytest.raises(TaskCountMismatch, match="expected 21.*found 20"):
|
|
collect_scheduled_task_inventory(
|
|
specs, FakeTaskProvider(tasks), expected_count=21
|
|
)
|
|
|
|
|
|
def test_build_manifest_has_code_data_and_exact_task_baselines_without_values(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
root = tmp_path / "legacy"
|
|
data = root / "shared-data"
|
|
data.mkdir(parents=True)
|
|
credential_key = "PASS" + "WORD"
|
|
credential_value = "not-" + "exported"
|
|
(root / "job.py").write_text(
|
|
f"{credential_key} = '{credential_value}'\n", encoding="utf-8"
|
|
)
|
|
(data / "dataset.json").write_text('{"token":"hidden"}', encoding="utf-8")
|
|
spec = ProjectSpec(
|
|
project_id="project",
|
|
root=root,
|
|
data_roots=(DataRootSpec(label="data", path=data),),
|
|
)
|
|
tasks = [
|
|
ScheduledTask(
|
|
task_id=f"task-{index:02d}",
|
|
command=str(root / "job.py"),
|
|
arguments=f"--password secret-{index}",
|
|
)
|
|
for index in range(21)
|
|
]
|
|
|
|
manifest = build_baseline_manifest(
|
|
(spec,),
|
|
FakeTaskProvider(tasks),
|
|
generated_at=datetime(2026, 7, 27, 5, 0, tzinfo=timezone.utc),
|
|
)
|
|
|
|
assert manifest["schema_version"] == 1
|
|
assert manifest["generated_at"] == "2026-07-27T05:00:00+00:00"
|
|
assert manifest["scheduled_tasks"]["actual_count"] == 21
|
|
assert len(manifest["projects"][0]["code_inventory"]) == 1
|
|
assert manifest["projects"][0]["code_inventory"][0]["relative_path"] == "job.py"
|
|
assert manifest["projects"][0]["data_summaries"][0]["summary"]["file_count"] == 1
|
|
serialized = json.dumps(manifest)
|
|
assert credential_value not in serialized
|
|
assert "dataset.json" not in serialized
|
|
assert "secret-0" not in serialized
|
|
|
|
|
|
def test_manifest_write_is_atomic_and_leaves_no_temporary_file(tmp_path: Path) -> None:
|
|
target = tmp_path / "baseline" / "manifest.json"
|
|
write_manifest_atomic(target, {"version": 1})
|
|
write_manifest_atomic(target, {"version": 2})
|
|
|
|
assert json.loads(target.read_text(encoding="utf-8")) == {"version": 2}
|
|
assert list(target.parent.glob(".*.tmp")) == []
|