🚧 WIP
This commit is contained in:
Jonas Bengtsson
@@ -0,0 +1,72 @@
|
||||
"""Shared fixtures: temporary git repositories and tracker configuration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def _git(*args: str, cwd: Path) -> None:
|
||||
subprocess.run(
|
||||
["git", *args],
|
||||
cwd=str(cwd),
|
||||
check=True,
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def git_repo(tmp_path: Path) -> Path:
|
||||
"""A fresh git repository with one initial commit on 'main'."""
|
||||
repo = tmp_path / "project"
|
||||
repo.mkdir()
|
||||
_git("init", "-b", "main", cwd=repo)
|
||||
_git("config", "user.name", "Test User", cwd=repo)
|
||||
_git("config", "user.email", "test@example.com", cwd=repo)
|
||||
(repo / "README.md").write_text("initial\n", encoding="utf-8")
|
||||
_git("add", "README.md", cwd=repo)
|
||||
_git("commit", "-m", "initial", cwd=repo)
|
||||
return repo
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tracker_config(git_repo: Path, tmp_path: Path) -> dict:
|
||||
"""Tracker configuration dict pointing at the fixture repository."""
|
||||
return {
|
||||
"application": {"timezone": "Europe/Stockholm"},
|
||||
"repository": {"path": str(git_repo), "branch": "main"},
|
||||
"git": {
|
||||
"user_name": "SCADA Change Tracker",
|
||||
"user_email": "git-auto-tracker@localhost",
|
||||
},
|
||||
"files": {
|
||||
"include": ["**/*"],
|
||||
"exclude": ["**/*.tmp", "**/*.log", ".git/**"],
|
||||
},
|
||||
"security": {
|
||||
"forbidden_patterns": ["**/*.key", "**/*.pem", "**/secrets.*", "**/credentials.*"],
|
||||
},
|
||||
"author_resolution": {
|
||||
"resolvers": ["sidecar_metadata"],
|
||||
"minimum_confidence": "probable",
|
||||
"fallback": "unknown",
|
||||
},
|
||||
"logging": {"level": "INFO", "console": False, "file": None},
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def write_config(tmp_path: Path) -> object:
|
||||
"""Factory: write a config dict to YAML and return its path."""
|
||||
|
||||
import yaml
|
||||
|
||||
class _Writer:
|
||||
def __call__(self, config: dict, name: str = "git-auto-tracker.yaml") -> Path:
|
||||
path = tmp_path / name
|
||||
path.write_text(yaml.safe_dump(config), encoding="utf-8")
|
||||
return path
|
||||
|
||||
return _Writer()
|
||||
@@ -0,0 +1,208 @@
|
||||
"""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_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
|
||||
@@ -0,0 +1,141 @@
|
||||
"""Unit tests for the sidecar resolver and the resolver chain (Sections 9-11)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from git_auto_tracker.authors.base import AuthorResult
|
||||
from git_auto_tracker.authors.resolver import ResolverChain
|
||||
from git_auto_tracker.authors.sidecar import SidecarMetadataResolver
|
||||
from git_auto_tracker.changes.models import Confidence
|
||||
from git_auto_tracker.config import AuthorResolutionConfig
|
||||
from git_auto_tracker.errors import ConfigurationError
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def resolver() -> SidecarMetadataResolver:
|
||||
return SidecarMetadataResolver()
|
||||
|
||||
|
||||
class TestSidecarResolver:
|
||||
def test_json_sidecar(self, tmp_path: Path, resolver: SidecarMetadataResolver) -> None:
|
||||
(tmp_path / "pump.xml").write_text("<x/>", encoding="utf-8")
|
||||
(tmp_path / "pump.xml.metadata").write_text(
|
||||
'{"last_modified_by": "DOMAIN\\\\jsmith",'
|
||||
' "name": "John Smith",'
|
||||
' "last_modified_at": "2026-09-11T14:31:22+02:00"}',
|
||||
encoding="utf-8",
|
||||
)
|
||||
result = resolver.resolve(tmp_path, "pump.xml")
|
||||
assert result.identifier == "DOMAIN\\jsmith"
|
||||
assert result.name == "John Smith"
|
||||
assert result.confidence is Confidence.PROBABLE
|
||||
expected = dt.datetime(
|
||||
2026, 9, 11, 14, 31, 22, tzinfo=dt.timezone(dt.timedelta(hours=2))
|
||||
)
|
||||
assert result.timestamp == expected
|
||||
|
||||
def test_keyvalue_sidecar(self, tmp_path: Path, resolver: SidecarMetadataResolver) -> None:
|
||||
(tmp_path / "pump.xml.metadata").write_text(
|
||||
"last_modified_by=DOMAIN\\adoe\n", encoding="utf-8"
|
||||
)
|
||||
result = resolver.resolve(tmp_path, "pump.xml")
|
||||
assert result.identifier == "DOMAIN\\adoe"
|
||||
|
||||
def test_missing_sidecar_is_unknown(
|
||||
self, tmp_path: Path, resolver: SidecarMetadataResolver
|
||||
) -> None:
|
||||
result = resolver.resolve(tmp_path, "absent.xml")
|
||||
assert result.identifier == "UNKNOWN"
|
||||
assert result.confidence is Confidence.UNKNOWN
|
||||
|
||||
def test_malformed_sidecar_is_unknown(
|
||||
self, tmp_path: Path, resolver: SidecarMetadataResolver
|
||||
) -> None:
|
||||
(tmp_path / "bad.xml.metadata").write_text("{not json", encoding="utf-8")
|
||||
result = resolver.resolve(tmp_path, "bad.xml")
|
||||
assert result.identifier == "UNKNOWN"
|
||||
|
||||
def test_sidecar_without_author_is_unknown(
|
||||
self, tmp_path: Path, resolver: SidecarMetadataResolver
|
||||
) -> None:
|
||||
(tmp_path / "partial.xml.metadata").write_text('{"revision": 7}', encoding="utf-8")
|
||||
result = resolver.resolve(tmp_path, "partial.xml")
|
||||
assert result.identifier == "UNKNOWN"
|
||||
|
||||
|
||||
class _FakeResolver:
|
||||
"""Test resolver with configurable results, registered under a name."""
|
||||
|
||||
def __init__(self, name: str, result: AuthorResult) -> None:
|
||||
self.name = name
|
||||
self._result = result
|
||||
self.calls = 0
|
||||
|
||||
def resolve(self, repository_root: Path, file_path: str) -> AuthorResult:
|
||||
self.calls += 1
|
||||
return self._result
|
||||
|
||||
|
||||
class TestResolverChain:
|
||||
def _chain(self, resolvers: list, minimum: str = "probable") -> ResolverChain:
|
||||
chain = ResolverChain(
|
||||
[r for r in resolvers],
|
||||
Confidence(minimum),
|
||||
AuthorResult.unknown("fallback"),
|
||||
)
|
||||
return chain
|
||||
|
||||
def test_first_usable_wins(self, tmp_path: Path) -> None:
|
||||
weak = _FakeResolver(
|
||||
"a", AuthorResult("DOMAIN\\weak", confidence=Confidence.WEAK, source="a")
|
||||
)
|
||||
strong = _FakeResolver(
|
||||
"b", AuthorResult("DOMAIN\\strong", confidence=Confidence.PROBABLE, source="b")
|
||||
)
|
||||
chain = self._chain([weak, strong])
|
||||
result = chain.resolve(tmp_path, "f.xml")
|
||||
assert result.identifier == "DOMAIN\\strong"
|
||||
assert weak.calls == 1
|
||||
assert strong.calls == 1 # weak skipped, strong consulted
|
||||
|
||||
def test_below_threshold_results_are_not_used(self, tmp_path: Path) -> None:
|
||||
weak = _FakeResolver(
|
||||
"a", AuthorResult("DOMAIN\\weak", confidence=Confidence.WEAK, source="a")
|
||||
)
|
||||
chain = self._chain([weak], minimum="probable")
|
||||
result = chain.resolve(tmp_path, "f.xml")
|
||||
assert result.identifier == "UNKNOWN" # fallback
|
||||
|
||||
def test_exception_in_resolver_continues_chain(self, tmp_path: Path) -> None:
|
||||
class _Boom:
|
||||
name = "boom"
|
||||
|
||||
def resolve(self, repository_root: Path, file_path: str) -> AuthorResult:
|
||||
raise RuntimeError("audit log unavailable")
|
||||
|
||||
good = _FakeResolver(
|
||||
"b", AuthorResult("DOMAIN\\ok", confidence=Confidence.AUTHORITATIVE, source="b")
|
||||
)
|
||||
chain = self._chain([_Boom(), good])
|
||||
result = chain.resolve(tmp_path, "f.xml")
|
||||
assert result.identifier == "DOMAIN\\ok"
|
||||
|
||||
def test_fallback_when_all_unknown(self, tmp_path: Path) -> None:
|
||||
unknown = _FakeResolver("a", AuthorResult.unknown("a"))
|
||||
chain = self._chain([unknown])
|
||||
result = chain.resolve(tmp_path, "f.xml")
|
||||
assert result.identifier == "UNKNOWN"
|
||||
|
||||
def test_unknown_resolver_name_rejected(self) -> None:
|
||||
config = AuthorResolutionConfig(resolvers=["nope"], minimum_confidence="probable")
|
||||
with pytest.raises(ConfigurationError):
|
||||
ResolverChain.from_config(config)
|
||||
|
||||
def test_sidecar_config_name(self) -> None:
|
||||
config = AuthorResolutionConfig(resolvers=["sidecar_metadata"])
|
||||
chain = ResolverChain.from_config(config)
|
||||
assert isinstance(chain._resolvers[0], SidecarMetadataResolver)
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Unit tests for glob filters and security validation (Sections 7, 33)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from git_auto_tracker.changes.filters import (
|
||||
find_forbidden,
|
||||
is_excluded,
|
||||
is_included,
|
||||
matches,
|
||||
validate_security,
|
||||
)
|
||||
from git_auto_tracker.changes.models import ChangeRecord, ChangeSet, ChangeType
|
||||
from git_auto_tracker.config import FileFilters, SecurityConfig
|
||||
from git_auto_tracker.errors import SecurityError
|
||||
|
||||
|
||||
class TestMatches:
|
||||
@pytest.mark.parametrize(
|
||||
("path", "pattern", "expected"),
|
||||
[
|
||||
("a.tmp", "**/*.tmp", True),
|
||||
("Stations/PS101/data.tmp", "**/*.tmp", True),
|
||||
("Stations/PS101/config.xml", "**/*.tmp", False),
|
||||
(".git/config", ".git/**", True),
|
||||
(".git", ".git/**", True),
|
||||
("runtime/cache.dat", "runtime/**", True),
|
||||
("other/cache.dat", "runtime/**", False),
|
||||
("Stations/deep/nested/alarm.xml", "Stations/**", True),
|
||||
("System/global.xml", "**/global.xml", True),
|
||||
("Stations/PS101/global.xml", "Stations/**/global.xml", True),
|
||||
("global.xml", "Stations/**/global.xml", False),
|
||||
("spaces in name.xml", "**/*.xml", True),
|
||||
],
|
||||
)
|
||||
def test_glob_semantics(self, path: str, pattern: str, expected: bool) -> None:
|
||||
assert matches(path, pattern) is expected
|
||||
|
||||
|
||||
class TestIncludeExclude:
|
||||
def test_include_all_by_default(self) -> None:
|
||||
filters = FileFilters(include=[], exclude=[])
|
||||
assert is_included("anything.xml", filters)
|
||||
|
||||
def test_positive_scope(self) -> None:
|
||||
filters = FileFilters(include=["Stations/**"], exclude=[])
|
||||
assert is_included("Stations/PS1.xml", filters)
|
||||
assert not is_included("System/global.xml", filters)
|
||||
|
||||
def test_exclude_wins(self) -> None:
|
||||
filters = FileFilters(include=["**/*"], exclude=["**/*.log"])
|
||||
assert is_included("a.xml", filters)
|
||||
assert not is_excluded("a.xml", filters)
|
||||
assert is_included("a.log", filters)
|
||||
assert is_excluded("a.log", filters)
|
||||
|
||||
|
||||
class TestSecurity:
|
||||
def test_forbidden_detected(self) -> None:
|
||||
security = SecurityConfig(forbidden_patterns=["**/*.key"])
|
||||
change_set = ChangeSet(
|
||||
added=[ChangeRecord(path="certs/server.key", change_type=ChangeType.ADDED)]
|
||||
)
|
||||
assert find_forbidden(["certs/server.key"], security) == ["certs/server.key"]
|
||||
with pytest.raises(SecurityError):
|
||||
validate_security(change_set, security)
|
||||
|
||||
def test_no_forbidden_passes(self) -> None:
|
||||
security = SecurityConfig(forbidden_patterns=["**/*.key"])
|
||||
change_set = ChangeSet(
|
||||
modified=[ChangeRecord(path="config.xml", change_type=ChangeType.MODIFIED)]
|
||||
)
|
||||
validate_security(change_set, security) # no raise
|
||||
|
||||
def test_rename_destination_checked(self) -> None:
|
||||
security = SecurityConfig(forbidden_patterns=["**/secret.pem"])
|
||||
change_set = ChangeSet(
|
||||
renamed=[
|
||||
ChangeRecord(
|
||||
path="secret.pem",
|
||||
change_type=ChangeType.RENAMED,
|
||||
previous_path="plain.txt",
|
||||
)
|
||||
]
|
||||
)
|
||||
with pytest.raises(SecurityError):
|
||||
validate_security(change_set, security)
|
||||
@@ -0,0 +1,206 @@
|
||||
"""Unit tests for the repository lock, config, ULID and commit message."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as dt
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from git_auto_tracker.audit import new_ulid
|
||||
from git_auto_tracker.changes.models import (
|
||||
AuthorIdentity,
|
||||
ChangeRecord,
|
||||
ChangeSet,
|
||||
ChangeType,
|
||||
Confidence,
|
||||
TrackerRun,
|
||||
)
|
||||
from git_auto_tracker.commits.message_builder import build_commit_message
|
||||
from git_auto_tracker.config import load_config, redact
|
||||
from git_auto_tracker.errors import ConfigurationError, LockError
|
||||
from git_auto_tracker.locking.repository_lock import RepositoryLock, default_lock_path
|
||||
|
||||
|
||||
class TestULID:
|
||||
def test_format(self) -> None:
|
||||
ulid = new_ulid()
|
||||
assert len(ulid) == 26
|
||||
assert all(ch in "0123456789ABCDEFGHJKMNPQRSTVWXYZ" for ch in ulid)
|
||||
|
||||
def test_monotonic_within_same_ms(self) -> None:
|
||||
moment = dt.datetime(2026, 9, 11, 12, 0, 0, tzinfo=dt.UTC)
|
||||
assert new_ulid(moment) != new_ulid(moment)
|
||||
|
||||
def test_sortable(self) -> None:
|
||||
earlier = new_ulid(dt.datetime(2026, 1, 1, tzinfo=dt.UTC))
|
||||
later = new_ulid(dt.datetime(2027, 1, 1, tzinfo=dt.UTC))
|
||||
assert earlier < later
|
||||
|
||||
|
||||
class TestLock:
|
||||
def test_lock_is_sibling_outside_repo(self, tmp_path: Path) -> None:
|
||||
repo = tmp_path / "project"
|
||||
repo.mkdir()
|
||||
lock_path = default_lock_path(repo)
|
||||
assert lock_path.parent == repo.parent
|
||||
assert lock_path.name == "project.git-auto-tracker.lock"
|
||||
|
||||
def test_acquire_and_release(self, tmp_path: Path) -> None:
|
||||
lock = RepositoryLock(tmp_path / "x.lock")
|
||||
lock.acquire()
|
||||
assert (tmp_path / "x.lock").exists()
|
||||
lock.release()
|
||||
assert not (tmp_path / "x.lock").exists()
|
||||
lock.release() # idempotent
|
||||
|
||||
def test_second_acquire_refused(self, tmp_path: Path) -> None:
|
||||
first = RepositoryLock(tmp_path / "x.lock")
|
||||
first.acquire()
|
||||
second = RepositoryLock(tmp_path / "x.lock")
|
||||
with pytest.raises(LockError):
|
||||
second.acquire()
|
||||
|
||||
def test_stale_lock_from_dead_pid_removed(self, tmp_path: Path) -> None:
|
||||
stale = tmp_path / "x.lock"
|
||||
stale.write_text(
|
||||
'{"process_id": 999999999, "hostname": "definitely-not-this-host",'
|
||||
' "start_time": "2020-01-01T00:00:00+00:00"}',
|
||||
encoding="utf-8",
|
||||
)
|
||||
lock = RepositoryLock(stale)
|
||||
lock.acquire() # stale -> recovered
|
||||
assert stale.exists() # now ours
|
||||
|
||||
def test_young_foreign_lock_not_stolen(self, tmp_path: Path) -> None:
|
||||
young = tmp_path / "x.lock"
|
||||
young.write_text(
|
||||
'{"process_id": 999999999, "hostname": "definitely-not-this-host",'
|
||||
' "start_time": "' + dt.datetime.now(dt.UTC).isoformat() + '"}',
|
||||
encoding="utf-8",
|
||||
)
|
||||
lock = RepositoryLock(young)
|
||||
with pytest.raises(LockError):
|
||||
lock.acquire()
|
||||
|
||||
def test_corrupt_lock_recovered(self, tmp_path: Path) -> None:
|
||||
corrupt = tmp_path / "x.lock"
|
||||
corrupt.write_text("not json at all", encoding="utf-8")
|
||||
lock = RepositoryLock(corrupt)
|
||||
lock.acquire()
|
||||
assert corrupt.exists()
|
||||
|
||||
|
||||
class TestConfig:
|
||||
def test_full_config_loads(self, tmp_path: Path, tracker_config: dict) -> None:
|
||||
import yaml
|
||||
|
||||
path = tmp_path / "cfg.yaml"
|
||||
path.write_text(yaml.safe_dump(tracker_config), encoding="utf-8")
|
||||
config = load_config(path)
|
||||
assert config.repository_path == Path(tracker_config["repository"]["path"])
|
||||
assert config.branch == "main"
|
||||
assert config.author_resolution.minimum_confidence == "probable"
|
||||
|
||||
def test_missing_file(self, tmp_path: Path) -> None:
|
||||
with pytest.raises(ConfigurationError):
|
||||
load_config(tmp_path / "absent.yaml")
|
||||
|
||||
def test_missing_repository_key(self, tmp_path: Path) -> None:
|
||||
path = tmp_path / "cfg.yaml"
|
||||
path.write_text("application: {}\n", encoding="utf-8")
|
||||
with pytest.raises(ConfigurationError):
|
||||
load_config(path)
|
||||
|
||||
def test_env_override(self, tmp_path: Path, tracker_config: dict) -> None:
|
||||
import yaml
|
||||
|
||||
path = tmp_path / "cfg.yaml"
|
||||
path.write_text(yaml.safe_dump(tracker_config), encoding="utf-8")
|
||||
other = tmp_path / "other-project"
|
||||
other.mkdir()
|
||||
config = load_config(
|
||||
path,
|
||||
environ={"GIT_AUTO_TRACKER_REPOSITORY__PATH": str(other)},
|
||||
)
|
||||
assert config.repository_path == other
|
||||
|
||||
def test_redact(self) -> None:
|
||||
data = {
|
||||
"git": {"user_email": "x@y.z", "user_name": "n"},
|
||||
"remote": {"password": "hunter2"},
|
||||
"files": {"include": ["**/*"]},
|
||||
}
|
||||
redacted = redact(data)
|
||||
assert redacted["git"]["user_email"] == "<redacted>"
|
||||
assert redacted["remote"]["password"] == "<redacted>"
|
||||
assert redacted["files"]["include"] == ["**/*"]
|
||||
|
||||
|
||||
class TestCommitMessage:
|
||||
def _run(self) -> TrackerRun:
|
||||
return TrackerRun(
|
||||
run_id="01J7K8HYV84ZKFH1J4T9R8DZQN",
|
||||
started_at=dt.datetime(2026, 9, 11, 18, 0, 3, tzinfo=dt.UTC),
|
||||
host="SCADA-SRV-01",
|
||||
)
|
||||
|
||||
def test_full_message_structure(self) -> None:
|
||||
change_set = ChangeSet(
|
||||
added=[ChangeRecord(path="Stations/PS101/pump2.xml", change_type=ChangeType.ADDED)],
|
||||
modified=[
|
||||
ChangeRecord(
|
||||
path="Stations/PS101/project.xml",
|
||||
change_type=ChangeType.MODIFIED,
|
||||
author=AuthorIdentity(
|
||||
identifier="DOMAIN\\jsmith",
|
||||
source="scada_audit_log",
|
||||
confidence=Confidence.AUTHORITATIVE,
|
||||
timestamp=dt.datetime(2026, 9, 11, 15, 42, 9, tzinfo=dt.UTC),
|
||||
),
|
||||
),
|
||||
ChangeRecord(
|
||||
path="System/global.xml",
|
||||
change_type=ChangeType.MODIFIED,
|
||||
author=AuthorIdentity.unknown(),
|
||||
),
|
||||
],
|
||||
deleted=[ChangeRecord(path="Stations/PS027/old.xml", change_type=ChangeType.DELETED)],
|
||||
)
|
||||
message = build_commit_message(change_set, self._run())
|
||||
lines = message.splitlines()
|
||||
|
||||
assert lines[0].startswith("SCADA automatic change capture - ")
|
||||
assert "Summary:" in lines
|
||||
assert "Added: 1" in message
|
||||
assert "Last modified by: DOMAIN\\jsmith" in message
|
||||
assert "Last modified by: UNKNOWN" in message
|
||||
assert "Stations/PS027/old.xml" in message
|
||||
# Trailers last (Section 36)
|
||||
assert "Tracker-Version: 1.0.0" in message
|
||||
assert "Tracker-Run: 01J7K8HYV84ZKFH1J4T9R8DZQN" in message
|
||||
assert "Tracker-Host: SCADA-SRV-01" in message
|
||||
assert "Tracker-Changes: 4" in message
|
||||
trailers_start = message.index("Tracker-Version:")
|
||||
body = message[:trailers_start]
|
||||
assert "Tracker-Run" not in body
|
||||
|
||||
def test_deleted_files_get_no_author_block(self) -> None:
|
||||
change_set = ChangeSet(
|
||||
deleted=[ChangeRecord(path="gone.xml", change_type=ChangeType.DELETED)]
|
||||
)
|
||||
message = build_commit_message(change_set, self._run())
|
||||
idx = message.index("gone.xml")
|
||||
following = message[idx:].splitlines()
|
||||
assert not any("Last modified by" in line for line in following[1:3])
|
||||
|
||||
def test_rename_rendered_with_arrow(self) -> None:
|
||||
change_set = ChangeSet(
|
||||
renamed=[
|
||||
ChangeRecord(
|
||||
path="new.xml", change_type=ChangeType.RENAMED, previous_path="old.xml"
|
||||
)
|
||||
]
|
||||
)
|
||||
message = build_commit_message(change_set, self._run())
|
||||
assert "old.xml -> new.xml" in message
|
||||
@@ -0,0 +1,72 @@
|
||||
"""Unit tests for porcelain parsing and ChangeType mapping (Section 6)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from git_auto_tracker.changes.models import ChangeType
|
||||
from git_auto_tracker.errors import GitError
|
||||
from git_auto_tracker.git.status import RawStatusEntry, parse_porcelain_z
|
||||
|
||||
|
||||
def _build(entries: list[tuple[str, str, str | None]]) -> bytes:
|
||||
"""Serialize porcelain -z records: 'XY path' + optional origin path."""
|
||||
parts: list[bytes] = []
|
||||
for xy, path, previous in entries:
|
||||
parts.append(f"{xy} {path}".encode())
|
||||
if previous:
|
||||
parts.append(previous.encode())
|
||||
return b"\x00".join(parts) + b"\x00"
|
||||
|
||||
|
||||
class TestParsePorcelainZ:
|
||||
def test_modified(self) -> None:
|
||||
raw = _build([(" M", "config.xml", None)])
|
||||
(entry,) = parse_porcelain_z(raw)
|
||||
assert entry.path == "config.xml"
|
||||
assert entry.change_type is ChangeType.MODIFIED
|
||||
|
||||
def test_untracked_is_added(self) -> None:
|
||||
raw = _build([("??", "new.xml", None)])
|
||||
(entry,) = parse_porcelain_z(raw)
|
||||
assert entry.change_type is ChangeType.ADDED
|
||||
|
||||
def test_deleted(self) -> None:
|
||||
raw = _build([(" D", "gone.xml", None)])
|
||||
(entry,) = parse_porcelain_z(raw)
|
||||
assert entry.change_type is ChangeType.DELETED
|
||||
|
||||
def test_staged_deletion(self) -> None:
|
||||
raw = _build([("D ", "gone.xml", None)])
|
||||
(entry,) = parse_porcelain_z(raw)
|
||||
assert entry.change_type is ChangeType.DELETED
|
||||
|
||||
def test_rename_with_previous(self) -> None:
|
||||
raw = _build([("R ", "new.xml", "old.xml")])
|
||||
(entry,) = parse_porcelain_z(raw)
|
||||
assert entry.change_type is ChangeType.RENAMED
|
||||
assert entry.previous_path == "old.xml"
|
||||
|
||||
def test_filenames_with_spaces_and_unicode(self) -> None:
|
||||
raw = _build([(" M", "Station PS101/räumliche Datei.xml", None)])
|
||||
(entry,) = parse_porcelain_z(raw)
|
||||
assert entry.path == "Station PS101/räumliche Datei.xml"
|
||||
|
||||
def test_trailing_empty_chunks_ignored(self) -> None:
|
||||
raw = b"\x00\x00 M a.xml\x00\x00"
|
||||
entries = parse_porcelain_z(raw)
|
||||
assert len(entries) == 1
|
||||
|
||||
def test_invalid_xy_rejected(self) -> None:
|
||||
with pytest.raises(GitError):
|
||||
parse_porcelain_z(b"XX path.xml\x00")
|
||||
|
||||
|
||||
class TestChangeTypeMapping:
|
||||
def test_staged_addition(self) -> None:
|
||||
entry = RawStatusEntry.from_zy_pair("A ", "a.xml", None)
|
||||
assert entry.change_type is ChangeType.ADDED
|
||||
|
||||
def test_staged_rename(self) -> None:
|
||||
entry = RawStatusEntry.from_zy_pair("R ", "a.xml", "b.xml")
|
||||
assert entry.change_type is ChangeType.RENAMED
|
||||
Reference in New Issue
Block a user