355 lines
12 KiB
Python
355 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from gyxx_flow.source_sync import SourceSyncService, SyncState
|
|
|
|
|
|
def _digest(content: str) -> str:
|
|
return hashlib.sha256(content.encode()).hexdigest()
|
|
|
|
|
|
def _write_project(
|
|
root: Path,
|
|
*,
|
|
source_content: str | None = "baseline\n",
|
|
target_content: str | None = "baseline\n",
|
|
baseline_source: str = "baseline\n",
|
|
baseline_target: str = "baseline\n",
|
|
transformed: bool = False,
|
|
source_relative_path: str = "job.py",
|
|
target_relative_path: str = "src/runtime/job.py",
|
|
) -> tuple[Path, Path, Path]:
|
|
source_root = root / "upstream"
|
|
repository_root = root / "repository"
|
|
manifest_path = repository_root / "config" / "source-manifests" / "demo.json"
|
|
config_path = repository_root / "config" / "source-projects.json"
|
|
|
|
source_root.mkdir(parents=True, exist_ok=True)
|
|
source_relative = Path(source_relative_path)
|
|
source_path_is_safe = (
|
|
not source_relative.is_absolute() and ".." not in source_relative.parts
|
|
)
|
|
target_relative = Path(target_relative_path)
|
|
target_path_is_safe = (
|
|
not target_relative.is_absolute() and ".." not in target_relative.parts
|
|
)
|
|
if source_content is not None and source_path_is_safe:
|
|
source_path = source_root / source_relative_path
|
|
source_path.parent.mkdir(parents=True, exist_ok=True)
|
|
source_path.write_text(source_content, encoding="utf-8", newline="")
|
|
if target_content is not None and target_path_is_safe:
|
|
target_path = repository_root / target_relative_path
|
|
target_path.parent.mkdir(parents=True, exist_ok=True)
|
|
target_path.write_text(target_content, encoding="utf-8", newline="")
|
|
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
manifest_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"module": "demo",
|
|
"files": [
|
|
{
|
|
"source_relative_path": source_relative_path,
|
|
"target_relative_path": target_relative_path,
|
|
"source_sha256": _digest(baseline_source),
|
|
"target_sha256": _digest(baseline_target),
|
|
"transformed": transformed,
|
|
"category": "source",
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
config_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"projects": [
|
|
{
|
|
"module": "demo",
|
|
"manifest": "config/source-manifests/demo.json",
|
|
"root_env": "GYXX_SOURCE_DEMO_ROOT",
|
|
"target_root": "src/runtime",
|
|
}
|
|
],
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return repository_root, source_root, config_path
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("source_content", "target_content", "expected"),
|
|
[
|
|
("baseline\n", "baseline\n", SyncState.IN_SYNC),
|
|
("upstream fix\n", "baseline\n", SyncState.SOURCE_CHANGED),
|
|
("baseline\n", "local fix\n", SyncState.TARGET_CHANGED),
|
|
("upstream fix\n", "local fix\n", SyncState.CONFLICT),
|
|
(None, "baseline\n", SyncState.SOURCE_MISSING),
|
|
("baseline\n", None, SyncState.TARGET_MISSING),
|
|
],
|
|
)
|
|
def test_status_uses_manifest_as_three_way_hash_baseline(
|
|
tmp_path: Path,
|
|
source_content: str | None,
|
|
target_content: str | None,
|
|
expected: SyncState,
|
|
) -> None:
|
|
repository_root, source_root, config_path = _write_project(
|
|
tmp_path,
|
|
source_content=source_content,
|
|
target_content=target_content,
|
|
)
|
|
|
|
service = SourceSyncService(
|
|
repository_root,
|
|
config_path=config_path,
|
|
source_roots={"demo": source_root},
|
|
)
|
|
|
|
comparison = service.status("demo")[0]
|
|
assert comparison.state is expected
|
|
assert comparison.safe_to_apply is (
|
|
expected is SyncState.SOURCE_CHANGED
|
|
)
|
|
|
|
|
|
def test_status_resolves_source_root_from_environment_without_writing(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
repository_root, source_root, config_path = _write_project(
|
|
tmp_path,
|
|
source_content="upstream fix\n",
|
|
)
|
|
manifest_before = (repository_root / "config/source-manifests/demo.json").read_bytes()
|
|
target_before = (repository_root / "src/runtime/job.py").read_bytes()
|
|
|
|
service = SourceSyncService(
|
|
repository_root,
|
|
config_path=config_path,
|
|
environ={"GYXX_SOURCE_DEMO_ROOT": str(source_root)},
|
|
)
|
|
|
|
assert service.status("demo")[0].state is SyncState.SOURCE_CHANGED
|
|
assert (repository_root / "config/source-manifests/demo.json").read_bytes() == manifest_before
|
|
assert (repository_root / "src/runtime/job.py").read_bytes() == target_before
|
|
|
|
|
|
def test_apply_safe_copies_source_and_atomically_refreshes_manifest(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
repository_root, source_root, config_path = _write_project(
|
|
tmp_path,
|
|
source_content="upstream fix\n",
|
|
)
|
|
manifest_path = repository_root / "config/source-manifests/demo.json"
|
|
replace_calls: list[tuple[Path, Path]] = []
|
|
|
|
import gyxx_flow.source_sync.service as service_module
|
|
|
|
real_replace = service_module.os.replace
|
|
|
|
def recording_replace(source: str | Path, target: str | Path) -> None:
|
|
replace_calls.append((Path(source), Path(target)))
|
|
real_replace(source, target)
|
|
|
|
monkeypatch.setattr(service_module.os, "replace", recording_replace)
|
|
service = SourceSyncService(
|
|
repository_root,
|
|
config_path=config_path,
|
|
source_roots={"demo": source_root},
|
|
)
|
|
|
|
result = service.apply_safe("demo")
|
|
|
|
assert result.applied == ("src/runtime/job.py",)
|
|
assert result.skipped == ()
|
|
assert (repository_root / "src/runtime/job.py").read_text(
|
|
encoding="utf-8"
|
|
) == "upstream fix\n"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
entry = manifest["files"][0]
|
|
assert entry["source_sha256"] == _digest("upstream fix\n")
|
|
assert entry["target_sha256"] == _digest("upstream fix\n")
|
|
assert any(target == manifest_path for _, target in replace_calls)
|
|
assert not list(manifest_path.parent.glob("*.tmp"))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("transformed", "target_content", "expected_state"),
|
|
[
|
|
(True, "baseline\n", SyncState.SOURCE_CHANGED),
|
|
(False, "local fix\n", SyncState.CONFLICT),
|
|
],
|
|
)
|
|
def test_apply_safe_does_not_overwrite_transformed_or_conflicting_targets(
|
|
tmp_path: Path,
|
|
transformed: bool,
|
|
target_content: str,
|
|
expected_state: SyncState,
|
|
) -> None:
|
|
repository_root, source_root, config_path = _write_project(
|
|
tmp_path,
|
|
source_content="upstream fix\n",
|
|
target_content=target_content,
|
|
transformed=transformed,
|
|
)
|
|
target_path = repository_root / "src/runtime/job.py"
|
|
manifest_path = repository_root / "config/source-manifests/demo.json"
|
|
manifest_before = manifest_path.read_bytes()
|
|
|
|
service = SourceSyncService(
|
|
repository_root,
|
|
config_path=config_path,
|
|
source_roots={"demo": source_root},
|
|
)
|
|
result = service.apply_safe("demo")
|
|
|
|
assert result.applied == ()
|
|
assert result.skipped == ("src/runtime/job.py",)
|
|
assert service.status("demo")[0].state is expected_state
|
|
assert target_path.read_text(encoding="utf-8") == target_content
|
|
assert manifest_path.read_bytes() == manifest_before
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("source_relative_path", "target_relative_path"),
|
|
[
|
|
("../outside.py", "src/runtime/job.py"),
|
|
("job.py", "../outside.py"),
|
|
("C:/outside.py", "src/runtime/job.py"),
|
|
("job.py", "C:/outside.py"),
|
|
],
|
|
)
|
|
def test_rejects_absolute_paths_and_parent_traversal(
|
|
tmp_path: Path,
|
|
source_relative_path: str,
|
|
target_relative_path: str,
|
|
) -> None:
|
|
repository_root, source_root, config_path = _write_project(
|
|
tmp_path,
|
|
source_relative_path=source_relative_path,
|
|
target_relative_path=target_relative_path,
|
|
)
|
|
service = SourceSyncService(
|
|
repository_root,
|
|
config_path=config_path,
|
|
source_roots={"demo": source_root},
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="unsafe relative path"):
|
|
service.status("demo")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"source_relative_path",
|
|
[".env", "credentials.json", "browser/cookies.json", "private.pem"],
|
|
)
|
|
def test_rejects_sensitive_source_files(
|
|
tmp_path: Path,
|
|
source_relative_path: str,
|
|
) -> None:
|
|
repository_root, source_root, config_path = _write_project(
|
|
tmp_path,
|
|
source_relative_path=source_relative_path,
|
|
)
|
|
service = SourceSyncService(
|
|
repository_root,
|
|
config_path=config_path,
|
|
source_roots={"demo": source_root},
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="sensitive"):
|
|
service.status("demo")
|
|
|
|
|
|
def test_status_reports_unmapped_source_files_but_not_excluded_or_sensitive_files(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
repository_root, source_root, config_path = _write_project(tmp_path)
|
|
manifest_path = repository_root / "config/source-manifests/demo.json"
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest["intentionally_excluded"] = [
|
|
{"pattern": "tmp_*.py", "reason": "temporary diagnostics"},
|
|
{
|
|
"source_relative_path": "inspect_manual.py",
|
|
"reason": "manual diagnostic",
|
|
},
|
|
]
|
|
manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
|
|
(source_root / "new_job.py").write_text("print('new')\n", encoding="utf-8")
|
|
(source_root / "tmp_probe.py").write_text("temporary\n", encoding="utf-8")
|
|
(source_root / "inspect_manual.py").write_text("manual\n", encoding="utf-8")
|
|
(source_root / ".env").write_text("TOKEN=secret\n", encoding="utf-8")
|
|
cache = source_root / "__pycache__"
|
|
cache.mkdir()
|
|
(cache / "job.pyc").write_bytes(b"cache")
|
|
root_profile = source_root / ".chrome_profile"
|
|
root_profile.mkdir()
|
|
(root_profile / "session.txt").write_text("browser state\n", encoding="utf-8")
|
|
root_logs = source_root / "logs"
|
|
root_logs.mkdir()
|
|
(root_logs / "run.log").write_text("generated log\n", encoding="utf-8")
|
|
|
|
comparisons = SourceSyncService(
|
|
repository_root,
|
|
config_path=config_path,
|
|
source_roots={"demo": source_root},
|
|
).status("demo")
|
|
|
|
assert [item.source_relative_path for item in comparisons] == [
|
|
"job.py",
|
|
"new_job.py",
|
|
]
|
|
candidate = comparisons[1]
|
|
assert candidate.state is SyncState.NEW_SOURCE
|
|
assert candidate.target_relative_path == "src/runtime/new_job.py"
|
|
assert candidate.safe_to_apply is False
|
|
|
|
|
|
def test_apply_safe_rolls_back_targets_when_manifest_commit_fails(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
repository_root, source_root, config_path = _write_project(
|
|
tmp_path,
|
|
source_content="upstream fix\n",
|
|
)
|
|
target_path = repository_root / "src/runtime/job.py"
|
|
manifest_path = repository_root / "config/source-manifests/demo.json"
|
|
original_target = target_path.read_bytes()
|
|
original_manifest = manifest_path.read_bytes()
|
|
|
|
import gyxx_flow.source_sync.service as service_module
|
|
|
|
def fail_manifest_commit(path: Path, payload: object) -> None:
|
|
del path, payload
|
|
raise OSError("injected manifest failure")
|
|
|
|
monkeypatch.setattr(
|
|
service_module,
|
|
"_atomic_write_json",
|
|
fail_manifest_commit,
|
|
)
|
|
service = SourceSyncService(
|
|
repository_root,
|
|
config_path=config_path,
|
|
source_roots={"demo": source_root},
|
|
)
|
|
|
|
with pytest.raises(OSError, match="injected manifest failure"):
|
|
service.apply_safe("demo")
|
|
|
|
assert target_path.read_bytes() == original_target
|
|
assert manifest_path.read_bytes() == original_manifest
|
|
assert not list(repository_root.rglob("*.tmp"))
|