Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
78 changes: 74 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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
Expand Down
2 changes: 2 additions & 0 deletions src/archunitpython/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from archunitpython.common import (
CheckOptions,
EmptyTestViolation,
LoggingOptions,
TechnicalError,
UserError,
Violation,
Expand Down Expand Up @@ -50,6 +51,7 @@
"Violation",
"EmptyTestViolation",
"CheckOptions",
"LoggingOptions",
"TechnicalError",
"UserError",
"extract_graph",
Expand Down
73 changes: 45 additions & 28 deletions src/archunitpython/common/extraction/extract_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]

Expand Down Expand Up @@ -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
)

Expand Down Expand Up @@ -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],
Expand Down Expand Up @@ -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(
Expand All @@ -179,17 +205,21 @@ 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
):
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(
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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 = (
Expand All @@ -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(
Expand All @@ -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]:
Expand Down Expand Up @@ -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)

Expand Down
79 changes: 79 additions & 0 deletions src/archunitpython/common/fluentapi/inspection.py
Original file line number Diff line number Diff line change
@@ -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 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(
"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)
Loading
Loading