Jyotisha/tests/test_import_yinduzhanxing.py

from __future__ import annotations

import hashlib
import json
import subprocess
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "scripts"
if str(SCRIPTS) not in sys.path:
    sys.path.insert(0, str(SCRIPTS))

import import_yinduzhanxing as importer  # noqa: E402

POLICY = ROOT / "references/cross_project_contract/sync_policy.v2.json"
ARCHIVE_COMMIT = "5db72537741fcedaa7b5498502d4a31b0f9fc147"


def _git(root: Path, *args: str) -> str:
    return subprocess.check_output(["git", "-C", str(root), *args], text=True).strip()


def _target(tmp_path: Path) -> Path:
    target = tmp_path / "commercial"
    target.mkdir()
    subprocess.run(["git", "init", "-q", str(target)], check=True)
    subprocess.run(["git", "-C", str(target), "config", "user.email", "tests@example.invalid"], check=True)
    subprocess.run(["git", "-C", str(target), "config", "user.name", "Tests"], check=True)
    (target / "README.md").write_text("commercial\n", encoding="utf-8")
    subprocess.run(["git", "-C", str(target), "add", "README.md"], check=True)
    subprocess.run(["git", "-C", str(target), "commit", "-qm", "base"], check=True)
    return target


def _source(tmp_path: Path) -> Path:
    source = tmp_path / "snapshot"
    (source / "scripts").mkdir(parents=True)
    (source / "references").mkdir()
    (source / "SKILL.md").write_text("# Research skill\n", encoding="utf-8")
    (source / "AGENTS.md").write_text("research agent\n", encoding="utf-8")
    (source / "references/strict-workflow-router.md").write_text("router\n", encoding="utf-8")
    (source / "scripts/unified_consultation_orchestrator.py").write_text("# source\n", encoding="utf-8")
    (source / "scripts/report_orchestrator.py").write_text("# report\n", encoding="utf-8")
    (source / "scripts/jyotish_api_server.py").write_text("# api\n", encoding="utf-8")
    return source


def _git_source(tmp_path: Path) -> Path:
    source = _source(tmp_path)
    subprocess.run(["git", "init", "-q", str(source)], check=True)
    subprocess.run(["git", "-C", str(source), "config", "user.email", "tests@example.invalid"], check=True)
    subprocess.run(["git", "-C", str(source), "config", "user.name", "Tests"], check=True)
    subprocess.run(
        ["git", "-C", str(source), "remote", "add", "origin", "https://github.com/732642856/yinduzhanxing.git"],
        check=True,
    )
    subprocess.run(["git", "-C", str(source), "add", "."], check=True)
    subprocess.run(["git", "-C", str(source), "commit", "-qm", "research base"], check=True)
    return source


def _policy(
    tmp_path: Path,
    *,
    mirror: list[dict[str, str]] | None = None,
    protected: list[str] | None = None,
) -> Path:
    value = json.loads(POLICY.read_text(encoding="utf-8"))
    if mirror is not None:
        value["modes"]["mirror"] = mirror
    if protected is not None:
        value["modes"]["protected"] = protected
    path = tmp_path / "policy.json"
    path.write_text(json.dumps(value), encoding="utf-8")
    return path


def test_v2_policy_is_strictly_one_way_and_has_no_reverse_gate() -> None:
    policy = importer.load_policy(POLICY)
    assert policy["direction"] == "research_to_commercial_only"
    assert policy["reverse_sync"] == "forbidden"
    assert policy["source_repository_url"] == "https://github.com/732642856/yinduzhanxing"
    assert "commercial_to_research" not in json.dumps(policy, sort_keys=True)
    assert policy["modes"]["protected"]


def test_policy_cannot_remove_mandatory_commercial_protections(tmp_path: Path) -> None:
    weakened = sorted(importer.REQUIRED_PROTECTED_PATTERNS - {"frontend/**"})
    policy = _policy(tmp_path, protected=weakened)
    with pytest.raises(importer.ImportRejected, match="required_protected_patterns_missing:frontend/\\*\\*"):
        importer.load_policy(policy)


def _archive_kwargs(source: Path) -> dict[str, str]:
    return {
        "source_mode": "archive",
        "source_commit": ARCHIVE_COMMIT,
        "expected_tree_sha256": importer.source_tree_hash(source),
    }


