feat: add check

.github/workflows/ci.yml

@@ -31,6 +31,9 @@ jobs:
         run: pip install -r requirements.txt -r requirements-dev.txt
 
       - name: lint
+        run: ruff check
+
+      - name: format
         run: ruff format --check
 
       - name: test

Makefile

@@ -0,0 +1,9 @@
+help:
+	@egrep -h '\s##\s' $(MAKEFILE_LIST) | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m  %-30s\033[0m %s\n", $$1, $$2}'
+
+format:  ## format the project
+	ruff format
+	ruff check --fix
+
+test:  ## run unit tests and generate coverage report
+	pytest --cov=./src/ ./tests/

TODO.md

@@ -3,4 +3,12 @@
 - [ ] [fully configure dependabot](https://docs.github.com/en/code-security/dependabot/working-with-dependabot/automating-dependabot-with-github-actions)
 - [ ] configure make file
 - [ ] dprint for other file formatting
-- [ ] automatic release versioning for pypi
+- [ ] automatic release versioning for pypi, run unit tests against package? Is this possible?
+- [ ] isort
+- [ ] configuration file using cli, use toml, custom rc file, env vars?
+
+
+Next steps:
+1. Add `cycl --check` as first feature, publish v1.0.0 to pypi
+2. Add ability to ignore known cycles, how to break the cycle?
+3. Add ability to reduce via tags, check out jquery --reduce-on maybe?

pyproject.toml

@@ -25,6 +25,30 @@ cycl = 'cycl.cli:app'
 [tool.ruff]
 line-length = 125
 
+[tool.ruff.lint]
+# desire is to add COM812 too
+select = ['ALL']
+ignore = [
+    'COM',  # flake8-commas
+    'D',  # pydocstyle
+    'ERA001',  # commented-out-code
+    'FIX002',  # line-contains-todo
+    'Q000',  # bad-quotes-inline-string
+    'Q003',  # avoidable-escaped-quote
+    'T201',  # print
+    'TD',  # flake8-todos
+]
+
+[tool.ruff.lint.per-file-ignores]
+'tests/**' = [
+    'ANN',  # flake8-annotations
+    'PLR2004',  # magic-value-comparison
+    'S101',  # assert
+]
+'__init__.py' = [
+    'F401',  # unused-import
+]
+
 [tool.ruff.format]
 quote-style = 'single'
 indent-style = 'space'
@@ -35,10 +59,13 @@ branch = true
 source = ['./src']
 
 [tool.coverage.report]
-fail_under = 40
+fail_under = 90
 show_missing = true
 skip_covered = true
 skip_empty = true
 omit = [
     '__*__.py'
 ]
+
+[tool.pytest.ini_options]
+log_cli_level = "INFO"

requirements-dev.txt

@@ -1,4 +1,5 @@
 pytest-cov==6.0.0
+pytest-subtests==0.14.1
 pytest-sugar==1.0.0
 pytest==8.3.4
 ruff==0.9.4

src/cycl/__init__.py

@@ -1,5 +1,6 @@
 # import importlib.metadata
 
-from .cycl import get_dependency_graph
+from .cycl import build_dependency_graph
 
+# todo: how to resolve?
 # __version__ = importlib.metadata.version(__package__ or __name__)

src/cycl/cli.py

@@ -1,15 +1,41 @@
 import argparse
+import logging
+import sys
 
+import networkx as nx
+
+from cycl import build_dependency_graph
+from cycl.utils.log_config import configure_logging
+
+
+def app() -> None:
+    parser = argparse.ArgumentParser(description='Check for cross-stack import/export circular dependencies.')
+    subparsers = parser.add_subparsers(dest='action', required=True)
+
+    parent_parser = argparse.ArgumentParser(add_help=False)
+    parent_parser.add_argument('--exit-zero', action='store_true', help='exit 0 regardless of result')
+    parent_parser.add_argument(
+        '--log-level',
+        choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
+        default='ERROR',
+        help='Set the logging level (default: ERROR)',
+    )
+
+    subparsers.add_parser('check', parents=[parent_parser], help='Check for cycles in AWS stack imports/exports')
 
-def app():
-    parser = argparse.ArgumentParser()
-    parser.add_argument('--verbosity', help='increase output verbosity')
     args = parser.parse_args()
+    configure_logging(getattr(logging, args.log_level))
 
-    if args.verbosity:
-        print('verbosity turned on')
-    else:
-        print('not turned on')
+    if args.action == 'check':
+        cycle_found = False
+        graph = build_dependency_graph()
+        cycles = nx.simple_cycles(graph)
+        for cycle in cycles:
+            cycle_found = True
+            print(f'Cycle found between nodes: {cycle}')
+        if cycle_found and not args.exit_zero:
+            sys.exit(1)
+    sys.exit(0)
 
 
 if __name__ == '__main__':

src/cycl/cycl.py

@@ -1,6 +1,25 @@
-def get_dependency_graph():
-    print('some dependency graph')
+from logging import getLogger
 
+import networkx as nx
 
-def main():
-    print('hdi')
+from cycl.utils.cfn import get_all_exports, get_all_imports, parse_name_from_id
+
+log = getLogger(__name__)
+
+
+def build_dependency_graph() -> nx.MultiDiGraph:
+    dep_graph = nx.MultiDiGraph()
+
+    exports = get_all_exports()
+    for export in exports:
+        export['ExportingStackName'] = parse_name_from_id(export['ExportingStackId'])
+        export['ImportingStackNames'] = get_all_imports(export_name=export['Name'])
+        edges = [
+            (export['ExportingStackName'], importing_stack_name) for importing_stack_name in export['ImportingStackNames']
+        ]
+        if edges:
+            dep_graph.add_edges_from(ebunch_to_add=edges)
+        else:
+            log.info('Export found with no import: %s', export['ExportingStackName'])
+            dep_graph.add_node(export['ExportingStackName'])
+    return dep_graph

src/cycl/utils/cfn.py

@@ -1,20 +1,26 @@
-from typing import List, Dict, Optional
+from __future__ import annotations
+
+from logging import getLogger
+from typing import TYPE_CHECKING
 
-from botocore.client import BaseClient
-from botocore.exceptions import ClientError
 import boto3
+from botocore.exceptions import ClientError
+
+if TYPE_CHECKING:
+    from botocore.client import BaseClient
+
+log = getLogger(__name__)
 
 
 def parse_name_from_id(stack_id: str) -> str:
     try:
         return stack_id.split('/')[1]
-    except IndexError as err:
-        # log warning message
-        ...
+    except IndexError:
+        log.warning('unable to parse name from stack_id: %s', stack_id)
     return ''
 
 
-def list_all_exports(cfn_client: Optional[BaseClient] = None) -> List[Dict]:
+def get_all_exports(cfn_client: BaseClient | None = None) -> list[dict]:
     if cfn_client is None:
         cfn_client = boto3.client('cloudformation')
 
@@ -27,20 +33,19 @@ def list_all_exports(cfn_client: Optional[BaseClient] = None) -> List[Dict]:
     return exports
 
 
-def get_imports(export_name: str, cfn_client: Optional[BaseClient] = None) -> List[str]:
+def get_all_imports(export_name: str, cfn_client: BaseClient | None = None) -> list[str]:
     if cfn_client is None:
         cfn_client = boto3.client('cloudformation')
 
     imports = []
     try:
         resp = cfn_client.list_imports(ExportName=export_name)
-        print(resp)
         imports.extend(resp['Imports'])
         while token := resp.get('NextToken'):
             resp = cfn_client.list_imports(ExportName=export_name, NextToken=token)
             imports.extend(resp['Imports'])
     except ClientError as err:
-        # todo: add logging here
-        if not 'is not imported by any stack' in repr(err):
-            raise err
+        if 'is not imported by any stack' not in repr(err):
+            raise
+        log.debug('')
     return imports

src/cycl/utils/log_config.py

@@ -0,0 +1,6 @@
+import logging
+
+
+def configure_logging(level: int) -> None:
+    log_format = '%(asctime)s [%(levelname)s] %(filename)s:%(lineno)d: %(message)s'
+    logging.basicConfig(level=level, format=log_format, datefmt='%Y-%m-%d %H:%M:%S')

tests/cycl/cli_test.py

@@ -0,0 +1,88 @@
+import logging
+import sys
+from unittest.mock import patch
+
+import networkx as nx
+import pytest
+
+import cycl.cli as cli_module
+from cycl.cli import app
+
+
+@pytest.fixture(autouse=True)
+def mock_build_build_dependency_graph():
+    with patch.object(cli_module, 'build_dependency_graph') as mock:
+        mock.return_value = nx.MultiDiGraph()
+        yield mock
+
+
+@pytest.fixture(autouse=True)
+def mock_configure_logging():
+    with patch.object(cli_module, 'configure_logging') as mock:
+        yield mock
+
+
+def test_app_check_acyclic():
+    sys.argv = ['cycl', 'check']
+    with pytest.raises(SystemExit) as err:
+        app()
+
+    assert err.value.code == 0
+
+
+def test_app_check_cyclic(capsys, mock_build_build_dependency_graph):
+    graph = nx.MultiDiGraph()
+    graph.add_edges_from(
+        [
+            (1, 2),
+            (2, 1),
+        ]
+    )
+    mock_build_build_dependency_graph.return_value = graph
+    sys.argv = ['cycl', 'check']
+
+    with pytest.raises(SystemExit) as err:
+        app()
+
+    assert err.value.code == 1
+    console_output = capsys.readouterr().out
+    assert 'Cycle found between nodes: [1, 2]' in console_output
+
+
+def test_app_check_cyclic_exit_zero(capsys, mock_build_build_dependency_graph):
+    graph = nx.MultiDiGraph()
+    graph.add_edges_from(
+        [
+            (1, 2),
+            (2, 1),
+        ]
+    )
+    mock_build_build_dependency_graph.return_value = graph
+    sys.argv = ['cycl', 'check', '--exit-zero']
+
+    with pytest.raises(SystemExit) as err:
+        app()
+
+    assert err.value.code == 0
+    console_output = capsys.readouterr().out
+    assert 'Cycle found between nodes: [1, 2]' in console_output
+
+
+@pytest.mark.parametrize(
+    ('arg_value', 'log_level'),
+    [
+        ('CRITICAL', logging.CRITICAL),
+        ('DEBUG', logging.DEBUG),
+        ('ERROR', logging.ERROR),
+        ('INFO', logging.INFO),
+        ('WARNING', logging.WARNING),
+    ],
+)
+def test_app_check_acyclic_log_level(mock_configure_logging, arg_value, log_level):
+    sys.argv = ['cycl', 'check', '--log-level', arg_value]
+
+    with pytest.raises(SystemExit) as err:
+        app()
+
+    assert err.value.code == 0
+    mock_configure_logging.assert_called_with(log_level)