225 lines
9.2 KiB
Python
225 lines
9.2 KiB
Python
"""Integration tests: full collection cycles against real git repositories.
|
|
|
|
Covers the Section 40 processing sequence and the safety/edge-case list
|
|
from Section 41.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from git_auto_tracker.application import ApplicationController
|
|
from git_auto_tracker.config import AppConfig
|
|
from git_auto_tracker.errors import (
|
|
ExitCode,
|
|
RepositoryError,
|
|
SecurityError,
|
|
TrackerError,
|
|
)
|
|
from git_auto_tracker.logging.setup import setup_logging
|
|
|
|
|
|
def _make_controller(config_dict: dict, tmp_path: Path) -> tuple[AppConfig, ApplicationController]:
|
|
config = AppConfig.from_dict(config_dict, config_dir=tmp_path)
|
|
setup_logging(config.logging, config_dir=tmp_path, repository_path=config.repository_path)
|
|
return config, ApplicationController(config)
|
|
|
|
|
|
def _git_out(*args: str, cwd: Path) -> str:
|
|
"""Run git; NUL-terminated output (-z) is returned as decoded text."""
|
|
proc = subprocess.run(
|
|
["git", *args],
|
|
cwd=str(cwd),
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
return proc.stdout.decode("utf-8")
|
|
|
|
|
|
def _commit_log(repo: Path) -> list[str]:
|
|
out = _git_out("log", "--format=%s", cwd=repo)
|
|
return out.splitlines()
|
|
|
|
|
|
def _run(config_dict: dict, tmp_path: Path, dry_run: bool = False):
|
|
_, controller = _make_controller(config_dict, tmp_path)
|
|
return controller.collect(dry_run=dry_run)
|
|
|
|
|
|
class TestRunCycle:
|
|
def test_clean_repository_no_commit(
|
|
self, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
_, controller = _make_controller(tracker_config, tmp_path)
|
|
outcome = controller.collect()
|
|
assert outcome.commit_hash is None
|
|
assert outcome.change_count == 0
|
|
|
|
def test_add_modify_delete_in_one_commit(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
stations = git_repo / "Stations" / "PS101"
|
|
stations.mkdir(parents=True)
|
|
(stations / "pump2.xml").write_text("<pump/>", encoding="utf-8")
|
|
(git_repo / "README.md").write_text("changed\n", encoding="utf-8")
|
|
(git_repo / "obsolete.xml").write_text("bye", encoding="utf-8")
|
|
_git_out("add", "obsolete.xml", cwd=git_repo)
|
|
_git_out("commit", "-m", "add obsolete", cwd=git_repo)
|
|
(git_repo / "obsolete.xml").unlink()
|
|
|
|
outcome = _run(tracker_config, tmp_path)
|
|
assert outcome.commit_hash is not None
|
|
assert outcome.change_count == 3
|
|
|
|
subjects = _commit_log(git_repo)
|
|
assert subjects[0].startswith("SCADA automatic change capture - ")
|
|
|
|
body = _git_out("log", "-1", "--format=%b", cwd=git_repo)
|
|
assert "Stations/PS101/pump2.xml" in body
|
|
assert "README.md" in body
|
|
assert "obsolete.xml" in body
|
|
assert "Tracker-Run: " in body
|
|
assert "Added: 1" in body
|
|
assert "Modified: 1" in body
|
|
assert "Deleted: 1" in body
|
|
|
|
# Author of the commit is the tracker identity (Section 17)
|
|
author = _git_out("log", "-1", "--format=%an <%ae>", cwd=git_repo).strip()
|
|
assert author == "SCADA Change Tracker <git-auto-tracker@localhost>"
|
|
|
|
def test_sidecar_attribution_appears_in_commit(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
(git_repo / "config.xml").write_text("<c/>", encoding="utf-8")
|
|
(git_repo / "config.xml.metadata").write_text(
|
|
'last_modified_by=DOMAIN\\jsmith\n', encoding="utf-8"
|
|
)
|
|
_run(tracker_config, tmp_path)
|
|
body = _git_out("log", "-1", "--format=%b", cwd=git_repo)
|
|
assert "Last modified by: DOMAIN\\jsmith" in body
|
|
|
|
def test_file_mtime_appears_in_commit_local_time(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
"""Section 12 file_timestamp rendered as 'Modified at' (local time)."""
|
|
import re as re_mod
|
|
|
|
(git_repo / "config.xml").write_text("<c/>", encoding="utf-8")
|
|
_run(tracker_config, tmp_path)
|
|
body = _git_out("log", "-1", "--format=%b", cwd=git_repo)
|
|
# Local-time format with explicit +/-HH:MM offset (Section 15)
|
|
assert re_mod.search(
|
|
r"Modified at: \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[+-]\d{2}:\d{2}\b", body
|
|
)
|
|
# The timestamp block sits directly under the changed file entry
|
|
assert " config.xml\n Last modified by: UNKNOWN\n Modified at:" in body
|
|
|
|
def test_excluded_files_not_committed(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
(git_repo / "junk.log").write_text("noise", encoding="utf-8")
|
|
(git_repo / "real.xml").write_text("<r/>", encoding="utf-8")
|
|
outcome = _run(tracker_config, tmp_path)
|
|
assert outcome.change_count == 1
|
|
status = _git_out("status", "--porcelain", cwd=git_repo)
|
|
assert "junk.log" in status # still untracked, left alone
|
|
assert "real.xml" not in status
|
|
|
|
def test_dry_run_leaves_repo_untouched(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
(git_repo / "new.xml").write_text("<n/>", encoding="utf-8")
|
|
before = _git_out("status", "--porcelain", "-z", cwd=git_repo)
|
|
outcome = _run(tracker_config, tmp_path, dry_run=True)
|
|
assert outcome.dry_run
|
|
assert outcome.commit_message is not None
|
|
assert "new.xml" in outcome.commit_message
|
|
after = _git_out("status", "--porcelain", "-z", cwd=git_repo)
|
|
assert before == after
|
|
assert len(_commit_log(git_repo)) == 1 # only the initial commit
|
|
|
|
def test_unicode_and_spaces(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
tricky = git_repo / "Station PS101"
|
|
tricky.mkdir()
|
|
(tricky / "räumliche Datei.xml").write_text("<u/>", encoding="utf-8")
|
|
outcome = _run(tracker_config, tmp_path)
|
|
assert outcome.change_count == 1
|
|
# ls-files -z output is NUL-separated raw UTF-8 (never C-quoted)
|
|
raw = _git_out("ls-files", "-z", cwd=git_repo)
|
|
files = [part for part in raw.split("\x00") if part]
|
|
assert "Station PS101/räumliche Datei.xml" in files
|
|
|
|
|
|
class TestSafetyAborts:
|
|
def test_pre_existing_staged_changes_abort(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
(git_repo / "staged.xml").write_text("<s/>", encoding="utf-8")
|
|
_git_out("add", "staged.xml", cwd=git_repo)
|
|
with pytest.raises(RepositoryError, match="staged changes"):
|
|
_run(tracker_config, tmp_path)
|
|
|
|
def test_unfinished_merge_state_aborts(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
(git_repo / "conflict.txt").write_text("base", encoding="utf-8")
|
|
_git_out("add", "conflict.txt", cwd=git_repo)
|
|
_git_out("commit", "-m", "base", cwd=git_repo)
|
|
_git_out("checkout", "-b", "feature", cwd=git_repo)
|
|
(git_repo / "conflict.txt").write_text("feature", encoding="utf-8")
|
|
_git_out("commit", "-am", "feature", cwd=git_repo)
|
|
_git_out("checkout", "main", cwd=git_repo)
|
|
(git_repo / "conflict.txt").write_text("main", encoding="utf-8")
|
|
_git_out("commit", "-am", "main", cwd=git_repo)
|
|
with contextlib.suppress(subprocess.CalledProcessError):
|
|
_git_out("merge", "feature", cwd=git_repo) # expected conflict
|
|
with pytest.raises(RepositoryError, match="MERGE_HEAD"):
|
|
_run(tracker_config, tmp_path)
|
|
def test_wrong_branch_aborts(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
_git_out("checkout", "-b", "other-branch", cwd=git_repo)
|
|
with pytest.raises(RepositoryError, match="configured branch"):
|
|
_run(tracker_config, tmp_path)
|
|
|
|
def test_forbidden_file_aborts_with_security_code(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
(git_repo / "server.key").write_text("PRIVATE", encoding="utf-8")
|
|
with pytest.raises(SecurityError):
|
|
_run(tracker_config, tmp_path)
|
|
# Exit code mapping (Section 25)
|
|
assert int(SecurityError("x").exit_code) == ExitCode.SECURITY_VIOLATION
|
|
# Nothing was committed
|
|
assert len(_commit_log(git_repo)) == 1
|
|
|
|
|
|
class TestLockIntegration:
|
|
def test_concurrent_run_blocked(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
from git_auto_tracker.locking.repository_lock import RepositoryLock, default_lock_path
|
|
|
|
_, controller = _make_controller(tracker_config, tmp_path)
|
|
lock = RepositoryLock(default_lock_path(controller.config.repository_path))
|
|
lock.acquire()
|
|
# A second execution attempts the same lock via the CLI path:
|
|
second = RepositoryLock(default_lock_path(controller.config.repository_path))
|
|
with pytest.raises(TrackerError):
|
|
second.acquire()
|
|
lock.release()
|
|
|
|
def test_lock_file_never_in_status(
|
|
self, git_repo: Path, tracker_config: dict, tmp_path: Path
|
|
) -> None:
|
|
(git_repo / "change.xml").write_text("<c/>", encoding="utf-8")
|
|
_run(tracker_config, tmp_path)
|
|
status = _git_out("status", "--porcelain", "--untracked-files=all", cwd=git_repo)
|
|
assert "git-auto-tracker.lock" not in status
|