234 lines
9.0 KiB
Python
234 lines
9.0 KiB
Python
"""Unit tests for the repository lock, config, ULID and commit message."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import re
|
|
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"
|
|
assert config.commit.file_separator == ";" # default
|
|
|
|
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,
|
|
file_timestamp=dt.datetime(2026, 9, 11, 6, 42, 9, tzinfo=dt.UTC),
|
|
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 "Deleted: 1" in message
|
|
# One-line-per-file entries with the default ';' separator (Section 15)
|
|
assert " A ; Stations/PS101/pump2.xml ; UNKNOWN ; " in message
|
|
assert " M ; Stations/PS101/project.xml ; DOMAIN\\jsmith ; " in message
|
|
assert " M ; System/global.xml ; UNKNOWN ; " in message
|
|
assert " D ; Stations/PS027/old.xml" in message
|
|
# Local-time timestamps with explicit +/-HH:MM offset
|
|
assert re.search(
|
|
r"; \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}[+-]\d{2}:\d{2}\b", message
|
|
)
|
|
# One 'Changes:' section; no multi-line author blocks remain
|
|
assert "Last modified by:" not in message
|
|
assert "Changes:" 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())
|
|
# Deleted entries are bare: letter, separator, path - no author/time
|
|
assert " D ; gone.xml" in message
|
|
assert "UNKNOWN" not in message
|
|
for line in message.splitlines():
|
|
if "gone.xml" in line:
|
|
assert line.count(";") == 1
|
|
|
|
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 " R ; old.xml -> new.xml" in message
|
|
|
|
def test_custom_separator(self) -> None:
|
|
change_set = ChangeSet(
|
|
added=[ChangeRecord(path="pump.xml", change_type=ChangeType.ADDED)],
|
|
deleted=[ChangeRecord(path="gone.xml", change_type=ChangeType.DELETED)],
|
|
)
|
|
run = self._run()
|
|
message = build_commit_message(change_set, run, file_separator="|")
|
|
assert " A | pump.xml ; UNKNOWN ; " not in message
|
|
assert " A | pump.xml | UNKNOWN | " in message
|
|
assert " D | gone.xml" in message
|