def test_dry_run_writes_no_target_file_and_records_bound_archive_identity(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    before = _git(target, "status", "--porcelain=v1")
    manifest = importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))
    after = _git(target, "status", "--porcelain=v1")

    assert before == after == ""
    assert manifest["source_mode"] == "archive"
    assert manifest["source_commit"] == ARCHIVE_COMMIT
    assert manifest["source_tree_hash"] == importer.source_tree_hash(source)
    assert manifest["mirror_files"][0]["status"] == "new"
    assert not (target / "references/upstream/yinduzhanxing/SKILL.md").exists()


@pytest.mark.parametrize(
    ("source_commit", "expected_tree_sha256", "error"),
    [
        (ARCHIVE_COMMIT, None, "archive_expected_tree_sha256_required"),
        (None, "0" * 64, "archive_source_commit_required"),
        (ARCHIVE_COMMIT, "0" * 64, "source_tree_sha256_mismatch"),
    ],
)
def test_archive_identity_requires_commit_and_matching_tree_hash(
    tmp_path: Path, source_commit: str | None, expected_tree_sha256: str | None, error: str
) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    with pytest.raises(importer.ImportRejected, match=error):
        importer.build_manifest(
            source,
            policy_path=POLICY,
            target=target,
            source_mode="archive",
            source_commit=source_commit,
            expected_tree_sha256=expected_tree_sha256,
        )




@pytest.mark.parametrize(
    ("source_commit", "expected_tree_sha256"),
    [(ARCHIVE_COMMIT, None), (None, "0" * 64)],
)
def test_explicit_provenance_pins_must_be_provided_together_for_git_sources(
    tmp_path: Path, source_commit: str | None, expected_tree_sha256: str | None
) -> None:
    source, target = _git_source(tmp_path), _target(tmp_path)
    with pytest.raises(importer.ImportRejected, match="source_commit_and_expected_tree_sha256_must_be_provided_together"):
        importer.build_manifest(
            source,
            policy_path=POLICY,
            target=target,
            source_mode="git",
            source_commit=source_commit,
            expected_tree_sha256=expected_tree_sha256,
        )


def test_unbound_non_git_source_no_longer_emits_unknown_snapshot(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    with pytest.raises(importer.ImportRejected, match="archive_source_commit_required"):
        importer.build_manifest(source, policy_path=POLICY, target=target)


def test_git_source_must_be_clean_and_is_bound_to_head(tmp_path: Path) -> None:
    source, target = _git_source(tmp_path), _target(tmp_path)
    manifest = importer.build_manifest(source, policy_path=POLICY, target=target)
    assert manifest["source_mode"] == "git"
    assert manifest["source_commit"] == _git(source, "rev-parse", "HEAD")

    (source / "SKILL.md").write_text("# Dirty tracked research skill\n", encoding="utf-8")
    with pytest.raises(importer.ImportRejected, match="git_source_must_be_clean"):
        importer.build_manifest(source, policy_path=POLICY, target=target)

    subprocess.run(["git", "-C", str(source), "restore", "SKILL.md"], check=True)
    (source / "untracked.txt").write_text("not committed\n", encoding="utf-8")
    with pytest.raises(importer.ImportRejected, match="git_source_must_be_clean"):
        importer.build_manifest(source, policy_path=POLICY, target=target)


def test_git_source_cannot_be_forced_through_archive_mode(tmp_path: Path) -> None:
    source, target = _git_source(tmp_path), _target(tmp_path)
    with pytest.raises(importer.ImportRejected, match="archive_source_must_not_be_git_checkout"):
        importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))


def test_git_source_origin_must_match_policy_repository(tmp_path: Path) -> None:
    source, target = _git_source(tmp_path), _target(tmp_path)
    subprocess.run(
        ["git", "-C", str(source), "remote", "set-url", "origin", "https://github.com/example/not-the-upstream.git"],
        check=True,
    )
    with pytest.raises(importer.ImportRejected, match="git_source_repository_mismatch"):
        importer.build_manifest(source, policy_path=POLICY, target=target)

    subprocess.run(
        ["git", "-C", str(source), "remote", "set-url", "origin", "https://evil.example/732642856/yinduzhanxing.git"],
        check=True,
    )
    with pytest.raises(importer.ImportRejected, match="git_source_repository_mismatch"):
        importer.build_manifest(source, policy_path=POLICY, target=target)

    subprocess.run(
        ["git", "-C", str(source), "remote", "set-url", "origin", "https://github.com/attacker/732642856/yinduzhanxing.git"],
        check=True,
    )
    with pytest.raises(importer.ImportRejected, match="git_source_repository_mismatch"):
        importer.build_manifest(source, policy_path=POLICY, target=target)


