gyxx-flow/tests/test_source_sync_cli.py

from __future__ import annotations

import hashlib
import io
import json
from pathlib import Path

from gyxx_flow.cli import EXIT_CONFIGURATION, EXIT_SUCCESS, main
from gyxx_flow.core.config import Settings
from gyxx_flow.source_sync import SourceSyncService


def _sha256(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def _project(tmp_path: Path) -> tuple[Settings, Path, Path]:
    repository = tmp_path / "repository"
    source = tmp_path / "source"
    target = repository / "src" / "module" / "job.py"
    manifest_path = repository / "config" / "source-manifests" / "module.json"
    source_file = source / "job.py"
    source_file.parent.mkdir(parents=True)
    target.parent.mkdir(parents=True)
    manifest_path.parent.mkdir(parents=True)
    old = b"old\n"
    new = b"new\n"
    source_file.write_bytes(new)
    target.write_bytes(old)
    manifest_path.write_text(
        json.dumps(
            {
                "schema_version": 1,
                "module": "module",
                "source_project": "legacy",
                "files": [
                    {
                        "source_relative_path": "job.py",
                        "target_relative_path": "src/module/job.py",
                        "category": "source",
                        "source_sha256": _sha256(old),
                        "target_sha256": _sha256(old),
                        "transformed": False,
                    }
                ],
                "intentionally_excluded": [],
                "snapshot": "test",
            }
        ),
        encoding="utf-8",
    )
    (repository / "config" / "source-projects.json").write_text(
        json.dumps(
            {
                "schema_version": 1,
                "projects": [
                    {
                        "module": "module",
                        "source_project": "legacy",
                        "manifest": "config/source-manifests/module.json",
                        "root_env": "MODULE_SOURCE_ROOT",
                        "target_root": "src/module",
                    }
                ],
            }
        ),
        encoding="utf-8",
    )
    return Settings(project_root=repository, data_root=tmp_path / "data"), source, target


def test_sources_status_is_read_only_and_does_not_build_workflow_registry(
    tmp_path: Path,
) -> None:
    settings, source, target = _project(tmp_path)
    output = io.StringIO()

    exit_code = main(
        [
            "sources",
            "status",
            "--project",
            "module",
            "--source-root",
            f"module={source}",
            "--json",
        ],
        settings=settings,
        stdout=output,
    )

    assert exit_code == EXIT_SUCCESS
    assert target.read_bytes() == b"old\n"
    payload = json.loads(output.getvalue())
    assert payload["summary"] == {"source_changed": 1}
    assert payload["files"][0]["state"] == "source_changed"


def test_sources_apply_requires_execute_before_writing(tmp_path: Path) -> None:
    settings, source, target = _project(tmp_path)
    errors = io.StringIO()

    exit_code = main(
        [
            "sources",
            "apply",
            "--project",
            "module",
            "--source-root",
            f"module={source}",
        ],
        settings=settings,
        stderr=errors,
    )

    assert exit_code == EXIT_CONFIGURATION
    assert "--execute" in errors.getvalue()
    assert target.read_bytes() == b"old\n"


def test_sources_apply_copies_only_after_explicit_execute(tmp_path: Path) -> None:
    settings, source, target = _project(tmp_path)
    output = io.StringIO()

    exit_code = main(
        [
            "sources",
            "apply",
            "--project",
            "module",
            "--source-root",
            f"module={source}",
            "--execute",
            "--json",
        ],
        settings=settings,
        stdout=output,
    )

    assert exit_code == EXIT_SUCCESS
    assert target.read_bytes() == b"new\n"
    payload = json.loads(output.getvalue())
    assert payload["applied"] == ["src/module/job.py"]


def test_sources_status_reports_new_unmapped_source_candidate(tmp_path: Path) -> None:
    settings, source, target = _project(tmp_path)
    del target
    (source / "new_job.py").write_text("new candidate\n", encoding="utf-8")
    output = io.StringIO()

    exit_code = main(
        [
            "sources",
            "status",
            "--project",
            "module",
            "--source-root",
            f"module={source}",
            "--json",
        ],
        settings=settings,
        stdout=output,
    )

    assert exit_code == EXIT_SUCCESS
    payload = json.loads(output.getvalue())
    assert payload["summary"] == {"new_source": 1, "source_changed": 1}
    candidate = next(
        item for item in payload["files"] if item["state"] == "new_source"
    )
    assert candidate["source_relative_path"] == "new_job.py"
    assert candidate["safe_to_apply"] is False


def test_repository_source_project_config_declares_migration_target_roots() -> None:
    repository = Path(__file__).parents[1]
    service = SourceSyncService(
        repository,
        source_roots={
            "content_marketing": repository,
            "product_commerce": repository,
            "shop_intelligence": repository,
            "supply_chain": repository,
        },
    )

    assert {
        project.module: project.target_root
        for project in service._projects
    } == {
        "content_marketing": "src/gyxx_flow/modules/content_marketing",
        "product_commerce": (
            "src/gyxx_flow/modules/product_commerce"
        ),
        "shop_intelligence": (
            "src/gyxx_flow/modules/shop_intelligence"
        ),
        "supply_chain": "src/gyxx_flow/modules/supply_chain",
    }