Init

ptdiag/ini.py

@@ -0,0 +1,112 @@
+"""Exact, verified INI value inspection and replacement."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+import shutil
+from typing import Any
+
+from .filesystem import sha256_file
+
+
+class IniError(ValueError):
+	"""Raised when an INI override cannot be applied exactly."""
+
+
+def inspect_ini_value(path: Path, section: str, key: str) -> dict[str, Any]:
+	if path.is_symlink() or not path.is_file():
+		raise IniError(f"INI path is missing or unsafe: {path}")
+	text = _read_ini_text(path)
+	_, value = _find_ini_value(text, section, key)
+	state = path.stat()
+	return {
+		"path": str(path),
+		"section": section,
+		"key": key,
+		"value": value,
+		"sha256": sha256_file(path),
+		"size": state.st_size,
+		"mtime_ns": state.st_mtime_ns,
+	}
+
+
+def override_ini_value(
+	path: Path,
+	section: str,
+	key: str,
+	expected: str,
+	value: str,
+) -> dict[str, Any]:
+	before = inspect_ini_value(path, section, key)
+	if before["value"] != expected:
+		raise IniError(
+			f"INI value mismatch for [{section}] {key}: "
+			f"expected {expected!r}, found {before['value']!r}"
+		)
+	text = _read_ini_text(path)
+	line_index, _ = _find_ini_value(text, section, key)
+	lines = text.splitlines(keepends=True)
+	body = lines[line_index].rstrip("\r\n")
+	newline = lines[line_index][len(body) :]
+	left, separator, right = body.partition("=")
+	if not separator:
+		raise IniError(f"INI key has no equals separator: [{section}] {key}")
+	leading = right[: len(right) - len(right.lstrip())]
+	trailing = right[len(right.rstrip()) :]
+	lines[line_index] = f"{left}={leading}{value}{trailing}{newline}"
+	updated = "".join(lines)
+	if updated == text:
+		raise IniError(f"INI override made no change: [{section}] {key}")
+	temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
+	if temporary.exists() or temporary.is_symlink():
+		raise IniError(f"INI temporary path already exists: {temporary}")
+	try:
+		with temporary.open("w", encoding="utf-8", newline="") as stream:
+			stream.write(updated)
+			stream.flush()
+			os.fsync(stream.fileno())
+		shutil.copystat(path, temporary, follow_symlinks=False)
+		temporary.replace(path)
+	finally:
+		if temporary.exists() or temporary.is_symlink():
+			temporary.unlink()
+	after = inspect_ini_value(path, section, key)
+	if after["value"] != value:
+		raise IniError(f"INI override verification failed: [{section}] {key}")
+	if after["sha256"] == before["sha256"]:
+		raise IniError(f"INI hash did not change: [{section}] {key}")
+	return {
+		"ok": True,
+		"before": before,
+		"after": after,
+		"expected": expected,
+		"value": value,
+	}
+
+
+def _read_ini_text(path: Path) -> str:
+	with path.open("r", encoding="utf-8", newline="") as stream:
+		return stream.read()
+
+
+def _find_ini_value(text: str, section: str, key: str) -> tuple[int, str]:
+	current_section: str | None = None
+	matches: list[tuple[int, str]] = []
+	for index, line in enumerate(text.splitlines(keepends=True)):
+		stripped = line.strip()
+		if stripped.startswith("[") and stripped.endswith("]"):
+			current_section = stripped[1:-1].strip()
+			continue
+		if current_section is None or current_section.casefold() != section.casefold():
+			continue
+		if not stripped or stripped.startswith((";", "#")):
+			continue
+		left, separator, right = line.rstrip("\r\n").partition("=")
+		if separator and left.strip().casefold() == key.casefold():
+			matches.append((index, right.strip()))
+	if not matches:
+		raise IniError(f"INI key not found: [{section}] {key}")
+	if len(matches) != 1:
+		raise IniError(f"INI key is duplicated: [{section}] {key}")
+	return matches[0]