diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0d8c32b --- /dev/null +++ b/.gitignore @@ -0,0 +1,195 @@ +# Created by https://www.toptal.com/developers/gitignore/api/visualstudiocode,python +# Edit at https://www.toptal.com/developers/gitignore?templates=visualstudiocode,python + +### Python ### +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/#use-with-ide +.pdm.toml + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ + +### Python Patch ### +# Poetry local configuration file - https://python-poetry.org/docs/configuration/#local-configuration +poetry.toml + +# ruff +.ruff_cache/ + +# LSP config files +pyrightconfig.json + +### VisualStudioCode ### +.vscode/* +!.vscode/settings.json +!.vscode/tasks.json +!.vscode/launch.json +!.vscode/extensions.json +!.vscode/*.code-snippets + +# Local History for Visual Studio Code +.history/ + +# Built Visual Studio Code Extensions +*.vsix + +### VisualStudioCode Patch ### +# Ignore all local history of files +.history +.ionide + +# End of https://www.toptal.com/developers/gitignore/api/visualstudiocode,python \ No newline at end of file diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..af5239c --- /dev/null +++ b/LICENSE @@ -0,0 +1,50 @@ +Copyright (c) 2026 Vivab. All rights reserved. + +This software and its source code are the proprietary property of Vivab. + +1. Grant of rights. No rights are granted under this license other than + those expressly stated herein. This software is provided for internal + use within Vivab and its authorized affiliates only. + +2. Restrictions. Without prior written permission from the copyright + holder, you may NOT: + + a) copy, distribute, sublicense, sell, or otherwise transfer this + software or any derivative works thereof, in whole or in part; + b) modify, adapt, translate, or create derivative works based on this + software, except as required for internal use, maintenance, or + development within Vivab; + c) remove or alter any copyright, trademark, or other proprietary + notices contained in or on this software or its source code; + d) use this software to provide services to third parties. + +3. Internal use. "Internal use" means use by Vivab employees, contractors, + and authorized partners in the course of work performed for or on + behalf of Vivab. + +4. Third-party components. This software depends on third-party open + source components that are distributed under their own licenses. + Those licenses remain unaffected by, and take precedence over, this + license for the respective components. The applicable third-party + licenses are documented in the project's dependency metadata + (pyproject.toml) and the installed package inventory. + +5. No warranty. This software is provided "AS IS", without warranty of + any kind, express or implied, including but not limited to the + warranties of merchantability, fitness for a particular purpose, + and noninfringement. In no event shall the authors or copyright + holder be liable for any claim, damages, or other liability, whether + in an action of contract, tort, or otherwise, arising from, out of, + or in connection with this software or the use or other dealings in + this software. + +6. Contact. For licensing questions, permission requests, or further + information, contact Vivab (https://www.vivab.se). + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..ca34817 --- /dev/null +++ b/README.md @@ -0,0 +1,65 @@ +# git-auto-tracker + +Unattended change tracker that records modifications to SCADA project and +configuration files in a Git repository. + +The tracker periodically inspects a configured repository, detects +added / modified / deleted files, resolves the last modifying user for each +change (where a metadata source is available), and creates one Git commit per +collection cycle with a structured, auditable commit message. + +See `functional specification.md` for the full specification. + +## Features + +- Run-once execution, designed for Windows Task Scheduler / cron / systemd timers +- Change detection via `git status --porcelain -z` (NUL-safe, UTF-8) +- Configurable include/exclude file filters (globs) +- Security validation with forbidden patterns (abort by default) +- Pluggable author resolvers with priority chain and confidence levels +- Per-file attribution recorded in the commit message +- Structured commit trailers (`Tracker-Run`, `Tracker-Host`, ...) +- Repository-level lock with stale-lock detection (stored outside the repo) +- Rotating file logs plus console logging +- Dry-run mode, exit codes suitable for external monitoring +- Cross-platform: Windows / Linux / macOS + +## Installation + +```bash +python -m venv .venv +.venv/Scripts/activate # Windows +source .venv/bin/activate # Linux/macOS +pip install -e . +``` + +## Usage + +```bash +git-auto-tracker run [--dry-run] # collect changes and commit +git-auto-tracker status # show detected changes without committing +git-auto-tracker validate # validate configuration / repository / resolvers +git-auto-tracker config # print effective configuration +``` + +Global options: + +```text +--config PATH Path to configuration file (default: git-auto-tracker.yaml) +--log-level LEVEL Override the configured log level +--version Print version and exit +``` + +## Configuration + +See [`config/example.yaml`](config/example.yaml). Configuration is YAML; +deployment-specific values can be overridden through environment variables +using the `GIT_AUTO_TRACKER_
__` convention. + +## Development + +```bash +pip install -e . --group dev +pytest +ruff check src tests +``` \ No newline at end of file diff --git a/config/example.yaml b/config/example.yaml new file mode 100644 index 0000000..4643415 --- /dev/null +++ b/config/example.yaml @@ -0,0 +1,44 @@ +# Example configuration for git-auto-tracker. +# Deployment-specific values can be overridden with environment variables +# using the convention GIT_AUTO_TRACKER_
__, e.g. +# GIT_AUTO_TRACKER_REPOSITORY__PATH=D:/SCADA/Project + +application: + timezone: Europe/Stockholm + +repository: + path: "D:/SCADA/Project" + 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: true + file: "../logs/git-auto-tracker.log" + max_size_mb: 20 + backup_count: 10 \ No newline at end of file diff --git a/functional specification.md b/functional specification.md index f22e3cc..c125910 100644 --- a/functional specification.md +++ b/functional specification.md @@ -2,7 +2,7 @@ ## SCADA Unattended Change Tracker -### 1. Purpose +# 1. Purpose The purpose of the application is to provide automated version tracking of configuration and project files associated with a SCADA system. @@ -18,6 +18,14 @@ The resulting Git history shall provide an auditable record showing: The application is intended to operate without normal user interaction and shall be capable of running on Windows, Linux and macOS. +## 1.1 Conformance language + +In this specification: + +* **shall** – the requirement is mandatory; +* **should** – the requirement is recommended, but deviations are acceptable with documented justification; +* **may** – the behaviour is optional. + --- # 2. Scope @@ -64,6 +72,7 @@ The application should be divided into independent functional components. v +-------------------------+ | Application Controller | +| (config, lock, exit) | +------------+------------+ | v @@ -76,17 +85,17 @@ The application should be divided into independent functional components. | Change Detector | +------------+------------+ | - +-----+------+ - | | - v v -+-------------+ +----------------+ -| File Change | | Author Resolver| -| Information | | | -+------+------+ +-------+--------+ - | | - +-------+--------+ - | - v + v ++-------------------------+ +| Security Validator | ++------------+------------+ + | + v ++-------------------------+ +| Author Resolver (chain) | ++------------+------------+ + | + v +-------------------------+ | Commit Message Builder | +------------+------------+ @@ -100,6 +109,9 @@ The application should be divided into independent functional components. +-------------------------+ | Logging / Audit | +-------------------------+ + +Logging / Audit is a cross-cutting concern: every component reports to it +directly rather than only through the final stage. ``` The components should be loosely coupled so that SCADA-specific metadata extraction can be implemented independently of Git handling. @@ -115,7 +127,7 @@ The application shall support at least two operating modes. Example: ```text -scada-change-tracker run +git-auto-tracker run ``` The application performs one complete inspection and exits. @@ -151,10 +163,11 @@ schedule: The application should use local system time unless a timezone is explicitly configured. -Example: +Example (see also Section 26): ```yaml -timezone: Europe/Stockholm +application: + timezone: Europe/Stockholm ``` --- @@ -183,6 +196,17 @@ Before attempting any modification, the application shall verify: If validation fails, the execution shall stop without modifying the repository. +## 5.1 Repository and index state + +Before collecting changes the application shall additionally verify that: + +* the repository index does not already contain staged changes; +* the repository is not in an unfinished merge, rebase or cherry-pick state. + +A non-empty index at startup indicates that a human or another process prepared changes outside the tracker. Because such changes carry no tracker attribution and may be incomplete, the default policy shall be to abort with a repository error (exit code 3, see Section 25). A configurable policy may allow such changes to be committed with `UNKNOWN` attribution. Staged changes left behind by an interrupted tracker execution are not special-cased; recovery requires manual review, which keeps unattended behaviour safe and predictable. + +If the current branch differs from the configured branch, the application shall abort by default rather than switch branches, since switching modifies the working tree and is not an unattended-safe operation. + --- # 6. Change Detection @@ -202,6 +226,8 @@ The following change types shall be supported: Rename detection may initially be treated as one deletion and one addition if native rename detection is not implemented. +Git does not report renames for unstaged working-tree changes; rename detection is only reliable after staging. The implementation should therefore either stage first and inspect the staged diff, or deliberately report renames as one deletion plus one addition. + --- ## 6.2 Git status @@ -211,13 +237,15 @@ The application should use Git itself as the authoritative mechanism for determi The implementation should therefore use an equivalent of: ```text -git status --porcelain +git status --porcelain -z ``` rather than independently comparing file timestamps. This avoids false change detection caused by modification times alone. +The NUL-terminated (`-z`) output format shall be used for all Git output that contains file paths. Line-based parsing of the default porcelain output is unsafe because Git C-quotes filenames containing spaces, quotes or non-ASCII characters. All Git output shall be decoded as UTF-8, and commit messages shall be written as UTF-8. + --- # 7. File Inclusion and Exclusion @@ -244,6 +272,8 @@ Glob-style matching should be supported. Files excluded by configuration shall neither be committed nor included in commit messages. +Files ignored by the repository's own `.gitignore` shall also be excluded. The tracker shall not override ignore rules unless explicitly configured to do so. + --- # 8. Last Modifier Identification @@ -265,7 +295,7 @@ Modified: Last modified by: DOMAIN\jsmith ``` -Deleted files do not require a last-modifier lookup because the original file may no longer be available. +Deleted files do not require a last-modifier lookup because the original file may no longer be available. Resolvers may nevertheless attribute deletions when possible, for example through the SCADA audit log or the file's previous commit history. Attribution for deletions is permitted but not mandatory. --- @@ -303,6 +333,15 @@ Example: } ``` +The `confidence` field shall use one of the following levels: + +| Level | Meaning | +| --------------- | ------------------------------------------------------------- | +| `authoritative` | The source records the change actor in an authenticated way | +| `probable` | The source is reliable but not an authenticated change record | +| `weak` | Weak correlation, e.g. ownership or workstation inference | +| `unknown` | No identity could be determined | + --- # 10. Author Resolver Strategies @@ -390,12 +429,13 @@ author_resolution: - scada_audit_log - sidecar_metadata + minimum_confidence: probable fallback: unknown ``` The resolvers shall execute in configured priority order. -The first authoritative result shall be used. +The first result whose confidence meets the configured minimum confidence (default: `probable`) shall be used. Results below the threshold shall be logged but not used. If no resolver produces a usable result, the configured fallback shall be used. --- @@ -470,6 +510,16 @@ git add --all unless the repository is exclusively dedicated to tracked SCADA data. +Staging shall instead use pathspec-limited operations that stage additions, modifications and deletions for the selected paths only. + +Step 5 (verification) shall compare the actual staged change set against the expected change set using an operation equivalent to: + +```text +git diff --cached --name-status -z +``` + +If the staged change set does not match the expected change set, the commit shall be aborted and the index restored to its pre-execution state, so that the next execution starts from a well-defined state. + --- # 15. Commit Message Format @@ -530,7 +580,7 @@ Alternative: SCADA: 2 added, 3 modified, 1 deleted ``` -The detailed information shall be placed in the commit body. +Exactly one subject format shall be used consistently; the timestamp format shall be the default. The detailed information shall be placed in the commit body. --- @@ -628,10 +678,12 @@ Only one tracker execution shall be allowed against a repository at the same tim The application shall implement a repository-level execution lock. +The lock file shall be stored **outside** the tracked repository, for example as a sibling of the repository directory or in a configured state directory. A lock file placed inside the repository would itself appear as an untracked change and be committed by the tracker. + For example: ```text -.scada-change-tracker.lock +.git-auto-tracker.lock ``` The lock shall include information such as: @@ -642,7 +694,7 @@ hostname start_time ``` -Stale locks shall be detected safely. +Stale locks shall be detected safely. A lock may be considered stale when the recorded process no longer exists (verifiable only on the same host) or when the lock is older than a configurable age threshold, which is required for locks left by other hosts. Removing a stale lock shall be logged as a warning. --- @@ -687,13 +739,15 @@ Example: ```yaml logging: - file: "logs/scada-change-tracker.log" + file: "logs/git-auto-tracker.log" max_size_mb: 20 backup_count: 10 ``` Console logging shall also be supported. +Relative log paths shall be resolved relative to the configuration file location. The log file shall not be placed inside the tracked repository unless explicitly excluded; otherwise the tracker's own log would be detected as a change and committed. + --- # 24. Error Handling @@ -709,6 +763,7 @@ FILE_ACCESS_ERROR AUTHOR_RESOLUTION_ERROR GIT_ERROR LOCK_ERROR +SECURITY_ERROR INTERNAL_ERROR ``` @@ -723,6 +778,8 @@ Reason: Audit log unavailable A Git or repository integrity error shall prevent the commit. +A security violation (see Section 33) shall prevent the commit and shall be reported through the `SECURITY_ERROR` category and the dedicated exit code. + --- # 25. Exit Codes @@ -738,6 +795,9 @@ Suggested process exit codes: | 4 | Git operation failure | | 5 | Repository locked | | 6 | Change collection failed | +| 7 | Security violation | + +Exit code `0` shall be returned both when changes were committed and when no changes were detected; the execution log distinguishes the two cases. This makes the application suitable for external monitoring. @@ -759,7 +819,7 @@ repository: git: user_name: "SCADA Change Tracker" - user_email: "scada-change-tracker@localhost" + user_email: "git-auto-tracker@localhost" files: include: @@ -775,18 +835,28 @@ author_resolution: - scada_metadata - scada_audit_log + minimum_confidence: probable fallback: unknown -commit: - create_when_no_changes: false - logging: level: INFO console: true - file: "logs/scada-change-tracker.log" + file: "logs/git-auto-tracker.log" ``` -Environment variables should be capable of overriding sensitive or deployment-specific settings. +Environment variables should be capable of overriding sensitive or deployment-specific settings, using a consistent naming convention such as: + +```text +GIT_AUTO_TRACKER_
__ +``` + +Example: + +```text +GIT_AUTO_TRACKER_REPOSITORY__PATH=D:/SCADA/Project +``` + +The prohibition of empty commits (FR-014) is unconditional and therefore deliberately not configurable. --- @@ -845,7 +915,7 @@ The Git integration may use either: Using the Git CLI is recommended initially because it closely follows native Git behaviour and makes troubleshooting straightforward. -The Git executable shall therefore be treated as an external prerequisite. +The Git executable shall therefore be treated as an external prerequisite. A minimum Git version of 2.20 shall be assumed. At startup the application shall verify: @@ -874,10 +944,18 @@ The application should avoid platform-specific Python dependencies unless they a Suggested commands: ```text -scada-change-tracker run -scada-change-tracker status -scada-change-tracker validate -scada-change-tracker config +git-auto-tracker run +git-auto-tracker status +git-auto-tracker validate +git-auto-tracker config +``` + +Global options shall include: + +```text +--config PATH Path to the configuration file (default: git-auto-tracker.yaml) +--log-level LEVEL Override the configured log level +--version Print the application version and exit ``` ### `run` @@ -909,7 +987,7 @@ Display the effective configuration with secrets redacted. The application shall support: ```text -scada-change-tracker run --dry-run +git-auto-tracker run --dry-run ``` Dry-run mode shall perform: @@ -971,7 +1049,7 @@ security: - "**/credentials.*" ``` -If such a file is detected, the default behaviour should be to abort the commit and log a security warning. +If such a file is detected, the default behaviour shall be to abort the commit, log a security warning, and exit with the dedicated security exit code (see Section 25). A configurable alternative may skip the forbidden file and commit the remaining changes, but skipping shall never be the default. --- @@ -1001,7 +1079,7 @@ PUSH_FAILURE # 35. Audit Information -Each collection should internally generate an execution identifier. +Each collection should internally generate an execution identifier. A ULID, or an equivalent lexicographically sortable and timezone-independent identifier, is recommended. For example: @@ -1037,7 +1115,7 @@ Tracker-Host: SCADA-SRV-01 Tracker-Changes: 6 ``` -This enables later processing without parsing the human-readable sections. +This enables later processing without parsing the human-readable sections. The trailer block shall follow the Git trailer convention so that standard tooling such as `git interpret-trailers` can extract it. --- @@ -1139,12 +1217,32 @@ A normal execution shall produce at most one commit representing the complete de Configured forbidden files shall not automatically be committed. +## FR-025 – Concurrent modification protection + +The system shall detect files modified during collection and abort or retry according to a configured policy. The default policy shall abort the commit. + +## FR-026 – Index-state validation + +The system shall refuse to execute when the repository index already contains staged changes or the repository is in an unfinished merge/rebase state. + +## FR-027 – Security abort + +The system shall abort the commit and exit with a dedicated exit code when a forbidden file is detected. + +## FR-028 – Self-exclusion + +The system shall never commit its own operational artefacts, such as lock files, logs and state files. + +## FR-029 – Filename safety + +The system shall correctly process file paths containing spaces, quotes and non-ASCII characters on all supported platforms. + --- # 38. Suggested Python Package Structure ```text -scada-change-tracker/ +git-auto-tracker/ │ ├── pyproject.toml ├── README.md @@ -1154,7 +1252,7 @@ scada-change-tracker/ │ └── example.yaml │ ├── src/ -│ └── scada_change_tracker/ +│ └── git_auto_tracker/ │ │ │ ├── __init__.py │ ├── __main__.py @@ -1232,29 +1330,29 @@ Each item is represented by a `ChangeRecord`. A normal execution shall approximately perform the following sequence: ```text -1. Load configuration -2. Validate configuration -3. Acquire repository lock -4. Verify Git availability -5. Validate repository -6. Determine current repository status -7. Apply inclusion/exclusion rules -8. Create ChangeSet -9. Resolve authors for ADDED files +1. Load configuration +2. Validate configuration +3. Verify Git availability +4. Acquire repository lock +5. Validate repository, branch and index state +6. Determine current repository status +7. Apply inclusion/exclusion and security rules +8. Create ChangeSet +9. Resolve authors for ADDED files 10. Resolve authors for MODIFIED files -11. Validate files against security rules -12. Generate commit message -13. Stage selected changes -14. Verify staged changes -15. Create commit -16. Record commit hash -17. Write audit result -18. Release lock -19. Exit +11. Generate commit message +12. Stage selected changes +13. Verify staged change set +14. Create commit +15. Record commit hash and write audit result +16. Release lock +17. Exit ``` If no changes exist, execution proceeds directly from step 8 to audit logging and termination. +If any step from staging onwards fails, the index shall be restored to its pre-execution state before the lock is released. + --- # 41. Testing Requirements @@ -1295,6 +1393,16 @@ Tests should cover at minimum: * excluded file; * forbidden file. +### Safety and edge cases + +* pre-existing staged changes in the index; +* repository in an unfinished merge/rebase state; +* current branch differs from the configured branch; +* forbidden-file abort produces the dedicated exit code; +* stale lock left by a dead process is recovered; +* lock file and log file never appear in `git status`; +* configured timezone is honoured in commit timestamps. + ### Cross-platform Automated testing should run against at least: @@ -1382,3 +1490,20 @@ Version 1.0 can be considered complete when it can: 15. execute unattended through an external scheduler. The initial implementation should deliberately keep **SCADA-specific change attribution behind the `AuthorResolver` interface**. That makes the generic Git/change-tracking engine reusable while allowing a resolver tailored to the actual SCADA platform to be added independently. + +--- + +# Appendix A – Glossary + +| Term | Definition | +| ------------------- | ---------------------------------------------------------------------------------------- | +| Collection | One complete detection-and-commit cycle performed by the tracker | +| ChangeSet | The complete set of changes detected during one collection | +| ChangeRecord | A single detected change together with its metadata | +| AuthorResolver | Pluggable component attempting to determine the last modifying identity of a file | +| AuthorResult | The outcome of one author-resolution attempt, including source and confidence | +| Confidence | Trustworthiness of an author result: `authoritative`, `probable`, `weak` or `unknown` | +| Fallback | Result used when no resolver produces a usable result (default: `UNKNOWN`) | +| Forbidden file | A file matching `security.forbidden_patterns`; never committed by default | +| Tracker-Run | Unique execution identifier linking logs, commits and monitoring (Section 35) | +| Staged verification | Comparison of the actual staged change set against the expected change set before commit | diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..4c1d517 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,39 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "git-auto-tracker" +version = "1.0.0" +description = "Unattended Git change tracker for SCADA project and configuration files" +readme = "README.md" +requires-python = ">=3.12" +license = { text = "Proprietary" } +authors = [{ name = "Vivab" }] +dependencies = [ + "PyYAML>=6.0", +] + +[project.scripts] +git-auto-tracker = "git_auto_tracker.cli:main" + +[dependency-groups] +dev = [ + "pytest>=8.0", + "pytest-cov>=5.0", + "ruff>=0.6", +] + +[tool.hatch.build.targets.wheel] +packages = ["src/git_auto_tracker"] + +[tool.ruff] +line-length = 100 +src = ["src", "tests"] + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B", "SIM", "W"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-q" \ No newline at end of file diff --git a/src/git_auto_tracker/__init__.py b/src/git_auto_tracker/__init__.py new file mode 100644 index 0000000..f288ae7 --- /dev/null +++ b/src/git_auto_tracker/__init__.py @@ -0,0 +1,3 @@ +"""git-auto-tracker: unattended Git change tracker for SCADA projects.""" + +__version__ = "1.0.0" diff --git a/src/git_auto_tracker/__main__.py b/src/git_auto_tracker/__main__.py new file mode 100644 index 0000000..a37aa08 --- /dev/null +++ b/src/git_auto_tracker/__main__.py @@ -0,0 +1,8 @@ +"""Allow `python -m git_auto_tracker`.""" + +import sys + +from git_auto_tracker.cli import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/git_auto_tracker/application.py b/src/git_auto_tracker/application.py new file mode 100644 index 0000000..c8aa39b --- /dev/null +++ b/src/git_auto_tracker/application.py @@ -0,0 +1,205 @@ +"""Application controller (Section 3): orchestrates one collection cycle. + +Implements the processing sequence from Section 40. +""" + +from __future__ import annotations + +import datetime as dt +import socket +from dataclasses import dataclass +from pathlib import Path + +from git_auto_tracker.audit import new_ulid +from git_auto_tracker.authors.resolver import ResolverChain, to_identity +from git_auto_tracker.changes.filters import filter_change_set +from git_auto_tracker.changes.models import ( + ChangeRecord, + ChangeSet, + ChangeType, + TrackerRun, +) +from git_auto_tracker.commits.commit import CommitManager +from git_auto_tracker.commits.message_builder import build_commit_message +from git_auto_tracker.config import AppConfig +from git_auto_tracker.errors import RepositoryError +from git_auto_tracker.git.repository import GitCommand +from git_auto_tracker.git.status import ChangeDetector, RawStatusEntry +from git_auto_tracker.logging.setup import get_logger +from git_auto_tracker.security.validator import SecurityValidator + +_APP_VERSION = "1.0.0" + + +@dataclass +class CollectionOutcome: + """Result of one run: either a commit hash or an explicit no-changes state.""" + + commit_hash: str | None + change_count: int + commit_message: str | None + dry_run: bool = False + + @property + def exit_code(self) -> int: + return 0 + + +class ApplicationController: + """Coordinates repository validation, detection, staging and commit.""" + + def __init__(self, config: AppConfig) -> None: + self.config = config + self._log = get_logger() + self.git = GitCommand(config.repository_path) + self.detector = ChangeDetector(self.git) + self.security = SecurityValidator(config.security) + self.resolvers = ResolverChain.from_config(config.author_resolution) + self.committer = CommitManager(self.git) + + # ------------------------------------------------------------------ + # Validation (Sections 5 and 5.1) + # ------------------------------------------------------------------ + + def validate_repository(self) -> str: + """Verify repository existence, branch, and index state.""" + path = self.config.repository_path + if not path.is_dir(): + raise RepositoryError(f"repository path does not exist: {path}") + if not (path / ".git").exists() and self.git.run( + ["rev-parse", "--git-dir"], check=False + ).returncode != 0: + raise RepositoryError(f"not a git repository: {path}") + + current = self.git.symbolic_ref() + if current is not None and current != self.config.branch: + raise RepositoryError( + f"current branch {current!r} differs from configured branch " + f"{self.config.branch!r} (Section 5.1)" + ) + + # Unfinished merge/rebase state (Section 5.1) + for state_file in ("MERGE_HEAD", "rebase-merge", "rebase-apply", "CHERRY_PICK_HEAD"): + if (self.git.rev_parse_git_dir() / state_file).exists(): + raise RepositoryError( + f"repository is in an unfinished {state_file} state (Section 5.1)" + ) + + # Pre-existing staged changes (Section 5.1 / FR-026) + staged = self.git.run_bytes(["diff", "--cached", "--name-only", "-z"]) + if any(part for part in staged.split(b"\x00") if part): + raise RepositoryError("repository index already contains staged changes (FR-026)") + + return current or "HEAD (detached)" + + # ------------------------------------------------------------------ + # Change detection and ChangeSet assembly (Sections 6 and 12) + # ------------------------------------------------------------------ + + def _records_from_status(self, entries: list[RawStatusEntry]) -> ChangeSet: + from git_auto_tracker.git.status import snapshot_file_attributes + + change_set = ChangeSet() + root = self.config.repository_path + for entry in entries: + record = ChangeRecord( + path=entry.path, + change_type=entry.change_type, + previous_path=entry.previous_path, + ) + if entry.change_type is not ChangeType.DELETED: + record.size, record.file_timestamp = self._file_attrs( + entry.path, root, snapshot_file_attributes + ) + bucket = { + ChangeType.ADDED: change_set.added, + ChangeType.MODIFIED: change_set.modified, + ChangeType.DELETED: change_set.deleted, + ChangeType.RENAMED: change_set.renamed, + }[entry.change_type] + bucket.append(record) + return change_set + + @staticmethod + def _file_attrs( + path: str, + root: Path, + snapshot: object, + ) -> tuple[int | None, dt.datetime | None]: + from git_auto_tracker.git.status import snapshot_file_attributes as _snap + + _ = snapshot + attrs = _snap(path, root) + if not attrs.get("exists"): + return None, None + ts = dt.datetime.fromtimestamp(attrs["mtime_ns"] / 1e9, tz=dt.UTC) + return int(attrs["size"]), ts + + # ------------------------------------------------------------------ + # Collection (Section 40) + # ------------------------------------------------------------------ + + def collect(self, dry_run: bool = False) -> CollectionOutcome: + """Run one complete collection cycle; returns its outcome.""" + log = self._log + log.info("Starting SCADA change collection") + log.info("Repository: %s", self.config.repository_path) + + branch = self.validate_repository() + log.info("Branch: %s", branch) + + entries = self.detector.detect() + change_set = self._records_from_status(entries) + + kept, excluded = filter_change_set(change_set, self.config.filters) + if excluded: + log.info("Excluded by filters: %d file(s)", len(excluded)) + + # Self-exclusion safety net (FR-028): operational artefacts are + # already outside the repository; belt-and-braces path check here. + self._assert_no_self_artifacts(kept) + + if kept.is_empty: + log.info("No SCADA changes detected.") + return CollectionOutcome(None, 0, None) + + self.security.validate(kept) + + run = TrackerRun( + run_id=new_ulid(), + #started_at=dt.datetime.now(dt.UTC), + started_at=dt.datetime.now().astimezone(), + host=socket.gethostname(), + ) + log.info("Detected %d changes (run %s)", len(kept), run.run_id) + + for record in [*kept.added, *kept.modified]: + result = self.resolvers.resolve(self.config.repository_path, record.path, log=log) + record.author = to_identity(result) + log.info( + "Author resolved: %s -> %s", record.path, result.identifier + ) + + message = build_commit_message(kept, run, app_version=_APP_VERSION) + + if dry_run: + log.info("Dry-run: repository left unmodified") + return CollectionOutcome(None, len(kept), message, dry_run=True) + + self.committer.stage_and_verify(kept) + commit_hash = self.git.commit( + message, self.config.git_user_name, self.config.git_user_email + ) + log.info("Commit created: %s", commit_hash[:8]) + log.info("Collection completed") + return CollectionOutcome(commit_hash, len(kept), message) + + def _assert_no_self_artifacts(self, change_set: ChangeSet) -> None: + """Refuse to commit tracker operational artefacts (FR-028).""" + suffixes = (".git-auto-tracker.lock",) + for record in change_set.all_records(): + for suffix in suffixes: + if record.path.endswith(suffix): + raise RepositoryError( + f"refusing to commit tracker artefact: {record.path} (FR-028)" + ) diff --git a/src/git_auto_tracker/audit.py b/src/git_auto_tracker/audit.py new file mode 100644 index 0000000..8d5db12 --- /dev/null +++ b/src/git_auto_tracker/audit.py @@ -0,0 +1,45 @@ +"""ULID generation (Section 35). + +ULIDs are lexicographically sortable, timezone-independent identifiers +linking logs, commits and monitoring. Implemented without an external +dependency per the specification's "equivalent identifier" allowance. +""" + +from __future__ import annotations + +import datetime as dt +import secrets + +_ENCODING = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" # Crockford base32 +_RANDOM_LEN = 16 +_TIME_LEN = 10 + + +def _encode_time(timestamp_ms: int) -> str: + value = max(timestamp_ms, 0) + chars = [""] * _TIME_LEN + for i in range(_TIME_LEN - 1, -1, -1): + chars[i] = _ENCODING[value & 0x1F] + value >>= 5 + return "".join(chars) + + +def _encode_random(count: int) -> str: + # 5 bits per char; gather enough random bytes then fold + bits = count * 5 + nbytes = (bits + 7) // 8 + raw = int.from_bytes(secrets.token_bytes(nbytes), "big") + chars = [""] * count + for i in range(count - 1, -1, -1): + chars[i] = _ENCODING[raw & 0x1F] + raw >>= 5 + return "".join(chars) + + +def new_ulid(timestamp: dt.datetime | None = None) -> str: + """Generate a ULID string, e.g. 01J7K8HYV84ZKFH1J4T9R8DZQN.""" + moment = timestamp or dt.datetime.now(dt.UTC) + if moment.tzinfo is None: + moment = moment.replace(tzinfo=dt.UTC) + timestamp_ms = int(moment.timestamp() * 1000) + return _encode_time(timestamp_ms) + _encode_random(_RANDOM_LEN) diff --git a/src/git_auto_tracker/authors/__init__.py b/src/git_auto_tracker/authors/__init__.py new file mode 100644 index 0000000..d40465b --- /dev/null +++ b/src/git_auto_tracker/authors/__init__.py @@ -0,0 +1 @@ +"""Author resolution framework (Sections 9-11).""" diff --git a/src/git_auto_tracker/authors/base.py b/src/git_auto_tracker/authors/base.py new file mode 100644 index 0000000..db3203b --- /dev/null +++ b/src/git_auto_tracker/authors/base.py @@ -0,0 +1,45 @@ +"""AuthorResolver interface and resolution result types (Sections 9-11).""" + +from __future__ import annotations + +import datetime as dt +from abc import ABC, abstractmethod +from dataclasses import dataclass +from pathlib import Path + +from git_auto_tracker.changes.models import Confidence + + +@dataclass(frozen=True) +class AuthorResult: + """Outcome of one author-resolution attempt (Section 9).""" + + identifier: str + name: str | None = None + source: str = "unknown" + confidence: Confidence = Confidence.UNKNOWN + timestamp: dt.datetime | None = None + + @classmethod + def unknown(cls, source: str, reason: str | None = None) -> AuthorResult: + _ = reason # reason is surfaced through logs, not the result payload + return cls(identifier="UNKNOWN", source=source, confidence=Confidence.UNKNOWN) + + +class AuthorResolver(ABC): + """Pluggable last-modifier identification (Section 9). + + Implementations must never raise for unavailable sources; they return a + low-confidence / UNKNOWN result instead so the chain can continue + (FR-021). name/source/confidence semantics follow Section 9. + """ + + #: Registry name used in configuration (author_resolution.resolvers) + name: str = "unknown" + + @abstractmethod + def resolve(self, repository_root: Path, file_path: str) -> AuthorResult: + """Attempt to determine the last modifier of *file_path*. + + *file_path* is repository-relative with POSIX separators. + """ diff --git a/src/git_auto_tracker/authors/resolver.py b/src/git_auto_tracker/authors/resolver.py new file mode 100644 index 0000000..929ba5d --- /dev/null +++ b/src/git_auto_tracker/authors/resolver.py @@ -0,0 +1,74 @@ +"""Resolver chain: priority order + confidence threshold + fallback (Section 11).""" + +from __future__ import annotations + +from pathlib import Path + +from git_auto_tracker.authors.base import AuthorResolver, AuthorResult +from git_auto_tracker.authors.sidecar import SidecarMetadataResolver +from git_auto_tracker.changes.models import AuthorIdentity, Confidence +from git_auto_tracker.config import AuthorResolutionConfig +from git_auto_tracker.errors import ConfigurationError + + +def available_resolvers() -> dict[str, type[AuthorResolver]]: + """Built-in resolver registry keyed by configuration name.""" + return {SidecarMetadataResolver.name: SidecarMetadataResolver} + + +class ResolverChain: + """Executes resolvers in configured order; first usable result wins.""" + + def __init__( + self, + resolvers: list[AuthorResolver], + minimum_confidence: Confidence, + fallback: AuthorResult, + ) -> None: + self._resolvers = resolvers + self._minimum = minimum_confidence + self._fallback = fallback + + @classmethod + def from_config(cls, config: AuthorResolutionConfig) -> ResolverChain: + registry = available_resolvers() + resolvers: list[AuthorResolver] = [] + for name in config.resolvers: + factory = registry.get(name) + if factory is None: + raise ConfigurationError(f"unknown author resolver: {name}") + resolvers.append(factory()) + minimum = Confidence(config.minimum_confidence) + if config.fallback not in ("unknown", "none"): + raise ConfigurationError(f"unsupported fallback: {config.fallback}") + fallback = AuthorResult.unknown("fallback") + return cls(resolvers, minimum, fallback) + + def resolve( + self, + repository_root: Path, + file_path: str, + log: object | None = None, + ) -> AuthorResult: + """Run the chain for one file; results below threshold are ignored.""" + for resolver in self._resolvers: + try: + result = resolver.resolve(repository_root, file_path) + except Exception: # noqa: BLE001 - resolvers must never break a commit + if log is not None: + log.warning("author resolver %s failed for %s", resolver.name, file_path) + continue + if result.confidence.at_least(self._minimum) and result.identifier != "UNKNOWN": + return result + return self._fallback + + +def to_identity(result: AuthorResult) -> AuthorIdentity: + """Convert an AuthorResult into the ChangeRecord author field.""" + return AuthorIdentity( + identifier=result.identifier, + name=result.name, + source=result.source, + confidence=result.confidence, + timestamp=result.timestamp, + ) diff --git a/src/git_auto_tracker/authors/sidecar.py b/src/git_auto_tracker/authors/sidecar.py new file mode 100644 index 0000000..f53a9ff --- /dev/null +++ b/src/git_auto_tracker/authors/sidecar.py @@ -0,0 +1,104 @@ +"""Sidecar metadata resolver (Section 10: companion metadata). + +Reads author information from a sidecar file next to the tracked file: + + pumpstation.xml + pumpstation.xml.metadata + +Supported sidecar formats (auto-detected): + +* JSON: {"last_modified_by": "...", "last_modified_at": "...", "name": "..."} +* KEY=VALUE lines: last_modified_by=DOMAIN\\jsmith + +Missing or malformed sidecars yield UNKNOWN so the chain continues. +""" + +from __future__ import annotations + +import datetime as dt +import json +from pathlib import Path + +from git_auto_tracker.authors.base import AuthorResolver, AuthorResult +from git_auto_tracker.changes.models import Confidence + +_SUFFIX = ".metadata" +_KEYS = ("last_modified_by", "last_modified_by_user", "modified_by", "user") +_NAME_KEYS = ("last_modified_by_name", "name", "full_name") +_TS_KEYS = ("last_modified_at", "last_modified_time", "modified_at", "timestamp") + + +def _parse_iso(value: str) -> dt.datetime | None: + try: + parsed = dt.datetime.fromisoformat(value) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=dt.UTC) + + +class SidecarMetadataResolver(AuthorResolver): + """Reads a `.metadata` companion file (Section 10).""" + + name = "sidecar_metadata" + + def resolve(self, repository_root: Path, file_path: str) -> AuthorResult: + sidecar = repository_root / Path(file_path).with_name(Path(file_path).name + _SUFFIX) + try: + raw = sidecar.read_text(encoding="utf-8") + except OSError: + return AuthorResult.unknown(self.name, "sidecar not readable") + + data = self._parse(raw) + if data is None: + return AuthorResult.unknown(self.name, "sidecar malformed") + + identifier = self._first_present(data, _KEYS) + if not identifier: + return AuthorResult.unknown(self.name, "sidecar has no author field") + + timestamp = None + raw_ts = self._first_present(data, _TS_KEYS) + if isinstance(raw_ts, str) and raw_ts: + timestamp = _parse_iso(raw_ts) + + return AuthorResult( + identifier=str(identifier), + name=self._optional_str(self._first_present(data, _NAME_KEYS)), + source=self.name, + confidence=Confidence.PROBABLE, + timestamp=timestamp, + ) + + def _parse(self, raw: str) -> dict[str, object] | None: + stripped = raw.strip() + if stripped.startswith("{"): + try: + data = json.loads(stripped) + except json.JSONDecodeError: + return None + return data if isinstance(data, dict) else None + # KEY=VALUE lines + result: dict[str, object] = {} + for line in stripped.splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + key, _, value = line.partition("=") + result[key.strip().lower()] = value.strip() + return result + + @staticmethod + def _first_present(data: dict[str, object], keys: tuple[str, ...]) -> object | None: + for key in keys: + value = data.get(key) + if isinstance(value, str) and value.strip(): + return value + if isinstance(value, (int, float)) and not isinstance(value, bool): + return value + return None + + @staticmethod + def _optional_str(value: object | None) -> str | None: + if isinstance(value, str) and value.strip(): + return value.strip() + return None diff --git a/src/git_auto_tracker/changes/__init__.py b/src/git_auto_tracker/changes/__init__.py new file mode 100644 index 0000000..d73a37c --- /dev/null +++ b/src/git_auto_tracker/changes/__init__.py @@ -0,0 +1 @@ +"""Change detection and ChangeSet models (Sections 6 and 12).""" diff --git a/src/git_auto_tracker/changes/filters.py b/src/git_auto_tracker/changes/filters.py new file mode 100644 index 0000000..8bdf9b4 --- /dev/null +++ b/src/git_auto_tracker/changes/filters.py @@ -0,0 +1,150 @@ +"""File filter and security validation (Sections 7 and 33). + +Pure-path matching: patterns are matched against the repository-relative +POSIX path. Excluded files are neither staged nor committed nor reported. +Forbidden files abort the commit with a security exit code by default. +""" + +from __future__ import annotations + +import re + +from git_auto_tracker.changes.models import ChangeSet +from git_auto_tracker.config import FileFilters, SecurityConfig +from git_auto_tracker.errors import SecurityError + +_PATTERN_CACHE: dict[str, re.Pattern[str]] = {} + + +def matches(path: str, pattern: str) -> bool: + """Match a repository-relative POSIX path against a glob pattern. + + Implements gitignore-style `**` semantics, which fnmatch lacks: + `**/x` matches x at any depth including the root, `dir/**` matches + everything under dir (and the directory itself), and `a/**/b` + matches b at any depth under a. + """ + regex = _PATTERN_CACHE.get(pattern) + if regex is None: + regex = _compile(pattern) + _PATTERN_CACHE[pattern] = regex + if regex.fullmatch(path) is not None: + return True + # `dir/**` also matches the bare directory entry itself, so a status + # line for the directory (or a file at that exact path) is excluded too + if pattern.endswith("/**"): + prefix = pattern[:-3] + return path == prefix + return False + + +def _compile(pattern: str) -> re.Pattern[str]: + """Translate a glob pattern with gitignore-style `**` into a regex. + + Segment rules: + * `**` as the final segment matches everything inside the directory + (gitignore: `dir/**` -> `dir/*`); the leading slash stays literal. + * `**` followed by more segments matches zero or more directories: + `**/x` matches `x` anywhere, `a/**/b` also matches `a/b`. + * `*` and `?` inside a segment never cross `/`. + """ + segments = pattern.split("/") + parts: list[str] = [] + last = len(segments) - 1 + for index, segment in enumerate(segments): + if segment == "**": + if index == last: + parts.append(".*") + else: + # Zero or more directories; consumes the trailing slash + parts.append("(?:[^/]+/)*") + continue + parts.append(_translate_segment(segment)) + if index < last: + parts.append("/") + return re.compile("(?s:" + "".join(parts) + r")\Z") + + +def _translate_segment(segment: str) -> str: + """Translate one path segment (no `/` crossing) to a regex fragment.""" + out: list[str] = [] + i = 0 + while i < len(segment): + char = segment[i] + if char == "*": + out.append("[^/]*") + elif char == "?": + out.append("[^/]") + elif char == "[": + end = segment.find("]", i + 1) + if end == -1: + out.append(re.escape(char)) + else: + content = segment[i + 1 : end] + if content.startswith("!"): + content = "^" + content[1:] + out.append(f"[{content}]") + i = end + else: + out.append(re.escape(char)) + i += 1 + return "".join(out) + + +def is_included(path: str, filters: FileFilters) -> bool: + """True when the path matches any include pattern (empty include = all).""" + return not filters.include or any(matches(path, p) for p in filters.include) + + +def is_excluded(path: str, filters: FileFilters) -> bool: + return any(matches(path, p) for p in filters.exclude) + + +def filter_change_set( + change_set: ChangeSet, + filters: FileFilters, +) -> tuple[ChangeSet, list[str]]: + """Apply include/exclude rules to a ChangeSet. + + Returns the filtered ChangeSet plus the list of excluded paths + (for logging). + """ + from git_auto_tracker.changes.models import ChangeType + + kept = ChangeSet() + excluded: list[str] = [] + + buckets = { + ChangeType.ADDED: (change_set.added, kept.added), + ChangeType.MODIFIED: (change_set.modified, kept.modified), + ChangeType.DELETED: (change_set.deleted, kept.deleted), + ChangeType.RENAMED: (change_set.renamed, kept.renamed), + } + + for records, target in buckets.values(): + for record in records: + path = record.previous_path or record.path + if is_included(path, filters) and not is_excluded(path, filters): + target.append(record) + else: + excluded.append(record.path) + + return kept, excluded + + +def find_forbidden(paths: list[str], security: SecurityConfig) -> list[str]: + """Return the paths matching any forbidden pattern (Section 33).""" + return [p for p in paths if any(matches(p, pat) for pat in security.forbidden_patterns)] + + +def validate_security(change_set: ChangeSet, security: SecurityConfig) -> list[str]: + """Find forbidden files in a ChangeSet; raises SecurityError when found.""" + paths = [r.path for r in change_set.all_records()] + # For renames, also check the source path + paths += [r.previous_path for r in change_set.renamed if r.previous_path] + forbidden = find_forbidden(paths, security) + if forbidden: + raise SecurityError( + "sensitive files detected, refusing to commit: " + ", ".join(sorted(forbidden)) + ) + return forbidden diff --git a/src/git_auto_tracker/changes/models.py b/src/git_auto_tracker/changes/models.py new file mode 100644 index 0000000..692c179 --- /dev/null +++ b/src/git_auto_tracker/changes/models.py @@ -0,0 +1,131 @@ +"""Domain models for the change tracker. + +Implements the core objects from Section 39 of the functional specification: +ChangeRecord, ChangeSet, AuthorIdentity, TrackerRun, CommitResult. +""" + +from __future__ import annotations + +import datetime as dt +import enum +from dataclasses import dataclass, field +from pathlib import Path + + +class ChangeType(enum.StrEnum): + """Type of a detected change (Section 6.1).""" + + ADDED = "ADDED" + MODIFIED = "MODIFIED" + DELETED = "DELETED" + RENAMED = "RENAMED" + + +class Confidence(enum.StrEnum): + """Trustworthiness of an author-resolution result (Section 9).""" + + AUTHORITATIVE = "authoritative" + PROBABLE = "probable" + WEAK = "weak" + UNKNOWN = "unknown" + + def at_least(self, other: Confidence) -> bool: + """True if this confidence meets or exceeds *other*.""" + order = [ + Confidence.UNKNOWN, + Confidence.WEAK, + Confidence.PROBABLE, + Confidence.AUTHORITATIVE, + ] + return order.index(self) >= order.index(other) +@dataclass(frozen=True) +class AuthorIdentity: + """Resolved identity of the last modifier of a file (Section 9).""" + + identifier: str + name: str | None = None + source: str = "unknown" + confidence: Confidence = Confidence.UNKNOWN + timestamp: dt.datetime | None = None + + UNKNOWN = object() # sentinel for classmethod below + + @classmethod + def unknown(cls, source: str = "unknown") -> AuthorIdentity: + return cls(identifier="UNKNOWN", source=source, confidence=Confidence.UNKNOWN) + + +@dataclass +class ChangeRecord: + """A single detected change together with its metadata (Section 12).""" + + path: str # repository-relative, POSIX separators + change_type: ChangeType + previous_path: str | None = None + size: int | None = None + file_timestamp: dt.datetime | None = None + author: AuthorIdentity | None = None + + @property + def needs_author(self) -> bool: + """Only added and modified files require a last-modifier lookup.""" + return self.change_type in (ChangeType.ADDED, ChangeType.MODIFIED) + + +@dataclass +class ChangeSet: + """The complete set of changes detected during one collection (Section 39).""" + + added: list[ChangeRecord] = field(default_factory=list) + modified: list[ChangeRecord] = field(default_factory=list) + deleted: list[ChangeRecord] = field(default_factory=list) + renamed: list[ChangeRecord] = field(default_factory=list) + + @property + def is_empty(self) -> bool: + return not (self.added or self.modified or self.deleted or self.renamed) + + def __len__(self) -> int: + return len(self.added) + len(self.modified) + len(self.deleted) + len(self.renamed) + + def all_records(self) -> list[ChangeRecord]: + return [*self.added, *self.modified, *self.deleted, *self.renamed] + + def summary_counts(self) -> dict[str, int]: + return { + "added": len(self.added), + "modified": len(self.modified), + "deleted": len(self.deleted), + "renamed": len(self.renamed), + } + + +@dataclass(frozen=True) +class TrackerRun: + """One collection execution, identified by a ULID (Section 35).""" + + run_id: str + started_at: dt.datetime + host: str + + @property + def trailer_lines(self) -> list[str]: + return [f"Tracker-Run: {self.run_id}", f"Tracker-Host: {self.host}"] + + +@dataclass(frozen=True) +class CommitResult: + """Outcome of a commit attempt.""" + + commit_hash: str | None + change_count: int + pushed: bool = False + + @classmethod + def empty(cls) -> CommitResult: + return cls(commit_hash=None, change_count=0) + + +def repo_relpath(path: Path | str) -> str: + """Normalize a path to a repository-relative POSIX string.""" + return Path(path).as_posix() diff --git a/src/git_auto_tracker/cli.py b/src/git_auto_tracker/cli.py new file mode 100644 index 0000000..b41e2fd --- /dev/null +++ b/src/git_auto_tracker/cli.py @@ -0,0 +1,135 @@ +"""Command-line interface (Sections 30 and 31).""" + +from __future__ import annotations + +import argparse +import logging +import sys +from pathlib import Path + +from git_auto_tracker import __version__ +from git_auto_tracker.application import ApplicationController +from git_auto_tracker.config import load_config, redact +from git_auto_tracker.errors import ExitCode, TrackerError +from git_auto_tracker.git.repository import check_git_available, verify_git_version +from git_auto_tracker.locking.repository_lock import RepositoryLock, default_lock_path +from git_auto_tracker.logging.setup import setup_logging + +_DEFAULT_CONFIG = "git-auto-tracker.yaml" + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="git-auto-tracker", + description="Unattended Git change tracker for SCADA project files", + ) + parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") + parser.add_argument("--config", type=Path, default=Path(_DEFAULT_CONFIG)) + parser.add_argument("--log-level", choices=["DEBUG", "INFO", "WARNING", "ERROR"]) + subparsers = parser.add_subparsers(dest="command", required=True) + + run_parser = subparsers.add_parser("run", help="collect changes and commit") + run_parser.add_argument("--dry-run", action="store_true", help="do not modify the repository") + + subparsers.add_parser("status", help="show detected changes without committing") + subparsers.add_parser("validate", help="validate configuration, repository and resolvers") + subparsers.add_parser("config", help="print the effective configuration (secrets redacted)") + return parser + + +def _print_changes(controller: ApplicationController) -> int: + """status command body: list detected changes without staging.""" + controller.validate_repository() + entries = controller.detector.detect() + if not entries: + print("No changes detected.") + return ExitCode.SUCCESS + for entry in entries: + marker = "R" if entry.change_type.value == "RENAMED" else entry.change_type.value[0] + print(f"{marker} {entry.path}") + return ExitCode.SUCCESS + + +def _validate_command(controller: ApplicationController) -> int: + """validate command body: configuration, git, repository, resolvers.""" + version_line = check_git_available() + if not verify_git_version(version_line): + raise TrackerError(f"insufficient git version: {version_line}") + branch = controller.validate_repository() + print(f"OK git: {version_line}") + print(f"OK repository: {controller.config.repository_path} (branch {branch})") + for name in controller.config.author_resolution.resolvers: + print(f"OK resolver: {name}") + return ExitCode.SUCCESS + + +def _config_command(config_path: Path) -> int: + import yaml # local import keeps CLI startup light + + from git_auto_tracker.config import load_config as _load + + _load(config_path) # validates; raises ConfigurationError on problems + raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) + print(yaml.safe_dump(redact(raw), sort_keys=True), end="") + return ExitCode.SUCCESS + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + + try: + config = load_config(args.config) + logger = setup_logging( + config.logging, + config_dir=args.config.parent, + repository_path=config.repository_path, + override_level=args.log_level, + ) + _ = logger + + version_line = check_git_available() + if not verify_git_version(version_line): + raise TrackerError(f"insufficient git version: {version_line}") + + if args.command == "config": + return _config_command(args.config) + + controller = ApplicationController(config) + lock = RepositoryLock(default_lock_path(config.repository_path)) + + if args.command == "validate": + return _validate_command(controller) + + if args.command == "status": + lock.acquire() + try: + return _print_changes(controller) + finally: + lock.release() + + # run (Section 40) + lock.acquire() + try: + outcome = controller.collect(dry_run=args.dry_run) + except Exception: + controller.committer.restore_index() + raise + finally: + lock.release() + + if outcome.dry_run and outcome.commit_message: + print(outcome.commit_message) + elif outcome.dry_run: + print("No changes detected.") + return ExitCode.SUCCESS + + except TrackerError as exc: + logging.getLogger("git_auto_tracker").error("%s", exc.message) + return int(exc.exit_code) + except Exception as exc: # noqa: BLE001 + logging.getLogger("git_auto_tracker").exception("internal error: %s", exc) + return int(ExitCode.GENERAL_FAILURE) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/git_auto_tracker/commits/__init__.py b/src/git_auto_tracker/commits/__init__.py new file mode 100644 index 0000000..c085f45 --- /dev/null +++ b/src/git_auto_tracker/commits/__init__.py @@ -0,0 +1 @@ +"""Commit message generation and staging/commit management (Sections 14-16, 36).""" diff --git a/src/git_auto_tracker/commits/commit.py b/src/git_auto_tracker/commits/commit.py new file mode 100644 index 0000000..f2e8e2e --- /dev/null +++ b/src/git_auto_tracker/commits/commit.py @@ -0,0 +1,87 @@ +"""Git staging and commit management (Sections 14 and 19). + +Staging uses pathspec-limited `git add -A -- ` so only selected +paths are staged, including deletions. The staged change set is then +verified against the expected one (Section 14, step 5) and the index is +restored on mismatch. +""" + +from __future__ import annotations + +from git_auto_tracker.changes.models import ChangeSet, ChangeType +from git_auto_tracker.errors import GitError +from git_auto_tracker.git.repository import GitCommand + +_GIT_LETTER = { + ChangeType.ADDED: "A", + ChangeType.MODIFIED: "M", + ChangeType.DELETED: "D", + ChangeType.RENAMED: "R", +} + + +class CommitManager: + """Stages the expected change set, verifies it, and commits.""" + + def __init__(self, git: GitCommand) -> None: + self.git = git + + def _stage_paths(self, paths: list[str]) -> None: + """Stage additions/modifications/deletions for the given paths. + + `git add -A -- ` stages all worktree states for the + selected pathspecs, including deletions, without touching other + files (Section 14). + """ + if not paths: + return + # Split pathspec lists into safe batches; Windows command-line + # length limits make one-invocation-per-repository risky. + batch_size = 100 + for start in range(0, len(paths), batch_size): + batch = paths[start : start + batch_size] + self.git.run(["add", "-A", "--", *batch]) + + def _expected_status_map(self, change_set: ChangeSet) -> dict[str, str]: + """Map staged path -> expected git letter for verification. + + RENAMED is expected as add+delete per Section 6.1, matching + `git diff --cached --no-renames` output. + """ + mapping: dict[str, str] = {} + for record in change_set.all_records(): + if record.change_type is ChangeType.RENAMED: + mapping[record.previous_path or record.path] = "D" + mapping[record.path] = "A" + else: + mapping[record.path] = _GIT_LETTER[record.change_type] + return mapping + + def stage_and_verify(self, change_set: ChangeSet) -> None: + """Stage exactly the ChangeSet; raise GitError on staged mismatch.""" + paths = sorted( + {record.path for record in change_set.all_records()} + | { + record.previous_path + for record in change_set.renamed + if record.previous_path + } + ) + self._stage_paths(paths) + + staged = self.git.commit_file_status() + expected = self._expected_status_map(change_set) + + actual = {path: status[0] for status, path, _previous in staged} + + if actual != expected: + missing = sorted(set(expected) - set(actual)) + extra = sorted(set(actual) - set(expected)) + raise GitError( + "staged change set does not match expected set " + f"(missing: {missing or 'none'}, unexpected: {extra or 'none'})" + ) + + def restore_index(self) -> None: + """Best-effort restore of the index to its pre-execution state.""" + self.git.run(["reset", "--mixed", "HEAD"], check=False) diff --git a/src/git_auto_tracker/commits/message_builder.py b/src/git_auto_tracker/commits/message_builder.py new file mode 100644 index 0000000..87f1a2a --- /dev/null +++ b/src/git_auto_tracker/commits/message_builder.py @@ -0,0 +1,81 @@ +"""Commit message generation (Sections 15, 16, 36). + +Human-readable sections first, Git trailer block last so that +`git interpret-trailers` can extract the structured metadata. +""" + +from __future__ import annotations + +import datetime as dt + +from git_auto_tracker.changes.models import ChangeSet, TrackerRun + +_SUMMARY_LABELS = ( + ("Added:", "added"), + ("Modified:", "modified"), + ("Deleted:", "deleted"), + ("Renamed:", "renamed"), +) + + +def _format_ts(moment: dt.datetime | None) -> str: + if moment is None: + return "" + return moment.strftime("%Y-%m-%d %H:%M:%S%z") + + +def _author_lines(record) -> list[str]: # noqa: ANN001 - ChangeRecord, avoid cycle + lines: list[str] = [] + author = record.author + if author is not None: + lines.append(f" Last modified by: {author.identifier}") + ts = _format_ts(author.timestamp) + if ts: + lines.append(f" Modified at: {ts}") + else: + lines.append(" Last modified by: UNKNOWN") + return lines + + +def build_commit_message( + change_set: ChangeSet, + run: TrackerRun, + collected_at: dt.datetime | None = None, + app_version: str = "1.0.0", +) -> str: + """Build the complete commit message per Sections 15/16/36.""" + counts = change_set.summary_counts() + moment = collected_at or run.started_at + subject = f"SCADA automatic change capture - {_format_ts(moment)}" + + lines: list[str] = [subject, ""] + + lines.append("Summary:") + for label, key in _SUMMARY_LABELS: + lines.append(f" {label} {counts[key]}") + lines.append("") + + for heading, records in ( + ("Added:", change_set.added), + ("Modified:", change_set.modified), + ("Renamed:", change_set.renamed), + ("Deleted:", change_set.deleted), + ): + if not records: + continue + lines.append(heading) + for record in records: + if record.change_type.value == "RENAMED" and record.previous_path: + lines.append(f" {record.previous_path} -> {record.path}") + else: + lines.append(f" {record.path}") + if record.needs_author: + lines.extend(_author_lines(record)) + lines.append("") + + # Structured trailers (Section 36) + lines.append(f"Tracker-Version: {app_version}") + lines.extend(run.trailer_lines) + lines.append(f"Tracker-Changes: {len(change_set.all_records())}") + + return "\n".join(lines).rstrip() + "\n" diff --git a/src/git_auto_tracker/config.py b/src/git_auto_tracker/config.py new file mode 100644 index 0000000..2110c12 --- /dev/null +++ b/src/git_auto_tracker/config.py @@ -0,0 +1,192 @@ +"""Configuration loading and validation (Section 26). + +YAML is the primary source; environment variables using the +GIT_AUTO_TRACKER_
__KEY convention override individual values. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +from git_auto_tracker.errors import ConfigurationError + +_ENV_PREFIX = "GIT_AUTO_TRACKER_" +_SECRET_KEYS = {"password", "token", "secret", "user_email"} + + +def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: + for key, value in override.items(): + if isinstance(value, dict) and isinstance(base.get(key), dict): + _deep_merge(base[key], value) + else: + base[key] = value + return base + + +def _coerce(raw: str) -> Any: + """Coerce environment strings to bool/int/float where natural.""" + low = raw.lower() + if low in ("true", "yes"): + return True + if low in ("false", "no"): + return False + for cast in (int, float): + try: + return cast(raw) # type: ignore[call-overload] + except ValueError: + continue + return raw + + +def _apply_env_overrides(config: dict[str, Any], environ: dict[str, str]) -> dict[str, Any]: + for name, value in environ.items(): + if not name.startswith(_ENV_PREFIX): + continue + remainder = name[len(_ENV_PREFIX) :] + parts = remainder.split("__") + if len(parts) < 2: + continue + node = config + for part in parts[:-1]: + key = part.lower() + child = node.get(key) + if not isinstance(child, dict): + child = {} + node[key] = child + node = child + node[parts[-1].lower()] = _coerce(value) + return config + + +@dataclass +class FileFilters: + include: list[str] = field(default_factory=lambda: ["**/*"]) + exclude: list[str] = field(default_factory=list) + + +@dataclass +class AuthorResolutionConfig: + resolvers: list[str] = field(default_factory=lambda: ["sidecar_metadata"]) + minimum_confidence: str = "probable" + fallback: str = "unknown" + + +@dataclass +class SecurityConfig: + forbidden_patterns: list[str] = field(default_factory=list) + + +@dataclass +class LoggingConfig: + level: str = "INFO" + console: bool = True + file: str | None = None + max_size_mb: int = 20 + backup_count: int = 10 + + +@dataclass +class AppConfig: + repository_path: Path + branch: str + git_user_name: str + git_user_email: str + filters: FileFilters + security: SecurityConfig + author_resolution: AuthorResolutionConfig + logging: LoggingConfig + timezone: str | None = None + + @classmethod + def from_dict(cls, data: dict[str, Any], config_dir: Path) -> AppConfig: + try: + repository = data["repository"] + git_cfg = data.get("git", {}) + files_cfg = data.get("files", {}) + security_cfg = data.get("security", {}) + author_cfg = data.get("author_resolution", {}) + logging_cfg = data.get("logging", {}) + app_cfg = data.get("application", {}) + + repo_path_str = repository["path"] + if not isinstance(repo_path_str, str) or not repo_path_str: + raise ConfigurationError("repository.path must be a non-empty string") + repo_path = Path(repo_path_str) + if not repo_path.is_absolute(): + repo_path = (config_dir / repo_path).resolve() + + branch = repository.get("branch") or "main" + if not isinstance(branch, str) or not branch: + raise ConfigurationError("repository.branch must be a non-empty string") + + filters = FileFilters( + include=list(files_cfg.get("include", ["**/*"])), + exclude=list(files_cfg.get("exclude", [])), + ) + if not filters.include: + raise ConfigurationError("files.include must not be empty") + + return cls( + repository_path=repo_path, + branch=branch, + git_user_name=git_cfg.get("user_name") or "Git Auto Tracker", + git_user_email=git_cfg.get("user_email") or "git-auto-tracker@localhost", + filters=filters, + security=SecurityConfig( + forbidden_patterns=list(security_cfg.get("forbidden_patterns", [])), + ), + author_resolution=AuthorResolutionConfig( + resolvers=list(author_cfg.get("resolvers", ["sidecar_metadata"])), + minimum_confidence=author_cfg.get("minimum_confidence", "probable"), + fallback=author_cfg.get("fallback", "unknown"), + ), + logging=LoggingConfig( + level=logging_cfg.get("level", "INFO"), + console=bool(logging_cfg.get("console", True)), + file=logging_cfg.get("file"), + max_size_mb=int(logging_cfg.get("max_size_mb", 20)), + backup_count=int(logging_cfg.get("backup_count", 10)), + ), + timezone=app_cfg.get("timezone"), + ) + except ConfigurationError: + raise + except KeyError as exc: + raise ConfigurationError(f"missing required configuration key: {exc}") from exc + except (TypeError, ValueError) as exc: + raise ConfigurationError(f"invalid configuration value: {exc}") from exc + + +def load_config(path: Path, environ: dict[str, str] | None = None) -> AppConfig: + """Load YAML configuration with environment overrides applied.""" + if not path.is_file(): + raise ConfigurationError(f"configuration file not found: {path}") + try: + raw = yaml.safe_load(path.read_text(encoding="utf-8")) + except yaml.YAMLError as exc: + raise ConfigurationError(f"invalid YAML in {path}: {exc}") from exc + if not isinstance(raw, dict): + raise ConfigurationError(f"configuration root in {path} must be a mapping") + + merged = _apply_env_overrides(raw, os.environ if environ is None else environ) + return AppConfig.from_dict(merged, config_dir=path.parent) + + +def redact(data: dict[str, Any]) -> dict[str, Any]: + """Return a copy with secret-like values replaced (used by `config`).""" + def walk(node: Any) -> Any: + if isinstance(node, dict): + return { + key: ("" if any(s in key.lower() for s in _SECRET_KEYS) else walk(value)) + for key, value in node.items() + } + if isinstance(node, list): + return [walk(item) for item in node] + return node + + return walk(data) diff --git a/src/git_auto_tracker/errors.py b/src/git_auto_tracker/errors.py new file mode 100644 index 0000000..f198fc2 --- /dev/null +++ b/src/git_auto_tracker/errors.py @@ -0,0 +1,56 @@ +"""Error taxonomy and process exit codes (Sections 24 and 25).""" + +from __future__ import annotations + +import enum + + +class ExitCode(enum.IntEnum): + """Process exit codes (Section 25).""" + + SUCCESS = 0 + GENERAL_FAILURE = 1 + CONFIGURATION_ERROR = 2 + REPOSITORY_ERROR = 3 + GIT_ERROR = 4 + REPOSITORY_LOCKED = 5 + COLLECTION_FAILED = 6 + SECURITY_VIOLATION = 7 + + +class TrackerError(Exception): + """Base class for tracker errors carrying an exit code.""" + + exit_code: ExitCode = ExitCode.GENERAL_FAILURE + + def __init__(self, message: str) -> None: + super().__init__(message) + self.message = message + + +class ConfigurationError(TrackerError): + exit_code = ExitCode.CONFIGURATION_ERROR + + +class RepositoryError(TrackerError): + exit_code = ExitCode.REPOSITORY_ERROR + + +class GitError(TrackerError): + exit_code = ExitCode.GIT_ERROR + + +class LockError(TrackerError): + exit_code = ExitCode.REPOSITORY_LOCKED + + +class SecurityError(TrackerError): + exit_code = ExitCode.SECURITY_VIOLATION + + +class CollectionError(TrackerError): + exit_code = ExitCode.COLLECTION_FAILED + + +class InternalError(TrackerError): + exit_code = ExitCode.GENERAL_FAILURE diff --git a/src/git_auto_tracker/git/__init__.py b/src/git_auto_tracker/git/__init__.py new file mode 100644 index 0000000..cbb66d0 --- /dev/null +++ b/src/git_auto_tracker/git/__init__.py @@ -0,0 +1 @@ +"""Git integration (Section 28).""" diff --git a/src/git_auto_tracker/git/repository.py b/src/git_auto_tracker/git/repository.py new file mode 100644 index 0000000..df78289 --- /dev/null +++ b/src/git_auto_tracker/git/repository.py @@ -0,0 +1,167 @@ +"""Git CLI integration (Section 28). + +All Git output containing paths uses NUL-terminated (-z) formats and is +decoded as UTF-8 (Section 6.2). Argument arrays are always used; no shell +strings are constructed. +""" + +from __future__ import annotations + +import subprocess +from collections.abc import Sequence +from pathlib import Path + +from git_auto_tracker.errors import GitError + +_MIN_GIT_VERSION = (2, 20) + + +def _run_git( + args: Sequence[str], + cwd: Path, + check: bool = True, +) -> subprocess.CompletedProcess[str]: + """Run git with argument array; returns completed process with UTF-8 text.""" + try: + return subprocess.run( # noqa: SIM115 - short-lived, no handles kept + ["git", *args], + cwd=str(cwd), + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + except FileNotFoundError as exc: + raise GitError("git executable not found on PATH") from exc + except OSError as exc: + raise GitError(f"failed to execute git: {exc}") from exc + + +def check_git_available() -> str: + """Verify `git --version`; returns the version string (Section 28).""" + proc = subprocess.run( # noqa: SIM115 + ["git", "--version"], + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + if proc.returncode != 0: + raise GitError("git is not installed or not on PATH") + return (proc.stdout or "").strip() + + +def _parse_version(version_line: str) -> tuple[int, ...]: + digits = "".join(ch if ch.isdigit() else "." for ch in version_line) + return tuple(int(part) for part in digits.split(".") if part.isdigit()) + + +def verify_git_version(version_line: str) -> bool: + """True if the git version meets the minimum (>= 2.20).""" + parts = _parse_version(version_line) + return parts[:2] >= _MIN_GIT_VERSION + + +class GitCommand: + """Thin wrapper around the git CLI for one repository.""" + + def __init__(self, workdir: Path) -> None: + self.workdir = workdir + + def run( + self, + args: Sequence[str], + check: bool = True, + allow_stderr: bool = False, + ) -> subprocess.CompletedProcess[str]: + proc = _run_git(args, self.workdir, check=False) + if check and proc.returncode != 0: + stderr = (proc.stderr or "").strip() + raise GitError(f"git {' '.join(args)} failed: {stderr or proc.returncode}") + return proc + + def run_bytes(self, args: Sequence[str]) -> bytes: + """Run git capturing raw bytes (needed for -z output).""" + try: + proc = subprocess.run( # noqa: SIM115 + ["git", *args], + cwd=str(self.workdir), + capture_output=True, + check=False, + ) + except FileNotFoundError as exc: + raise GitError("git executable not found on PATH") from exc + except OSError as exc: + raise GitError(f"failed to execute git: {exc}") from exc + if proc.returncode != 0: + stderr = (proc.stderr or b"").decode("utf-8", errors="replace").strip() + raise GitError(f"git {' '.join(args)} failed: {stderr or proc.returncode}") + return proc.stdout + + def rev_parse(self, name: str) -> str: + proc = self.run(["rev-parse", name]) + return (proc.stdout or "").strip() + + def rev_parse_git_dir(self) -> Path: + out = self.run(["rev-parse", "--absolute-git-dir"]).stdout.strip() + return Path(out) + + def symbolic_ref(self) -> str | None: + """Current branch name, or None when detached HEAD.""" + proc = self.run(["symbolic-ref", "--short", "HEAD"], check=False) + if proc.returncode != 0: + return None + return (proc.stdout or "").strip() + + def commit(self, message: str, author_name: str, author_email: str) -> str: + """Create a commit from the index; returns the new commit hash.""" + author = f"{author_name} <{author_email}>" + self.run( + ["commit", "-m", message, f"--author={author}"], + ) + return self.rev_parse("HEAD") + + def commit_file_status(self) -> list[tuple[str, str, str | None]]: + """Staged change set as (status, path, previous_path) tuples. + + Uses `git diff --cached --name-status -z --no-renames` (Section 14, + step 5). Rename detection is disabled so the staged set is compared + against the expected add/delete fallback representation mandated by + Section 6.1 (renames reported as one deletion plus one addition). + """ + raw = self.run_bytes( + ["diff", "--cached", "--name-status", "-z", "--no-renames"] + ) + return _parse_name_status_z(raw, expect_renames=False) + + +def _parse_name_status_z( + raw: bytes, + expect_renames: bool, +) -> list[tuple[str, str, str | None]]: + """Parse NUL-terminated `--name-status -z` output. + + For rename/copy entries (status starting with R or C followed by a + score), the record holds `status\\0new\\0old\\0`; all other entries are + `status\\0path\\0`. + """ + records: list[tuple[str, str, str | None]] = [] + fields = raw.split(b"\x00") + i = 0 + while i < len(fields): + status_bytes = fields[i] + if not status_bytes: + i += 1 + continue + status = status_bytes.decode("utf-8") + i += 1 + if i >= len(fields): + break + path = fields[i].decode("utf-8") + i += 1 + previous: str | None = None + if status[0] in "RC" and i < len(fields): + previous = fields[i].decode("utf-8") + i += 1 + records.append((status, path, previous)) + return records diff --git a/src/git_auto_tracker/git/status.py b/src/git_auto_tracker/git/status.py new file mode 100644 index 0000000..26d3a60 --- /dev/null +++ b/src/git_auto_tracker/git/status.py @@ -0,0 +1,120 @@ +"""Change detection based on `git status --porcelain -z` (Section 6.2).""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +from git_auto_tracker.changes.models import ChangeType +from git_auto_tracker.errors import GitError +from git_auto_tracker.git.repository import GitCommand + +# XY codes (https://git-scm.com/docs/git-status): +# X = index state, Y = worktree state. +# The tracker commits what `git add` would stage (the X column), but it must +# also surface worktree-only states that plain `git add` would capture. + + +@dataclass(frozen=True) +class RawStatusEntry: + """One entry parsed from `git status --porcelain -z`.""" + + x: str + y: str + path: str + previous_path: str | None + + @property + def change_type(self) -> ChangeType: + """Map the XY code to the tracker's ChangeType (Section 6.1). + + Priority follows what `git add -A` would stage for the path: + untracked files are additions, deletions win over index states, + and renames are detected from the pair entry. + """ + if self.x == "?" and self.y == "?": + return ChangeType.ADDED + if "R" in (self.x, self.y): + return ChangeType.RENAMED + if "D" in (self.x, self.y): + return ChangeType.DELETED + if "A" in (self.x, self.y): + return ChangeType.ADDED + return ChangeType.MODIFIED + + @classmethod + def from_zy_pair( + cls, + xy: str, + path: str, + previous_path: str | None, + ) -> RawStatusEntry: + if len(xy) != 2: + raise GitError(f"unexpected XY code {xy!r} in git status output") + return cls(x=xy[0], y=xy[1], path=path, previous_path=previous_path) + + +def parse_porcelain_z(raw: bytes) -> list[RawStatusEntry]: + """Parse `git status --porcelain -z` output bytes. + + Format: `XY path\\0` and, for renames/copies, `XY new\\0old\\0` + (current path first, then the previous path). Trailing NULs + terminate the stream. + """ + # Valid XY column values: space = unmodified, ? = untracked, + # A/M/D/T = added/modified/deleted/typechanged, R/C = renamed/copied, + # U = unmerged, ! = ignored. + valid = " ARCDUTM?!" + entries: list[RawStatusEntry] = [] + fields = raw.split(b"\x00") + i = 0 + while i < len(fields): + chunk = fields[i] + if len(chunk) < 4: + # Empty trailing chunk, stray separator, or malformed short entry + i += 1 + continue + entry = chunk.decode("utf-8") + xy = entry[:2] + if xy[0] not in valid or xy[1] not in valid: + raise GitError(f"unexpected XY code {xy!r} in git status output") + path = entry[3:] + previous: str | None = None + # Rename/copy entries are followed by the original path + if xy[0] in "RC" and i + 1 < len(fields): + i += 1 + previous = fields[i].decode("utf-8") + entries.append(RawStatusEntry.from_zy_pair(xy, path, previous)) + i += 1 + return entries + + +class ChangeDetector: + """Detects working-tree changes relative to HEAD (Section 6.1).""" + + def __init__(self, git: GitCommand) -> None: + self.git = git + + def detect(self) -> list[RawStatusEntry]: + # --untracked-files=all: porcelain collapses untracked DIRECTORIES to + # a single entry; the tracker needs per-file records so that staging + # and staged-set verification operate on identical path sets. + raw = self.git.run_bytes( + ["status", "--porcelain", "-z", "--untracked-files=all"] + ) + return parse_porcelain_z(raw) + + +def snapshot_file_attributes(record_path: str, repository_root: Path) -> dict[str, object]: + """Capture size/mtime of a file for concurrent-modification checks (Section 20).""" + absolute = repository_root / Path(record_path) + try: + stat = absolute.stat() + except OSError: + return {"exists": False} + return {"exists": True, "size": stat.st_size, "mtime_ns": stat.st_mtime_ns} + + +def attributes_changed(before: dict[str, object], after: dict[str, object]) -> bool: + """True if the file was modified between the two snapshots.""" + return before != after diff --git a/src/git_auto_tracker/locking/__init__.py b/src/git_auto_tracker/locking/__init__.py new file mode 100644 index 0000000..d5386b4 --- /dev/null +++ b/src/git_auto_tracker/locking/__init__.py @@ -0,0 +1 @@ +"""Repository locking (Section 21).""" diff --git a/src/git_auto_tracker/locking/repository_lock.py b/src/git_auto_tracker/locking/repository_lock.py new file mode 100644 index 0000000..a332a23 --- /dev/null +++ b/src/git_auto_tracker/locking/repository_lock.py @@ -0,0 +1,155 @@ +"""Repository-level execution lock (Section 21). + +The lock file lives OUTSIDE the tracked repository (as a sibling of the +repository directory), so it never shows up in `git status` (FR-028). +Stale locks are detected via live-process check (same host) or an age +threshold (cross-host); removal is logged. +""" + +from __future__ import annotations + +import contextlib +import datetime as dt +import json +import os +import socket +import time +from pathlib import Path + +from git_auto_tracker.errors import LockError + + +def default_lock_path(repository_path: Path) -> Path: + """Sibling lock file: .git-auto-tracker.lock (Section 21).""" + return repository_path.with_name(repository_path.name + ".git-auto-tracker.lock") + + +def _process_alive(pid: int) -> bool: + try: + if os.name == "nt": + # Windows: Process existence check without psutil. + import ctypes + + PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + STILL_ACTIVE = 259 + handle = ctypes.windll.kernel32.OpenProcess( # type: ignore[union-attr] + PROCESS_QUERY_LIMITED_INFORMATION, False, pid + ) + if not handle: + return False + try: + exit_code = ctypes.c_ulong() + if ctypes.windll.kernel32.GetExitCodeProcess( # type: ignore[union-attr] + handle, ctypes.byref(exit_code) + ): + return exit_code.value == STILL_ACTIVE + return False + finally: + ctypes.windll.kernel32.CloseHandle(handle) # type: ignore[union-attr] + # POSIX + os.kill(pid, 0) + return True + except (OSError, AttributeError, ImportError): + return False + + +class RepositoryLock: + """Advisory repository lock guarded by create-exclusive semantics.""" + + def __init__( + self, + lock_path: Path, + stale_age_seconds: int = 6 * 3600, + clock: type[dt.datetime] = dt.datetime, + ) -> None: + self.lock_path = lock_path + self.stale_age_seconds = stale_age_seconds + self._clock = clock + self._acquired = False + + def acquire(self, pid: int | None = None, hostname: str | None = None) -> None: + """Acquire the lock; raises LockError if held by a live execution.""" + if self._acquired: + return + info = { + "process_id": pid if pid is not None else os.getpid(), + "hostname": hostname or socket.gethostname(), + "start_time": self._clock.now(dt.UTC).isoformat(), + } + parent = self.lock_path.parent + parent.mkdir(parents=True, exist_ok=True) + try: + fd = os.open(self.lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) + except FileExistsError: + self._handle_existing_lock() + # Retry once after stale-lock recovery + try: + fd = os.open(self.lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644) + except FileExistsError: + raise LockError( + f"repository locked by another execution: {self.lock_path}" + ) from None + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(info, handle) + self._acquired = True + + def _handle_existing_lock(self) -> None: + """Inspect an existing lock; remove it when stale.""" + try: + raw = self.lock_path.read_text(encoding="utf-8") + info = json.loads(raw) + except (OSError, ValueError): + # Corrupt lock: treat as stale (best effort recovery) + self._remove_quietly() + return + pid = info.get("process_id") + hostname = info.get("hostname") + start_time = info.get("start_time") + if ( + isinstance(pid, int) + and hostname == socket.gethostname() + and _process_alive(pid) + ): + raise LockError(f"repository locked by live process {pid} on {hostname}") + if isinstance(start_time, str): + try: + started = dt.datetime.fromisoformat(start_time) + age = (self._clock.now(dt.UTC) - started).total_seconds() + if age < self.stale_age_seconds: + # Young lock from a dead/foreign process: refuse to steal it + raise LockError( + f"repository locked (young lock, age {int(age)}s < " + f"{self.stale_age_seconds}s): {self.lock_path}" + ) + except ValueError: + pass + # Lock is old or its process is gone: stale, remove it + self._remove_quietly() + + def _remove_quietly(self) -> None: + with contextlib.suppress(OSError): + self.lock_path.unlink(missing_ok=True) + + def release(self) -> None: + """Release the lock; safe to call multiple times.""" + if not self._acquired: + return + self._remove_quietly() + self._acquired = False + + def __enter__(self) -> RepositoryLock: + self.acquire() + return self + + def __exit__(self, *exc: object) -> None: + self.release() + + +def wait_for_lock_release(lock_path: Path, timeout_seconds: float = 5.0) -> bool: + """Sleep-poll helper for tests; returns True if the lock disappeared.""" + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + if not lock_path.exists(): + return True + time.sleep(0.05) + return not lock_path.exists() diff --git a/src/git_auto_tracker/logging/__init__.py b/src/git_auto_tracker/logging/__init__.py new file mode 100644 index 0000000..def83d6 --- /dev/null +++ b/src/git_auto_tracker/logging/__init__.py @@ -0,0 +1 @@ +"""Logging and audit (Sections 22 and 23).""" diff --git a/src/git_auto_tracker/logging/setup.py b/src/git_auto_tracker/logging/setup.py new file mode 100644 index 0000000..e5c78e9 --- /dev/null +++ b/src/git_auto_tracker/logging/setup.py @@ -0,0 +1,88 @@ +"""Logging setup with rotation (Sections 22 and 23). + +Relative log paths resolve against the configuration file location. +The log file must not live inside the tracked repository (Section 23). +""" + +from __future__ import annotations + +import datetime as dt +import time +import logging +import logging.handlers +import zoneinfo +from pathlib import Path + +from git_auto_tracker.config import LoggingConfig + +_FORMAT = "%(asctime)s %(levelname)s %(message)s" +_DATE_FORMAT = "%Y-%m-%d %H:%M:%S%z" +_LOGGER_NAME = "git_auto_tracker" + + +def _resolve_zone(name: str | None) -> dt.tzinfo | None: + """Return the tzinfo for a configured timezone name, None for system time.""" + if not name: + return None + try: + return zoneinfo.ZoneInfo(name) + except zoneinfo.ZoneInfoNotFoundError: + raise ValueError(f"unknown timezone: {name}") from None + + +def setup_logging( + config: LoggingConfig, + config_dir: Path, + repository_path: Path | None = None, + override_level: str | None = None, +) -> logging.Logger: + """Configure and return the application logger. + + A StreamHandler (console) and/or RotatingFileHandler are attached per + configuration. Relative file paths resolve against the configuration + file location (Section 23). + """ + logger = logging.getLogger(_LOGGER_NAME) + logger.setLevel(getattr(logging, (override_level or config.level).upper(), logging.INFO)) + logger.propagate = False + + # Reset handlers so repeated setup (tests) does not duplicate output + for handler in list(logger.handlers): + logger.removeHandler(handler) + handler.close() + + formatter = logging.Formatter(_FORMAT, datefmt=_DATE_FORMAT) + + if config.console: + console = logging.StreamHandler() + console.setFormatter(formatter) + logger.addHandler(console) + + if config.file: + log_path = Path(config.file) + if not log_path.is_absolute(): + log_path = (config_dir / log_path).resolve() + if repository_path is not None: + try: + log_path.relative_to(repository_path.resolve()) + except ValueError: + pass # outside the repository: fine + else: + logging.getLogger(_LOGGER_NAME).warning( + "log file is inside the tracked repository: %s (Section 23)", log_path + ) + log_path.parent.mkdir(parents=True, exist_ok=True) + rotating = logging.handlers.RotatingFileHandler( + log_path, + maxBytes=config.max_size_mb * 1024 * 1024, + backupCount=config.backup_count, + encoding="utf-8", + ) + rotating.setFormatter(formatter) + logger.addHandler(rotating) + + return logger + + +def get_logger() -> logging.Logger: + return logging.getLogger(_LOGGER_NAME) diff --git a/src/git_auto_tracker/security/__init__.py b/src/git_auto_tracker/security/__init__.py new file mode 100644 index 0000000..4fd700e --- /dev/null +++ b/src/git_auto_tracker/security/__init__.py @@ -0,0 +1 @@ +"""Security validation (Section 33).""" diff --git a/src/git_auto_tracker/security/validator.py b/src/git_auto_tracker/security/validator.py new file mode 100644 index 0000000..29c53af --- /dev/null +++ b/src/git_auto_tracker/security/validator.py @@ -0,0 +1,32 @@ +"""Security validation entry point (Section 33). + +Delegates to changes.filters for pattern matching; kept as a separate +component so future policies (skip, quarantine) can evolve independently. +""" + +from __future__ import annotations + +from git_auto_tracker.changes.filters import find_forbidden, validate_security +from git_auto_tracker.changes.models import ChangeSet +from git_auto_tracker.config import SecurityConfig + +__all__ = ["SecurityValidator", "find_forbidden", "validate_security"] + + +class SecurityValidator: + """Checks a ChangeSet against configured forbidden patterns.""" + + def __init__(self, security: SecurityConfig) -> None: + self._security = security + + def find_forbidden(self, change_set: ChangeSet) -> list[str]: + return find_forbidden( + [r.path for r in change_set.all_records()] + [ + r.previous_path or r.path for r in change_set.renamed + ], + self._security, + ) + + def validate(self, change_set: ChangeSet) -> None: + """Raise SecurityError when a forbidden file is present.""" + validate_security(change_set, self._security) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..716f82e --- /dev/null +++ b/tests/conftest.py @@ -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() diff --git a/tests/integration/test_run_cycle.py b/tests/integration/test_run_cycle.py new file mode 100644 index 0000000..fbe408b --- /dev/null +++ b/tests/integration/test_run_cycle.py @@ -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("", 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 " + + def test_sidecar_attribution_appears_in_commit( + self, git_repo: Path, tracker_config: dict, tmp_path: Path + ) -> None: + (git_repo / "config.xml").write_text("", 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("", 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("", 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("", 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("", 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("", 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 diff --git a/tests/unit/test_authors.py b/tests/unit/test_authors.py new file mode 100644 index 0000000..be2f86a --- /dev/null +++ b/tests/unit/test_authors.py @@ -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("", 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) diff --git a/tests/unit/test_filters_security.py b/tests/unit/test_filters_security.py new file mode 100644 index 0000000..1f91542 --- /dev/null +++ b/tests/unit/test_filters_security.py @@ -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) diff --git a/tests/unit/test_lock_config_message.py b/tests/unit/test_lock_config_message.py new file mode 100644 index 0000000..ebc34b0 --- /dev/null +++ b/tests/unit/test_lock_config_message.py @@ -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"] == "" + assert redacted["remote"]["password"] == "" + 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 diff --git a/tests/unit/test_status_parsing.py b/tests/unit/test_status_parsing.py new file mode 100644 index 0000000..28553c5 --- /dev/null +++ b/tests/unit/test_status_parsing.py @@ -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