From cf837dba5bef3d0b07e804ff29f0b58d841394bd Mon Sep 17 00:00:00 2001 From: Lukas Niessen Date: Wed, 30 Sep 2026 17:51:51 +0200 Subject: [PATCH 1/3] feat: add per-check inspection and styled assertion output --- src/archunitpython/__init__.py | 2 + .../common/extraction/extract_graph.py | 73 +++--- .../common/fluentapi/inspection.py | 79 +++++++ .../common/logging/inspection.py | 94 ++++++++ src/archunitpython/common/logging/types.py | 2 + src/archunitpython/common/pattern_matching.py | 9 +- .../common/projection/project_edges.py | 13 +- src/archunitpython/files/fluentapi/files.py | 6 + src/archunitpython/layers/fluentapi/layers.py | 2 + .../metrics/extraction/extract_class_info.py | 17 ++ .../metrics/fluentapi/metrics.py | 35 +++ src/archunitpython/slices/fluentapi/slices.py | 3 + src/archunitpython/testing/assertion.py | 22 +- .../testing/common/color_utils.py | 13 +- tests/common/test_inspection.py | 213 ++++++++++++++++++ tests/common/test_pretty_output.py | 51 +++++ 16 files changed, 596 insertions(+), 38 deletions(-) create mode 100644 src/archunitpython/common/fluentapi/inspection.py create mode 100644 src/archunitpython/common/logging/inspection.py create mode 100644 tests/common/test_inspection.py create mode 100644 tests/common/test_pretty_output.py diff --git a/src/archunitpython/__init__.py b/src/archunitpython/__init__.py index 634f5c4..b771c39 100644 --- a/src/archunitpython/__init__.py +++ b/src/archunitpython/__init__.py @@ -7,6 +7,7 @@ from archunitpython.common import ( CheckOptions, EmptyTestViolation, + LoggingOptions, TechnicalError, UserError, Violation, @@ -50,6 +51,7 @@ "Violation", "EmptyTestViolation", "CheckOptions", + "LoggingOptions", "TechnicalError", "UserError", "extract_graph", diff --git a/src/archunitpython/common/extraction/extract_graph.py b/src/archunitpython/common/extraction/extract_graph.py index e0ddc14..80a397b 100644 --- a/src/archunitpython/common/extraction/extract_graph.py +++ b/src/archunitpython/common/extraction/extract_graph.py @@ -9,6 +9,7 @@ from archunitpython.common.extraction.graph import Edge, Graph, ImportKind from archunitpython.common.fluentapi.checkable import CheckOptions +from archunitpython.common.logging.inspection import debug, debug_enabled GraphCacheKey = tuple[str, tuple[str, ...], bool] @@ -57,8 +58,7 @@ def matches(self, import_: _LocatedImport) -> bool: if not self.modules: return True return any( - import_.module_name == module - or import_.module_name.startswith(f"{module}.") + import_.module_name == module or import_.module_name.startswith(f"{module}.") for module in self.modules ) @@ -97,21 +97,46 @@ def extract_graph( ignore_type_checking_imports = bool(options and options.ignore_type_checking_imports) cache_key = _build_cache_key(project_path, excludes, ignore_type_checking_imports) + debug( + "Extraction root: %s; excludes: %s; ignore type-only imports: %s", + project_path, + excludes, + ignore_type_checking_imports, + ) if options and options.clear_cache: + debug("Clearing graph cache for %s", project_path) _graph_cache.pop(cache_key, None) if cache_key in _graph_cache: + debug("Graph cache hit: %d edges", len(_graph_cache[cache_key])) + _inspect_graph(_graph_cache[cache_key]) return _graph_cache[cache_key] + debug("Graph cache miss") result = _extract_graph_uncached( project_path, excludes, ignore_type_checking_imports=ignore_type_checking_imports, ) + _inspect_graph(result) _graph_cache[cache_key] = result return result +def _inspect_graph(graph: Graph) -> None: + if not debug_enabled(): + return + debug("Extracted graph: %d edges (including self-edges)", len(graph)) + for edge in graph: + debug( + "Dependency: %s -> %s; external=%s; kinds=%s", + edge.source, + edge.target, + edge.external, + edge.import_kinds, + ) + + def _build_cache_key( project_path: str, exclude_patterns: list[str], @@ -167,6 +192,7 @@ def _extract_graph_uncached( normalized_py_file_set = {_normalize(f) for f in py_files_set} for file_path in py_files: + debug("Processing file: %s", file_path) # Add self-referencing edge (ensures the file appears as a node) edges.append( Edge( @@ -179,10 +205,13 @@ def _extract_graph_uncached( imports = _extract_located_imports(file_path) for located_import in imports: import_kind = located_import.import_kind - if ( - ignore_type_checking_imports - and import_kind == ImportKind.TYPE_IMPORT - ): + if ignore_type_checking_imports and import_kind == ImportKind.TYPE_IMPORT: + debug( + "Skipped type-only import: %s:%s -> %s", + file_path, + located_import.line_number, + located_import.module_name, + ) continue for resolved, is_external in _resolve_import_targets( located_import, file_path, project_path @@ -190,6 +219,7 @@ def _extract_graph_uncached( if resolved and resolved != _normalize(file_path): # Check if the resolved path is in our project if not is_external and resolved not in normalized_py_file_set: + debug("Skipped excluded dependency: %s -> %s", file_path, resolved) continue edges.append( @@ -227,6 +257,7 @@ def _find_python_files(root: str, exclude: list[str]) -> list[str]: full_path, root, exclude, is_dir=False ): py_files.append(os.path.abspath(full_path)) + debug("Discovered file: %s", full_path) return py_files @@ -313,9 +344,7 @@ def _extract_located_imports(file_path: str) -> list[_LocatedImport]: conditional_import_ranges, ) for alias in node.names: - imports.append( - _LocatedImport(alias.name, kind, node.lineno, syntax_kind) - ) + imports.append(_LocatedImport(alias.name, kind, node.lineno, syntax_kind)) elif isinstance(node, ast.ImportFrom): syntax_kind = ( @@ -329,9 +358,7 @@ def _extract_located_imports(file_path: str) -> list[_LocatedImport]: type_checking_ranges, conditional_import_ranges, ) - fallback_module_name = ( - "." * node.level if node.level and node.module is None else None - ) + fallback_module_name = "." * node.level if node.level and node.module is None else None aliases = _module_aliases(node) if node.module else () for module_name in _import_from_module_names(node): imports.append( @@ -354,15 +381,9 @@ def _extract_located_imports(file_path: str) -> list[_LocatedImport]: conditional_import_ranges, ) for module_name in _extract_dynamic_import_names(node): - imports.append( - _LocatedImport(module_name, kind, node.lineno, syntax_kind) - ) + imports.append(_LocatedImport(module_name, kind, node.lineno, syntax_kind)) - return [ - import_ - for import_ in imports - if not _is_ignored_import(import_, ignore_directives) - ] + return [import_ for import_ in imports if not _is_ignored_import(import_, ignore_directives)] def _find_ignore_directives(source: str) -> dict[int, _IgnoreDirective]: @@ -555,14 +576,10 @@ def _resolve_import( Returns (resolved_path, is_external). The path is normalized with forward slashes. """ - if ( - kind - in ( - ImportKind.RELATIVE_IMPORT, - ImportKind.TYPE_IMPORT, - ) - and import_name.startswith(".") - ): + if kind in ( + ImportKind.RELATIVE_IMPORT, + ImportKind.TYPE_IMPORT, + ) and import_name.startswith("."): # Relative import return _resolve_relative_import(import_name, source_file, project_root) diff --git a/src/archunitpython/common/fluentapi/inspection.py b/src/archunitpython/common/fluentapi/inspection.py new file mode 100644 index 0000000..b0b57ab --- /dev/null +++ b/src/archunitpython/common/fluentapi/inspection.py @@ -0,0 +1,79 @@ +"""Observational instrumentation for the existing check contract.""" + +from __future__ import annotations + +from functools import wraps +from time import perf_counter +from typing import Callable, TypeVar, cast + +from archunitpython.common.assertion.violation import Violation +from archunitpython.common.fluentapi.checkable import CheckOptions +from archunitpython.common.logging.inspection import InspectionSession, current_session + +F = TypeVar("F", bound=Callable[..., list[Violation]]) + + +def inspect_check(check: F) -> F: + """Trace a check while preserving its options, results, and exceptions.""" + + @wraps(check) + def inspected(self: object, options: CheckOptions | None = None) -> list[Violation]: + logging_options = options.logging if options else None + session = ( + InspectionSession(logging_options) + if logging_options and logging_options.enabled + else None + ) + token = current_session.set(session) + started = perf_counter() if session else 0.0 + rule_name = type(self).__name__ + try: + if session: + session.emit("info", "Starting check: %s", rule_name) + # Describe only known rule configuration, never user callbacks. + for name in ( + "_project_path", + "_filters", + "_subject_filters", + "_object_filters", + "_module_filters", + "_pre_filters", + "_check_filters", + "_is_negated", + "_pattern", + "_regex", + "_source", + "_target", + "_layers", + "_allowed_dependencies", + "_forbidden_dependencies", + "_threshold", + "_comparison", + "_metric_attr", + "_name", + "_zone_type", + ): + if hasattr(self, name): + session.emit("debug", "%s: %s", name.removeprefix("_"), getattr(self, name)) + violations = check(self, options) + if session: + session.emit( + "warn" if violations else "info", + "Finished check: %s - %d violation(s) (%.3fs)", + rule_name, + len(violations), + perf_counter() - started, + ) + for violation in violations: + session.emit("warn", "Violation: %s", violation) + return violations + except Exception as error: + if session: + session.emit("error", "Check failed: %s - %s", rule_name, error) + raise + finally: + current_session.reset(token) + if session: + session.close() + + return cast(F, inspected) diff --git a/src/archunitpython/common/logging/inspection.py b/src/archunitpython/common/logging/inspection.py new file mode 100644 index 0000000..dd1eedf --- /dev/null +++ b/src/archunitpython/common/logging/inspection.py @@ -0,0 +1,94 @@ +"""Per-check diagnostic sessions, isolated from architecture evaluation.""" + +from __future__ import annotations + +import logging +import sys +from contextvars import ContextVar +from datetime import datetime +from pathlib import Path +from typing import TextIO + +from archunitpython.common.logging.types import LoggingOptions + +_LEVELS = { + "debug": logging.DEBUG, + "info": logging.INFO, + "warn": logging.WARNING, + "error": logging.ERROR, +} + + +class InspectionSession: + """Own a check's output without attaching process-wide logging handlers. + + Diagnostics are best effort: formatting and output errors must never replace + a rule result or an exception raised by the analysis itself. + """ + + def __init__(self, options: LoggingOptions) -> None: + self.options = options + self._file: TextIO | None = None + self._file_attempted = False + + def emit(self, level: str, message: str, *args: object) -> None: + if _LEVELS[level] < _LEVELS.get(self.options.level, logging.INFO): + return + try: + text = message % args if args else message + except Exception: + return + if self.options.console: + try: + logger = logging.getLogger("archunitpython") + if logger.hasHandlers(): + logger.log(_LEVELS[level], text) + else: + sys.stderr.write(f"[{level.upper()}] {text}\n") + except Exception: + pass + if self.options.log_file: + try: + if not self._file_attempted: + self._file_attempted = True + path = ( + Path(self.options.log_path) + if self.options.log_path + else Path("logs") + / datetime.now().strftime("archunit-%Y-%m-%d_%H-%M-%S-%f.log") + ) + path.parent.mkdir(parents=True, exist_ok=True) + self._file = path.open( + "a" if self.options.append_to_log_file else "w", encoding="utf-8" + ) + if self._file is not None: + self._file.write(f"[{datetime.now().isoformat()}] [{level.upper()}] {text}\n") + self._file.flush() + except Exception: + self.close() + + def close(self) -> None: + if self._file is not None: + try: + self._file.close() + except Exception: + pass + self._file = None + + +current_session: ContextVar[InspectionSession | None] = ContextVar( + "archunitpython_inspection", default=None +) + + +def debug(message: str, *args: object) -> None: + """Lazily emit details only inside an enabled debug check.""" + session = current_session.get() + if session is not None: + session.emit("debug", message, *args) + + +def debug_enabled() -> bool: + """Guard expensive diagnostic traversal before constructing its details.""" + session = current_session.get() + return session is not None and session.options.level == "debug" diff --git a/src/archunitpython/common/logging/types.py b/src/archunitpython/common/logging/types.py index fc0d744..72ade6a 100644 --- a/src/archunitpython/common/logging/types.py +++ b/src/archunitpython/common/logging/types.py @@ -16,3 +16,5 @@ class LoggingOptions: level: LogLevel = "info" log_file: bool = False append_to_log_file: bool = False + console: bool = True + log_path: str | None = None diff --git a/src/archunitpython/common/pattern_matching.py b/src/archunitpython/common/pattern_matching.py index 1956602..98b1c48 100644 --- a/src/archunitpython/common/pattern_matching.py +++ b/src/archunitpython/common/pattern_matching.py @@ -2,6 +2,7 @@ from __future__ import annotations +from archunitpython.common.logging.inspection import debug from archunitpython.common.types import Filter @@ -47,7 +48,9 @@ def matches_pattern(file_path: str, filter_: Filter) -> bool: else: target_string = normalize_path(file_path) - return bool(filter_.regexp.search(target_string)) + matched = bool(filter_.regexp.search(target_string)) + debug("Selector %s against %s (%s): %s", filter_.regexp.pattern, target_string, target, matched) + return matched def matches_pattern_classname(class_name: str, file_path: str, filter_: Filter) -> bool: @@ -65,7 +68,9 @@ def matches_pattern_classname(class_name: str, file_path: str, filter_: Filter) else: target_string = normalize_path(file_path) - return bool(filter_.regexp.search(target_string)) + matched = bool(filter_.regexp.search(target_string)) + debug("Selector %s against %s (%s): %s", filter_.regexp.pattern, target_string, target, matched) + return matched def matches_all_patterns(file_path: str, filters: list[Filter]) -> bool: diff --git a/src/archunitpython/common/projection/project_edges.py b/src/archunitpython/common/projection/project_edges.py index 8851cd3..a3c853d 100644 --- a/src/archunitpython/common/projection/project_edges.py +++ b/src/archunitpython/common/projection/project_edges.py @@ -5,6 +5,7 @@ from collections import defaultdict from archunitpython.common.extraction.graph import Edge +from archunitpython.common.logging.inspection import debug from archunitpython.common.projection.types import MapFunction, ProjectedEdge @@ -29,11 +30,19 @@ def project_edges( for edge in graph: mapped = mapper(edge) if mapped is None: + debug("Projection omitted: %s -> %s", edge.source, edge.target) continue + debug( + "Projection: %s -> %s becomes %s -> %s", + edge.source, + edge.target, + mapped.source_label, + mapped.target_label, + ) key = (mapped.source_label, mapped.target_label) groups[key].append(edge) - return [ + projected = [ ProjectedEdge( source_label=source_label, target_label=target_label, @@ -41,3 +50,5 @@ def project_edges( ) for (source_label, target_label), edges in groups.items() ] + debug("Projection complete: %d raw edges -> %d projected edges", len(graph), len(projected)) + return projected diff --git a/src/archunitpython/files/fluentapi/files.py b/src/archunitpython/files/fluentapi/files.py index 422ce32..90df29b 100644 --- a/src/archunitpython/files/fluentapi/files.py +++ b/src/archunitpython/files/fluentapi/files.py @@ -15,6 +15,7 @@ from archunitpython.common.assertion.violation import EmptyTestViolation, Violation from archunitpython.common.extraction.extract_graph import extract_graph from archunitpython.common.fluentapi.checkable import CheckOptions, RuleRationaleMixin +from archunitpython.common.fluentapi.inspection import inspect_check from archunitpython.common.pattern_matching import matches_all_patterns from archunitpython.common.projection.edge_projections import ( per_external_edge, @@ -329,6 +330,7 @@ def __init__(self, project_path: str | None, filters: list[Filter]) -> None: self._project_path = project_path self._filters = filters + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: graph = extract_graph(self._project_path, options=options) edges = project_edges(graph, per_internal_edge()) @@ -365,6 +367,7 @@ def __init__( self._object_filters = object_filters self._is_negated = is_negated + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: graph = extract_graph(self._project_path, options=options) edges = project_edges(graph, per_internal_edge()) @@ -397,6 +400,7 @@ def matching(self, module_name: Pattern) -> "DependOnExternalModuleCondition": self._module_filters.append(RegexFactory.path_matcher(module_name)) return self + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: graph = extract_graph(self._project_path, options=options) edges = project_edges(graph, per_external_edge()) @@ -424,6 +428,7 @@ def __init__( self._check_filters = check_filters self._is_negated = is_negated + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: nodes = _get_filtered_nodes(self._project_path, self._pre_filters, options) @@ -451,6 +456,7 @@ def __init__( self._message = message self._is_negated = is_negated + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: graph = extract_graph(self._project_path, options=options) nodes = project_to_nodes(graph) diff --git a/src/archunitpython/layers/fluentapi/layers.py b/src/archunitpython/layers/fluentapi/layers.py index 267600a..9e56ede 100644 --- a/src/archunitpython/layers/fluentapi/layers.py +++ b/src/archunitpython/layers/fluentapi/layers.py @@ -5,6 +5,7 @@ from archunitpython.common.assertion.violation import Violation from archunitpython.common.extraction.extract_graph import extract_graph from archunitpython.common.fluentapi.checkable import CheckOptions +from archunitpython.common.fluentapi.inspection import inspect_check from archunitpython.common.projection.edge_projections import per_internal_edge from archunitpython.common.projection.project_edges import project_edges from archunitpython.common.regex_factory import RegexFactory @@ -42,6 +43,7 @@ def where_layer(self, name: str) -> "LayerDependencyRuleBuilder": self._layers.setdefault(name, []) return LayerDependencyRuleBuilder(self, name) + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: graph = extract_graph(self._project_path, options=options) edges = project_edges(graph, per_internal_edge()) diff --git a/src/archunitpython/metrics/extraction/extract_class_info.py b/src/archunitpython/metrics/extraction/extract_class_info.py index e8fdd32..4f81034 100644 --- a/src/archunitpython/metrics/extraction/extract_class_info.py +++ b/src/archunitpython/metrics/extraction/extract_class_info.py @@ -9,6 +9,7 @@ _find_python_files, _resolve_exclude_patterns, ) +from archunitpython.common.logging.inspection import debug from archunitpython.metrics.common.types import ( ClassInfo, EnhancedClassInfo, @@ -41,6 +42,7 @@ def extract_class_info( classes: list[ClassInfo] = [] for file_path in py_files: + debug("Analyzing classes in: %s", file_path) classes.extend(_process_source_file(file_path)) return classes @@ -61,6 +63,7 @@ def extract_enhanced_class_info( results: list[FileAnalysisResult] = [] for file_path in py_files: + debug("Analyzing classes in: %s", file_path) result = _process_source_file_enhanced(file_path) if result.classes: results.append(result) @@ -88,6 +91,13 @@ def _process_source_file(file_path: str) -> list[ClassInfo]: if isinstance(node, ast.ClassDef): class_info = _extract_class(node, norm_path) classes.append(class_info) + debug( + "Class: %s in %s; %d methods; %d fields", + class_info.name, + norm_path, + len(class_info.methods), + len(class_info.fields), + ) return classes @@ -112,6 +122,13 @@ def _process_source_file_enhanced(file_path: str) -> FileAnalysisResult: if isinstance(node, ast.ClassDef): enhanced = _extract_enhanced_class(node, norm_path) result.classes.append(enhanced) + debug( + "Enhanced class: %s in %s; abstract=%s; protocol=%s", + enhanced.name, + norm_path, + enhanced.is_abstract, + enhanced.is_protocol, + ) result.total_types += 1 if enhanced.is_protocol: result.protocols += 1 diff --git a/src/archunitpython/metrics/fluentapi/metrics.py b/src/archunitpython/metrics/fluentapi/metrics.py index f50b937..8c572cc 100644 --- a/src/archunitpython/metrics/fluentapi/metrics.py +++ b/src/archunitpython/metrics/fluentapi/metrics.py @@ -12,6 +12,8 @@ from archunitpython.common.assertion.violation import Violation from archunitpython.common.fluentapi.checkable import CheckOptions, RuleRationaleMixin +from archunitpython.common.fluentapi.inspection import inspect_check +from archunitpython.common.logging.inspection import debug from archunitpython.common.pattern_matching import matches_pattern_classname from archunitpython.common.regex_factory import RegexFactory from archunitpython.common.types import Filter, Pattern @@ -187,12 +189,21 @@ def __init__( self._threshold = threshold self._comparison = comparison + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: classes = _get_filtered_classes(self._project_path, self._filters) violations: list[Violation] = [] for cls in classes: value = self._metric.calculate(cls) + debug( + "Metric %s for %s: value=%s; threshold=%s (%s)", + self._metric.name, + cls.name, + value, + self._threshold, + self._comparison, + ) if check_threshold(value, self._threshold, self._comparison): violations.append( MetricViolation( @@ -247,6 +258,7 @@ def __init__( self._threshold = threshold self._comparison: MetricComparison = comparison + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: import os @@ -268,6 +280,14 @@ def check(self, options: CheckOptions | None = None) -> list[Violation]: continue value = self._metric.calculate_from_file(file_path) + debug( + "Metric %s for %s: value=%s; threshold=%s (%s)", + self._metric.name, + file_path, + value, + self._threshold, + self._comparison, + ) if check_threshold(value, self._threshold, self._comparison): violations.append( FileCountViolation( @@ -381,6 +401,7 @@ def __init__( self._threshold = threshold self._comparison: MetricComparison = comparison + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: files = extract_enhanced_class_info(self._project_path) violations: list[Violation] = [] @@ -388,6 +409,14 @@ def check(self, options: CheckOptions | None = None) -> list[Violation]: for file_result in files: dm = calculate_file_distance_metrics(file_result, files) value = getattr(dm, self._metric_attr) + debug( + "Metric %s for %s: value=%s; threshold=%s (%s)", + self._metric_attr, + file_result.file_path, + value, + self._threshold, + self._comparison, + ) if check_threshold(value, self._threshold, self._comparison): violations.append( @@ -412,6 +441,7 @@ def __init__(self, project_path: str | None, filters: list[Filter], zone_type: s self._filters = filters self._zone_type = zone_type + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: files = extract_enhanced_class_info(self._project_path) violations: list[Violation] = [] @@ -420,6 +450,7 @@ def check(self, options: CheckOptions | None = None) -> list[Violation]: dm = calculate_file_distance_metrics(file_result, files) in_zone = dm.in_zone_of_pain if self._zone_type == "pain" else dm.in_zone_of_uselessness + debug("Zone %s for %s: in_zone=%s", self._zone_type, file_result.file_path, in_zone) if in_zone: violations.append( MetricViolation( @@ -504,12 +535,14 @@ def __init__( self._threshold = threshold self._comparison = comparison + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: classes = _get_filtered_classes(self._project_path, self._filters) violations: list[Violation] = [] for cls in classes: value = self._calculation(cls) + debug("Custom metric %s for %s: value=%s", self._name, cls.name, value) if check_threshold(value, self._threshold, self._comparison): violations.append( MetricViolation( @@ -542,12 +575,14 @@ def __init__( self._calculation = calculation self._assertion = assertion + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: classes = _get_filtered_classes(self._project_path, self._filters) violations: list[Violation] = [] for cls in classes: value = self._calculation(cls) + debug("Custom metric %s for %s: value=%s", self._name, cls.name, value) if not self._assertion(value, cls): violations.append( MetricViolation( diff --git a/src/archunitpython/slices/fluentapi/slices.py b/src/archunitpython/slices/fluentapi/slices.py index 70b0dbb..00a3686 100644 --- a/src/archunitpython/slices/fluentapi/slices.py +++ b/src/archunitpython/slices/fluentapi/slices.py @@ -15,6 +15,7 @@ from archunitpython.common.assertion.violation import Violation from archunitpython.common.extraction.extract_graph import extract_graph from archunitpython.common.fluentapi.checkable import CheckOptions, RuleRationaleMixin +from archunitpython.common.fluentapi.inspection import inspect_check from archunitpython.common.projection.project_edges import project_edges from archunitpython.common.projection.types import MapFunction from archunitpython.slices.assertion.admissible_edges import ( @@ -157,6 +158,7 @@ def __init__( self._puml_content = puml_content self._coherence_options = coherence_options + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: graph = extract_graph(self._project_path, options=options) rules, contained_nodes = generate_rule(self._puml_content) @@ -193,6 +195,7 @@ def __init__( self._source = source self._target = target + @inspect_check def check(self, options: CheckOptions | None = None) -> list[Violation]: graph = extract_graph(self._project_path, options=options) diff --git a/src/archunitpython/testing/assertion.py b/src/archunitpython/testing/assertion.py index 900c2b8..7c37406 100644 --- a/src/archunitpython/testing/assertion.py +++ b/src/archunitpython/testing/assertion.py @@ -4,6 +4,7 @@ from archunitpython.common.assertion.violation import Violation from archunitpython.common.fluentapi.checkable import Checkable, CheckOptions +from archunitpython.testing.common.color_utils import ColorUtils from archunitpython.testing.common.violation_factory import ViolationFactory @@ -11,26 +12,32 @@ def format_violations( violations: list[Violation], *, because: str | None = None, + color: bool | None = None, ) -> str: """Format violations into a human-readable string. Args: violations: List of violations to format. + because: Optional rationale for the rule. + color: None detects interactive terminals; False guarantees plain text. + True explicitly enables ANSI colors, unless NO_COLOR is set. Returns: Formatted string describing all violations. """ if not violations: - return "No violations found." + return ColorUtils.style("No violations found.", "32", color) - lines = [f"Found {len(violations)} architecture violation(s):"] + lines = [ColorUtils.style(f"Found {len(violations)} architecture violation(s):", "1;31", color)] if because: - lines.extend(["", f"Because: {because}"]) + lines.extend(["", ColorUtils.style(f"Because: {because}", "33", color)]) lines.append("") for i, violation in enumerate(violations, 1): tv = ViolationFactory.from_violation(violation) - lines.append(f" {i}. {tv.message}") - lines.append(f" {tv.details}") + lines.append(f" {i}. {ColorUtils.style(tv.message, '1;33', color)}") + detail_color = "36" if "value=" in tv.details else "34" + for detail in tv.details.split("\n"): + lines.append(f" {ColorUtils.style(detail, detail_color, color)}") lines.append("") return "\n".join(lines) @@ -39,12 +46,15 @@ def format_violations( def assert_passes( checkable: Checkable, options: CheckOptions | None = None, + *, + color: bool | None = None, ) -> None: """Assert that an architecture rule passes (no violations). Args: checkable: Any object with a check() method (implements Checkable). options: Optional check options. + color: Optional color override for failure output. Raises: AssertionError: If the rule has violations. @@ -52,4 +62,4 @@ def assert_passes( violations = checkable.check(options) if violations: because = getattr(checkable, "because_reason", None) - raise AssertionError(format_violations(violations, because=because)) + raise AssertionError(format_violations(violations, because=because, color=color)) diff --git a/src/archunitpython/testing/common/color_utils.py b/src/archunitpython/testing/common/color_utils.py index 07a875c..1d9d546 100644 --- a/src/archunitpython/testing/common/color_utils.py +++ b/src/archunitpython/testing/common/color_utils.py @@ -8,7 +8,9 @@ def _supports_color() -> bool: """Check if the terminal supports ANSI colors.""" - if os.environ.get("NO_COLOR"): + if "NO_COLOR" in os.environ or os.environ.get("CI", "").lower() == "true": + return False + if os.environ.get("TERM") == "dumb": return False if not hasattr(sys.stdout, "isatty"): return False @@ -24,6 +26,15 @@ def _wrap(code: str, text: str) -> str: class ColorUtils: """ANSI color utilities for terminal output.""" + @staticmethod + def style(text: str, code: str, enabled: bool | None = None) -> str: + """Style text with automatic detection or an explicit color choice.""" + if enabled is False or "NO_COLOR" in os.environ: + return text + if enabled is None and not _supports_color(): + return text + return f"\033[{code}m{text}\033[0m" + @staticmethod def red(text: str) -> str: return _wrap("31", text) diff --git a/tests/common/test_inspection.py b/tests/common/test_inspection.py new file mode 100644 index 0000000..2b22f71 --- /dev/null +++ b/tests/common/test_inspection.py @@ -0,0 +1,213 @@ +"""Diagnostics must describe execution without affecting its results.""" + +import logging +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Barrier + +import pytest + +from archunitpython import CheckOptions, LoggingOptions, metrics, project_files +from archunitpython.common.fluentapi.inspection import inspect_check +from archunitpython.common.logging.inspection import current_session, debug + + +@pytest.fixture +def project(tmp_path): + (tmp_path / "api.py").write_text("import storage\n", encoding="utf-8") + (tmp_path / "storage.py").write_text( + "class Storage:\n def save(self): pass\n", encoding="utf-8" + ) + return str(tmp_path) + + +def failing_rule(project): + return ( + project_files(project) + .with_name("api.py") + .should_not() + .depend_on_files() + .with_name("storage.py") + ) + + +def test_default_and_explicit_disabled_are_silent(project, caplog, tmp_path): + rule = failing_rule(project) + options = CheckOptions( + logging=LoggingOptions( + enabled=False, log_file=True, log_path=str(tmp_path / "disabled.log") + ) + ) + with caplog.at_level(logging.DEBUG, logger="archunitpython"): + assert rule.check() == rule.check(options) + assert not caplog.records + assert not (tmp_path / "disabled.log").exists() + + +@pytest.mark.parametrize("level", ["debug", "info", "warn", "error"]) +def test_levels_and_unchanged_results(project, caplog, level): + rule = failing_rule(project) + baseline = rule.check() + with caplog.at_level(logging.DEBUG, logger="archunitpython"): + result = rule.check( + CheckOptions(clear_cache=True, logging=LoggingOptions(enabled=True, level=level)) + ) + assert result == baseline + text = caplog.text + assert ("Starting check" in text) == (level in ("debug", "info")) + assert ("Violation:" in text) == (level != "error") + assert ("Discovered file:" in text) == (level == "debug") + if level == "debug": + for detail in ("Processing file:", "Dependency:", "Selector ", "Projection:", "cache miss"): + assert detail in text + + +def test_warm_cache_still_describes_graph(project, caplog): + rule = failing_rule(project) + rule.check() + with caplog.at_level(logging.DEBUG, logger="archunitpython"): + rule.check(CheckOptions(logging=LoggingOptions(enabled=True, level="debug"))) + assert "Graph cache hit" in caplog.text + assert "Dependency:" in caplog.text + + +def test_file_only_output_and_append(project, caplog, tmp_path): + path = tmp_path / "inspection.log" + path.write_text("old\n", encoding="utf-8") + options = CheckOptions( + logging=LoggingOptions( + enabled=True, + level="debug", + console=False, + log_file=True, + log_path=str(path), + append_to_log_file=True, + ) + ) + rule = failing_rule(project) + baseline = rule.check() + assert rule.check(options) == baseline + assert path.read_text(encoding="utf-8").startswith("old\n") + assert "Finished check" in path.read_text(encoding="utf-8") + assert "\x1b" not in path.read_text(encoding="utf-8") + assert not caplog.records + path.rename(tmp_path / "closed.log") # Windows also verifies the handle is closed. + + +def test_unwritable_log_does_not_change_result(project, tmp_path): + rule = failing_rule(project) + baseline = rule.check() + options = CheckOptions( + logging=LoggingOptions(enabled=True, log_file=True, console=False, log_path=str(tmp_path)) + ) + assert rule.check(options) == baseline + + +def test_broken_handler_does_not_change_result(project): + class BrokenHandler(logging.Handler): + def emit(self, record): + raise OSError("sink unavailable") + + logger = logging.getLogger("archunitpython") + handler = BrokenHandler() + baseline = failing_rule(project).check() + logger.addHandler(handler) + try: + assert ( + failing_rule(project).check(CheckOptions(logging=LoggingOptions(enabled=True))) + == baseline + ) + finally: + logger.removeHandler(handler) + + +def test_nested_silent_checks_restore_outer_session(caplog): + class Rule: + @inspect_check + def check(self, options=None): + debug("outer before") + if options: + Rule().check() + debug("outer after") + return [] + + with caplog.at_level(logging.DEBUG, logger="archunitpython"): + Rule().check(CheckOptions(logging=LoggingOptions(enabled=True, level="debug"))) + assert caplog.text.count("outer before") == 1 + assert caplog.text.count("outer after") == 1 + assert current_session.get() is None + + +def test_concurrent_sessions_do_not_mix_files(tmp_path): + barrier = Barrier(2) + + class Rule: + def __init__(self, label): + self.label = label + + @inspect_check + def check(self, options=None): + barrier.wait(timeout=10) + debug("private marker: %s", self.label) + return [] + + def run(label): + path = tmp_path / f"{label}.log" + Rule(label).check( + CheckOptions( + logging=LoggingOptions( + enabled=True, level="debug", console=False, log_file=True, log_path=str(path) + ) + ) + ) + return path.read_text(encoding="utf-8") + + with ThreadPoolExecutor(max_workers=2) as executor: + left, right = list(executor.map(run, ["left", "right"])) + assert "private marker: left" in left and "private marker: right" not in left + assert "private marker: right" in right and "private marker: left" not in right + + +def test_original_exception_is_preserved_and_session_closed(caplog): + error = RuntimeError("analysis failed") + + class Rule: + @inspect_check + def check(self, options=None): + raise error + + with caplog.at_level(logging.DEBUG, logger="archunitpython"): + with pytest.raises(RuntimeError) as caught: + Rule().check(CheckOptions(logging=LoggingOptions(enabled=True, level="error"))) + assert caught.value is error + assert "Check failed" in caplog.text + assert current_session.get() is None + + +def test_custom_metric_is_calculated_once_per_class(project, caplog): + calls = [] + + def calculate(cls): + calls.append(cls.name) + return 7 + + rule = metrics(project).custom_metric("example", "example", calculate).should_be_below(2) + with caplog.at_level(logging.DEBUG, logger="archunitpython"): + violations = rule.check(CheckOptions(logging=LoggingOptions(enabled=True, level="debug"))) + assert calls == ["Storage"] + assert len(violations) == 1 + assert "Custom metric example for Storage: value=7" in caplog.text + + +def test_debug_does_not_render_objects_at_info_level(): + class Unprintable: + def __str__(self): + raise AssertionError("must not stringify disabled debug payload") + + class Rule: + @inspect_check + def check(self, options=None): + debug("payload %s", Unprintable()) + return [] + + assert Rule().check(CheckOptions(logging=LoggingOptions(enabled=True, console=False))) == [] diff --git a/tests/common/test_pretty_output.py b/tests/common/test_pretty_output.py new file mode 100644 index 0000000..4707d23 --- /dev/null +++ b/tests/common/test_pretty_output.py @@ -0,0 +1,51 @@ +import re + +import pytest + +from archunitpython import format_violations +from archunitpython.common.projection.types import ProjectedEdge +from archunitpython.files.assertion.cycle_free import ViolatingCycle + + +@pytest.fixture +def violations(): + return [ + ViolatingCycle( + cycle=[ + ProjectedEdge(source_label="api.py", target_label="db.py"), + ProjectedEdge(source_label="db.py", target_label="api.py"), + ] + ) + ] + + +def test_colored_output_preserves_plain_text(violations, monkeypatch): + monkeypatch.delenv("NO_COLOR", raising=False) + plain = format_violations(violations, because="keep boundaries clear", color=False) + colored = format_violations(violations, because="keep boundaries clear", color=True) + assert "\x1b[1;31m" in colored + assert "\x1b[34m" in colored + assert re.sub(r"\x1b\[[0-9;]*m", "", colored) == plain + assert "1. Circular dependency detected" in plain + assert "Cycle: api.py -> db.py" in plain + + +@pytest.mark.parametrize("environment", [{"NO_COLOR": ""}, {"CI": "true"}, {"TERM": "dumb"}]) +def test_environment_disables_auto_color(violations, monkeypatch, environment): + monkeypatch.setattr("sys.stdout.isatty", lambda: True) + for key, value in environment.items(): + monkeypatch.setenv(key, value) + assert "\x1b[" not in format_violations(violations) + + +def test_redirected_output_is_plain(violations, monkeypatch): + monkeypatch.setattr("sys.stdout.isatty", lambda: False) + assert "\x1b[" not in format_violations(violations) + + +def test_success_color_and_explicit_no_color(monkeypatch): + monkeypatch.delenv("NO_COLOR", raising=False) + assert "\x1b[32m" in format_violations([], color=True) + assert format_violations([], color=False) == "No violations found." + monkeypatch.setenv("NO_COLOR", "") + assert format_violations([], color=True) == "No violations found." From 416ad73a58ab41d9bb6e8f48628444deda504945 Mon Sep 17 00:00:00 2001 From: Lukas Niessen Date: Wed, 30 Sep 2026 17:51:54 +0200 Subject: [PATCH 2/3] docs: explain detailed logging and colored reports --- README.md | 78 ++++++++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 74 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index efeb3ae..76ebef7 100644 --- a/README.md +++ b/README.md @@ -893,6 +893,10 @@ The most important differences: When tests fail, you get helpful output with file paths and violation details: +Interactive terminals also highlight the failure summary, numbered headings, +paths, metric values, and rule rationale. Output stays plain when redirected, +when `CI=true`, or when `NO_COLOR` is set (even to an empty value). + ``` Found 2 architecture violation(s): @@ -905,25 +909,91 @@ Found 2 architecture violation(s): ## 📝 Debug Logging & Configuration -We support logging to help you understand what files are being analyzed and troubleshoot test failures. Logging is disabled by default to keep test output clean. +Logging is disabled by default. Enable it per check to inspect file discovery, +graph extraction and cache use, selector decisions, edge projections, class +analysis, metric values, and the resulting violations. File, layer, slice, +count/LCOM/distance, and custom metric checks use the same logging options. + +| Level | What you see | +| --- | --- | +| `error` | Analysis exceptions (the original exception is still raised). | +| `warn` | Failed check summaries and every violation, plus errors. | +| `info` | Check start, completion, duration, and violations, plus errors. | +| `debug` | All of the above, plus rule configuration, discovered files, dependencies, cache decisions, selector matches, projected/omitted edges, classes, and measured values. | ### Enabling Debug Logging ```python -from archunitpython import CheckOptions -from archunitpython.common.logging.types import LoggingOptions +from archunitpython import CheckOptions, LoggingOptions, project_files + +rule = project_files("src/").should().have_no_cycles() options = CheckOptions( logging=LoggingOptions( enabled=True, level="debug", # "error" | "warn" | "info" | "debug" - log_file=True, # Creates logs/archunit-YYYY-MM-DD_HH-MM-SS.log + log_file=True, # Creates a unique timestamped file in logs/ ), ) violations = rule.check(options) ``` +For example, a debug trace includes entries like these (paths and timings depend +on your project): + +```text +[INFO] Starting check: CycleFreeFileCondition +[DEBUG] Graph cache miss +[DEBUG] Discovered file: /project/src/api.py +[DEBUG] Processing file: /project/src/api.py +[DEBUG] Dependency: /project/src/api.py -> /project/src/storage.py; external=False; kinds=[] +[DEBUG] Projection: /project/src/api.py -> /project/src/storage.py becomes /project/src/api.py -> /project/src/storage.py +[INFO] Finished check: CycleFreeFileCondition - 0 violation(s) (0.002s) +``` + +Console diagnostics use the standard `archunitpython` logger when handlers are +configured, respecting their levels and formatting; otherwise they go to stderr. +For pytest live logs, use `pytest --log-cli-level=DEBUG`. You can also configure +Python's standard `logging` module in your application. + +### Save detailed inspection without console noise + +```python +options = CheckOptions(logging=LoggingOptions( + enabled=True, + level="debug", + console=False, + log_file=True, + log_path="logs/architecture.log", + append_to_log_file=True, +)) +violations = rule.check(options) +``` + +File logs are UTF-8, timestamped, and contain no added ANSI colors. Automatic +paths are unique per check; an explicit `log_path` replaces that path. +`append_to_log_file=False` overwrites an explicit file at the start of each check. +Each check closes its own output file and isolates its diagnostics from nested +and concurrent checks. Use separate paths when concurrent checks write to files. +Diagnostic output is best effort: an unavailable file or failing logging handler +does not change rule results or hide analysis exceptions. Debug inspection does +not re-run custom predicates or metric calculations. + +### Format results independently of logging + +```python +from archunitpython import assert_passes, format_violations + +print(format_violations(violations)) # Detect terminal colors +text = format_violations(violations, color=False) # Stable plain text for artifacts +assert_passes(rule, options, color=False) # Plain assertion failure +``` + +Pass `color=True` to explicitly request ANSI styling; `NO_COLOR` always wins. +The numbered report, rationale, and details remain available at every log level, +including when logging is disabled. Existing plain-text report wording is retained. + ### CI Pipeline Integration ```yaml From efc7d6bffc43743a2171d6be71b501f5f8f2d8d6 Mon Sep 17 00:00:00 2001 From: Lukas Niessen Date: Wed, 30 Sep 2026 17:57:43 +0200 Subject: [PATCH 3/3] fix: keep inspection from invoking optional user properties --- .../common/fluentapi/inspection.py | 4 +-- .../common/logging/inspection.py | 7 +++- .../metrics/fluentapi/metrics.py | 26 +++++++++++---- tests/common/test_inspection.py | 33 ++++++++++++++++++- 4 files changed, 59 insertions(+), 11 deletions(-) diff --git a/src/archunitpython/common/fluentapi/inspection.py b/src/archunitpython/common/fluentapi/inspection.py index b0b57ab..e012744 100644 --- a/src/archunitpython/common/fluentapi/inspection.py +++ b/src/archunitpython/common/fluentapi/inspection.py @@ -53,8 +53,8 @@ def inspected(self: object, options: CheckOptions | None = None) -> list[Violati "_name", "_zone_type", ): - if hasattr(self, name): - session.emit("debug", "%s: %s", name.removeprefix("_"), getattr(self, name)) + if session.options.level == "debug" and name in vars(self): + session.emit("debug", "%s: %s", name.removeprefix("_"), vars(self)[name]) violations = check(self, options) if session: session.emit( diff --git a/src/archunitpython/common/logging/inspection.py b/src/archunitpython/common/logging/inspection.py index dd1eedf..c1484ec 100644 --- a/src/archunitpython/common/logging/inspection.py +++ b/src/archunitpython/common/logging/inspection.py @@ -8,6 +8,7 @@ from datetime import datetime from pathlib import Path from typing import TextIO +from uuid import uuid4 from archunitpython.common.logging.types import LoggingOptions @@ -55,7 +56,11 @@ def emit(self, level: str, message: str, *args: object) -> None: Path(self.options.log_path) if self.options.log_path else Path("logs") - / datetime.now().strftime("archunit-%Y-%m-%d_%H-%M-%S-%f.log") + / ( + datetime.now().strftime("archunit-%Y-%m-%d_%H-%M-%S-") + + uuid4().hex + + ".log" + ) ) path.parent.mkdir(parents=True, exist_ok=True) self._file = path.open( diff --git a/src/archunitpython/metrics/fluentapi/metrics.py b/src/archunitpython/metrics/fluentapi/metrics.py index 8c572cc..2e1e7a8 100644 --- a/src/archunitpython/metrics/fluentapi/metrics.py +++ b/src/archunitpython/metrics/fluentapi/metrics.py @@ -13,7 +13,7 @@ from archunitpython.common.assertion.violation import Violation from archunitpython.common.fluentapi.checkable import CheckOptions, RuleRationaleMixin from archunitpython.common.fluentapi.inspection import inspect_check -from archunitpython.common.logging.inspection import debug +from archunitpython.common.logging.inspection import debug, debug_enabled from archunitpython.common.pattern_matching import matches_pattern_classname from archunitpython.common.regex_factory import RegexFactory from archunitpython.common.types import Filter, Pattern @@ -99,6 +99,20 @@ def custom_metric( ) +def _log_metric_value( + metric: Any, subject: str, value: float, threshold: float, comparison: MetricComparison +) -> None: + if not debug_enabled(): + return + try: + name = metric.name + except Exception: + name = type(metric).__name__ + debug( + "Metric %s for %s: value=%s; threshold=%s (%s)", name, subject, value, threshold, comparison + ) + + def _get_filtered_classes(project_path: str | None, filters: list[Filter]) -> list[ClassInfo]: classes = extract_class_info(project_path) if not filters: @@ -196,9 +210,8 @@ def check(self, options: CheckOptions | None = None) -> list[Violation]: for cls in classes: value = self._metric.calculate(cls) - debug( - "Metric %s for %s: value=%s; threshold=%s (%s)", - self._metric.name, + _log_metric_value( + self._metric, cls.name, value, self._threshold, @@ -280,9 +293,8 @@ def check(self, options: CheckOptions | None = None) -> list[Violation]: continue value = self._metric.calculate_from_file(file_path) - debug( - "Metric %s for %s: value=%s; threshold=%s (%s)", - self._metric.name, + _log_metric_value( + self._metric, file_path, value, self._threshold, diff --git a/tests/common/test_inspection.py b/tests/common/test_inspection.py index 2b22f71..b3cb0a9 100644 --- a/tests/common/test_inspection.py +++ b/tests/common/test_inspection.py @@ -2,7 +2,6 @@ import logging from concurrent.futures import ThreadPoolExecutor -from pathlib import Path from threading import Barrier import pytest @@ -211,3 +210,35 @@ def check(self, options=None): return [] assert Rule().check(CheckOptions(logging=LoggingOptions(enabled=True, console=False))) == [] + + +def test_inspection_does_not_read_rule_properties(caplog): + class Rule: + @property + def _filters(self): + raise AssertionError("inspection must not invoke descriptors") + + @inspect_check + def check(self, options=None): + return [] + + with caplog.at_level(logging.DEBUG, logger="archunitpython"): + assert Rule().check(CheckOptions(logging=LoggingOptions(enabled=True, level="debug"))) == [] + + +def test_passing_metric_does_not_gain_a_name_requirement(project, caplog): + from archunitpython.metrics.fluentapi.metrics import ClassMetricCondition + + class Metric: + @property + def name(self): + raise RuntimeError("name unavailable") + + def calculate(self, cls): + return 0.0 + + rule = ClassMetricCondition(project, [], Metric(), 1, "below") + assert rule.check() == [] + with caplog.at_level(logging.DEBUG, logger="archunitpython"): + assert rule.check(CheckOptions(logging=LoggingOptions(enabled=True, level="debug"))) == [] + assert "Metric Metric for Storage" in caplog.text