def test_manifest_output_cannot_overwrite_commercial_repository_files(tmp_path: Path) -> None:
    protected = ROOT / "SKILL.md"
    before = protected.read_bytes()
    with pytest.raises(importer.ImportRejected, match="manifest_output_must_be_external_or_import_record_json"):
        importer.validate_output_path(protected)
    assert protected.read_bytes() == before

    external = tmp_path / "manifest.json"
    assert importer.validate_output_path(external) == external.resolve()
    internal = ROOT / "references/cross_project_contract/imports/test-output.json"
    assert importer.validate_output_path(internal) == internal.resolve()

    source, target = _source(tmp_path), _target(tmp_path)
    with pytest.raises(importer.ImportRejected, match="manifest_output_must_not_overlap_source"):
        importer.validate_output_path(source / "manifest.json", target=target, source=source)
    with pytest.raises(importer.ImportRejected, match="manifest_output_must_be_json"):
        importer.validate_output_path(tmp_path / "manifest.txt", target=target, source=source)


def test_fixed_archive_inputs_build_byte_equivalent_manifests(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    kwargs = _archive_kwargs(source)

    recorded_tests = ["tests/test_upstream_import_plan.py", "tests/test_import_yinduzhanxing.py", "tests/test_import_yinduzhanxing.py"]
    first = importer.build_manifest(source, policy_path=POLICY, target=target, tests_run=recorded_tests, **kwargs)
    second = importer.build_manifest(source, policy_path=POLICY, target=target, tests_run=list(reversed(recorded_tests)), **kwargs)

    assert first == second
    assert json.dumps(first, ensure_ascii=False, sort_keys=True) == json.dumps(second, ensure_ascii=False, sort_keys=True)
    assert first["source_skill_sha256"] == hashlib.sha256((source / "SKILL.md").read_bytes()).hexdigest()
    assert first["tests_run"] == ["tests/test_import_yinduzhanxing.py", "tests/test_upstream_import_plan.py"]


def test_apply_handles_new_update_and_unchanged_with_recomputable_hashes(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    destination = target / "references/upstream/yinduzhanxing/SKILL.md"

    first = importer.build_manifest(source, policy_path=POLICY, target=target, apply=True, **_archive_kwargs(source))
    assert first["mirror_files"][0]["status"] == "applied"
    assert destination.read_bytes() == (source / "SKILL.md").read_bytes()

    subprocess.run(["git", "-C", str(target), "add", "references/upstream/yinduzhanxing/SKILL.md"], check=True)
    subprocess.run(["git", "-C", str(target), "commit", "-qm", "record mirrored skill"], check=True)

    unchanged = importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))
    assert unchanged["mirror_files"][0]["status"] == "unchanged"
    assert unchanged["mirror_files"][0]["source_sha256"] == hashlib.sha256(destination.read_bytes()).hexdigest()

    (source / "SKILL.md").write_text("# Updated research skill\n", encoding="utf-8")
    update = importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))
    assert update["mirror_files"][0]["status"] == "update"
    assert update["mirror_files"][0]["target_sha256_before"] != update["mirror_files"][0]["target_sha256_after"]


