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")) == []