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"


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 _policy(tmp_path: Path, *, mirror: list[dict[str, str]] | None = None) -> Path:
    value = json.loads(POLICY.read_text(encoding="utf-8"))
    if mirror is not None:
        value["modes"]["mirror"] = mirror
    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 "commercial_to_research" not in json.dumps(policy, sort_keys=True)
    assert policy["modes"]["protected"]


def test_dry_run_writes_no_target_file_and_snapshot_commit_is_unknown(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)
    after = _git(target, "status", "--porcelain=v1")

    assert before == after == ""
    assert manifest["source_mode"] == "snapshot"
    assert manifest["source_commit"] == "unknown"
    assert manifest["mirror_files"][0]["status"] == "new"
    assert not (target / "references/upstream/yinduzhanxing/SKILL.md").exists()


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)
    assert first["mirror_files"][0]["status"] == "applied"
    assert destination.read_bytes() == (source / "SKILL.md").read_bytes()

    unchanged = importer.build_manifest(source, policy_path=POLICY, target=target)
    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)
    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")
    before = (target / "SKILL.md").read_bytes()

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

    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)
    assert not (target / "frontend/src/stolen.ts").exists()


@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)


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="symlink_escape|non_regular_or_symlink"):
        importer.build_manifest(source, policy_path=policy, target=target)


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)


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)
    assert not (target / "references/upstream/one.md").exists()
    assert not (target / "references/upstream/two.md").exists()