def test_semantic_files_never_overwrite_target(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    (target / "SKILL.md").write_text("commercial skill\n", encoding="utf-8")
    subprocess.run(["git", "-C", str(target), "add", "SKILL.md"], check=True)
    subprocess.run(["git", "-C", str(target), "commit", "-qm", "add commercial skill"], check=True)
    before = (target / "SKILL.md").read_bytes()

    manifest = importer.build_manifest(source, policy_path=POLICY, target=target, apply=True, **_archive_kwargs(source))

    assert (target / "SKILL.md").read_bytes() == before
    row = next(item for item in manifest["semantic_merge_files"] if item["path"] == "SKILL.md")
    assert row["status"] == "review_required"
    assert manifest["operator_review_required"] is True


def test_protected_mapping_is_rejected_instead_of_silently_skipped(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    policy = _policy(tmp_path, mirror=[{"source": "SKILL.md", "target": "frontend/src/stolen.ts", "license": "MIT"}])
    with pytest.raises(importer.ImportRejected, match="protected_rejected"):
        importer.build_manifest(source, policy_path=policy, target=target, apply=True, **_archive_kwargs(source))
    assert not (target / "frontend/src/stolen.ts").exists()


@pytest.mark.parametrize(
    "target_path",
    [
        "SKILL.md",
        "skills/jyotish-vedic-astrology/SKILL.md",
        "skills/jyotish-birth-time-rectification/SKILL.md",
        "skills/jyotish-birth-time-rectification/references/private.md",
    ],
)
def test_commercial_skill_surfaces_cannot_be_mirror_targets(tmp_path: Path, target_path: str) -> None:
    policy = _policy(tmp_path, mirror=[{"source": "SKILL.md", "target": target_path, "license": "MIT"}])
    with pytest.raises(importer.ImportRejected, match="protected_rejected"):
        importer.load_policy(policy)


@pytest.mark.parametrize("path", ["../escape", "/absolute/path", "safe/../../escape"])
def test_path_traversal_and_absolute_paths_are_rejected(tmp_path: Path, path: str) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    policy = _policy(tmp_path, mirror=[{"source": "SKILL.md", "target": path, "license": "MIT"}])
    with pytest.raises(importer.ImportRejected, match="unsafe_path"):
        importer.build_manifest(source, policy_path=policy, target=target, **_archive_kwargs(source))


def test_symlink_escape_is_rejected(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    outside = tmp_path / "outside.md"
    outside.write_text("outside\n", encoding="utf-8")
    (source / "linked.md").symlink_to(outside)
    policy = _policy(tmp_path, mirror=[{"source": "linked.md", "target": "references/upstream/linked.md", "license": "MIT"}])
    with pytest.raises(importer.ImportRejected, match="source_tree_symlink_rejected|symlink_escape|non_regular_or_symlink"):
        importer.build_manifest(source, policy_path=policy, target=target, **_archive_kwargs(source))


def test_target_parent_symlink_cannot_redirect_mirror_into_protected_surface(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    (target / "frontend").mkdir()
    (target / "frontend/.keep").write_text("protected\n", encoding="utf-8")
    (target / "references/upstream").mkdir(parents=True)
    (target / "references/upstream/yinduzhanxing").symlink_to("../../frontend")
    subprocess.run(["git", "-C", str(target), "add", "frontend", "references/upstream/yinduzhanxing"], check=True)
    subprocess.run(["git", "-C", str(target), "commit", "-qm", "add redirecting symlink"], check=True)

    with pytest.raises(importer.ImportRejected, match="symlink_component_rejected"):
        importer.build_manifest(source, policy_path=POLICY, target=target, apply=True, **_archive_kwargs(source))
    assert not (target / "frontend/SKILL.md").exists()


def test_target_worktree_must_be_clean_for_deterministic_manifest(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    (target / "untracked.txt").write_text("dirty target\n", encoding="utf-8")
    with pytest.raises(importer.ImportRejected, match="target_must_be_clean"):
        importer.build_manifest(source, policy_path=POLICY, target=target, **_archive_kwargs(source))


def test_sensitive_source_is_rejected(tmp_path: Path) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    (source / ".env").write_text("OPENAI_API_KEY=not-a-real-key\n", encoding="utf-8")
    policy = _policy(tmp_path, mirror=[{"source": ".env", "target": "references/upstream/env.txt", "license": "MIT"}])
    with pytest.raises(importer.ImportRejected, match="sensitive_filename|sensitive_content"):
        importer.build_manifest(source, policy_path=policy, target=target, **_archive_kwargs(source))


def test_overlapping_source_and_target_are_rejected(tmp_path: Path) -> None:
    target = _target(tmp_path)
    source = target / "source"
    source.mkdir()
    with pytest.raises(importer.ImportRejected, match="must_not_overlap"):
        importer.build_manifest(source, policy_path=POLICY, target=target)


def test_apply_rolls_back_earlier_files_when_later_replace_fails(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    source, target = _source(tmp_path), _target(tmp_path)
    (source / "second.md").write_text("second\n", encoding="utf-8")
    policy = _policy(tmp_path, mirror=[
        {"source": "SKILL.md", "target": "references/upstream/one.md", "license": "MIT"},
        {"source": "second.md", "target": "references/upstream/two.md", "license": "MIT"},
    ])
    original_replace = importer.os.replace
    calls = 0

    def fail_second(source_path: str, target_path: Path) -> None:
        nonlocal calls
        calls += 1
        if calls == 2:
            raise OSError("simulated replace failure")
        original_replace(source_path, target_path)

    monkeypatch.setattr(importer.os, "replace", fail_second)
    with pytest.raises(OSError, match="simulated"):
        importer.build_manifest(source, policy_path=policy, target=target, apply=True, **_archive_kwargs(source))
    assert not (target / "references/upstream/one.md").exists()
    assert not (target / "references/upstream/two.md").exists()