finish up adding topo

README.md

@@ -7,6 +7,8 @@
 
 cycl is a CLI and Python SDK to help identify cross-stack import/export circular dependencies, for a given AWS account and region.
 
+The successor to [circular-dependency-detector](https://github.com/tcm5343/circular-dependency-detector), which was built at the University of Texas at Austin.
+
 ## Getting started
 
 Install `cycl` by running `pip install cycl`.
@@ -17,10 +19,6 @@ Install `cycl` by running `pip install cycl`.
 - `cycl check --log-level` - set the logging level (default: WARNING)
 - `cycl check --cdk-out /path/to/cdk.out` - path to cdk.out, where stacks are CDK synthesized to CFN templates
 
-### SDK
-
-...
-
 ## How to use cycl?
 
 There are two main use cases for `cycl`.
@@ -34,4 +32,4 @@ Over the lifetime of a project, circular references are bound to be introduced.
 
 ## Contributing
 
-`cycl` is being actively developed, instructions to come as it becomes more stable.
+`cycl` is being actively developed, instructions will come as it becomes more stable.

TODO.md

@@ -1,10 +1,8 @@
 # TODO.md
 
-- output the topological generations of the graph
-  - `cycl check --generations`
 - ignoring known cycles, [this](https://cs.stackexchange.com/questions/90481/how-to-remove-cycles-from-a-directed-graph) may help
-  - `cycl check --ignore-cycle-contains`
-  - `cycl check --ignore-cycle`
+  - `cycl check --ignore`
+  - `cycl topo --ignore`
 - reducing the stacks, for example, a tag on a stack representing the github repo name
   - `cycl check --reduce-dependencies-on`
   - `cycl check --reduce-generations-on`
@@ -18,3 +16,4 @@
 - [ ] Test with stages that the correct manifest is analyzed
 - [ ] What if cdk out synth is for multiple accounts? We may need to determine what account we have credentials for and only analyze those templates
 - [ ] automatic documentation generation
+- [ ] configure tox

pyproject.toml

@@ -5,22 +5,22 @@ build-backend = 'setuptools.build_meta'
 [project]
 
 name = 'cycl'
-version='0.1.1'
-requires-python = ">=3.8"
+version='0.1.0'
+requires-python = '>=3.8'
 description = 'CLI and Python SDK to help identify cross-stack import/export circular dependencies, for a given AWS account and region.'
 readme = 'README.md'
-keywords = ['aws', 'cdk', 'cycle', 'circular', 'dependency', 'infrastructure']
+keywords = ['aws', 'cdk', 'cycle', 'circular', 'dependency', 'infrastructure', 'boto3', 'cloud']
 classifiers = [
-    'License :: OSI Approved :: MIT License',
     'Intended Audience :: Developers',
+    'License :: OSI Approved :: MIT License',
     'Natural Language :: English',
+    'Programming Language :: Python :: 3.10',
+    'Programming Language :: Python :: 3.11',
+    'Programming Language :: Python :: 3.12',
+    'Programming Language :: Python :: 3.13',
+    'Programming Language :: Python :: 3.8',
+    'Programming Language :: Python :: 3.9',
     'Programming Language :: Python :: 3',
-    "Programming Language :: Python :: 3.8",
-    "Programming Language :: Python :: 3.9",
-    "Programming Language :: Python :: 3.10",
-    "Programming Language :: Python :: 3.11",
-    "Programming Language :: Python :: 3.12",
-    "Programming Language :: Python :: 3.13",
     'Programming Language :: Python',
     'Topic :: Software Development',
 ]
@@ -81,5 +81,5 @@ omit = [
 ]
 
 [tool.pytest.ini_options]
-log_cli_level = "INFO"
+log_cli_level = 'INFO'
 addopts = ['--import-mode=importlib']

src/cycl/cli.py

@@ -1,4 +1,5 @@
 import argparse
+import json
 import logging
 import pathlib
 import sys
@@ -13,38 +14,61 @@
 
 
 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='WARNING',
-        help='set the logging level (default: WARNING)',
-    )
-    parent_parser.add_argument(
-        '--cdk-out',
-        type=pathlib.Path,
-        help='path to cdk.out, where stacks are CDK synthesized to CFN templates',
-    )
-
-    subparsers.add_parser('check', parents=[parent_parser], help='Check for cycles in AWS stack imports/exports')
+    parser = argparse.ArgumentParser(prog='cycl', description='Check for cross-stack import/export circular dependencies.')
+    sp = parser.add_subparsers(dest='cmd', required=True)
+
+    check_p = sp.add_parser('check', help='Check for cycles between stacks in AWS stack imports/exports')
+    check_p.add_argument('--exit-zero', action='store_true', help='Exit 0 regardless of result of cycle check')
+
+    topo_p = sp.add_parser('topo', help='Find topological generations if dependencies are acyclic')
+
+    # global options
+    for p in [check_p, topo_p]:
+        p.add_argument(
+            '--log-level',
+            choices=['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
+            default='WARNING',
+            help='Set the logging level (default: WARNING)',
+        )
+        # p.add_argument(
+        #     "--platform",
+        #     choices=["AWS"],
+        #     default='AWS',
+        #     help="Cloud platform to interact with",
+        # )
+        p.add_argument(
+            '--cdk-out',
+            type=pathlib.Path,
+            help='Path to cdk.out, where stacks are CDK synthesized to CFN templates',
+        )
+        # p.add_argument(
+        #     '-q', "--quiet",
+        #     type=pathlib.Path,
+        #     default=argparse.SUPPRESS,
+        #     help="Suppress output",
+        # )
+
+    if len(sys.argv) == 1:
+        parser.print_help()
+        sys.exit(0)
 
     args = parser.parse_args()
     configure_log(getattr(logging, args.log_level))
-    log.info(args)
-
-    if args.action == 'check':
-        cycle_found = False
-        graph = build_dependency_graph(cdk_out_path=args.cdk_out)
-        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:
+
+    dep_graph = build_dependency_graph(cdk_out_path=args.cdk_out)
+    cycles = list(nx.simple_cycles(dep_graph))
+    for cycle in cycles:
+        print(f'cycle found between nodes: {cycle}')
+
+    if args.cmd == 'check':
+        if cycles and not args.exit_zero:
+            sys.exit(1)
+    elif args.cmd == 'topo':
+        if cycles:
+            print('\nerror: graph is cyclic, topological generations can only be computed on an acyclic graph')
             sys.exit(1)
+        generations = [sorted(generation) for generation in nx.topological_generations(dep_graph)]
+        print(json.dumps(generations, indent=2))
     sys.exit(0)
 
 

src/cycl/cycl.py

@@ -3,22 +3,30 @@
 from logging import getLogger
 from pathlib import Path
 
+import boto3
 import networkx as nx
+from botocore.config import Config
 
 from cycl.utils.cdk import get_cdk_out_imports
 from cycl.utils.cfn import get_all_exports, get_all_imports, parse_name_from_id
 
 log = getLogger(__name__)
 
 
-def build_dependency_graph(cdk_out_path: Path | None = None) -> nx.MultiDiGraph:
-    dep_graph = nx.MultiDiGraph()
-
+def get_dependency_graph_data(cdk_out_path: Path | None = None) -> dict:
     cdk_out_imports = {}
     if cdk_out_path:
         cdk_out_imports = get_cdk_out_imports(Path(cdk_out_path))
 
-    exports = get_all_exports()
+    boto_config = Config(
+        retries={
+            'max_attempts': 10,
+            'mode': 'adaptive',
+        }
+    )
+    cfn_client = boto3.client('cloudformation', config=boto_config)
+
+    exports = get_all_exports(cfn_client=cfn_client)
     for export_name in cdk_out_imports:
         if export_name not in exports:
             log.warning(
@@ -29,14 +37,23 @@ def build_dependency_graph(cdk_out_path: Path | None = None) -> nx.MultiDiGraph:
 
     for export in exports.values():
         export['ExportingStackName'] = parse_name_from_id(export['ExportingStackId'])
-        export['ImportingStackNames'] = get_all_imports(export_name=export['Name'])
+        export['ImportingStackNames'] = get_all_imports(export_name=export['Name'], cfn_client=cfn_client)
         export.setdefault('ImportingStackNames', []).extend(cdk_out_imports.get(export['Name'], []))
+        if len(export['ImportingStackNames']) == 0:
+            log.info('Export found with no import: %s from %s', export['Name'], export['ExportingStackName'])
+    return exports
+
+
+def build_dependency_graph(cdk_out_path: Path | None = None) -> nx.MultiDiGraph:
+    dep_graph = nx.MultiDiGraph()
+    graph_data = get_dependency_graph_data(cdk_out_path)
+
+    for export in graph_data.values():
         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

tests/cli_test.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import sys
 from unittest.mock import patch
@@ -22,14 +23,15 @@ def mock_configure_log():
         yield mock
 
 
-def test_app_no_action(capsys):
+def test_app_no_cmd_displays_help(capsys):
     sys.argv = ['cycl']
     with pytest.raises(SystemExit) as err:
         app()
 
-    assert err.value.code == 2
-    console_output = capsys.readouterr().err
-    assert 'cycl: error: the following arguments are required: action' in console_output
+    assert err.value.code == 0
+    console_output = capsys.readouterr().out
+    assert 'usage: cycl [-h] {check,topo}' in console_output
+    assert 'Check for cross-stack import/export circular dependencies.' in console_output
 
 
 def test_app_unsupported_action(capsys):
@@ -39,7 +41,7 @@ def test_app_unsupported_action(capsys):
 
     assert err.value.code == 2
     console_output = capsys.readouterr().err
-    assert "cycl: error: argument action: invalid choice: 'something'" in console_output
+    assert "cycl: error: argument cmd: invalid choice: 'something'" in console_output
 
 
 def test_app_check_acyclic():
@@ -98,11 +100,65 @@ def test_app_check_cyclic_exit_zero(capsys, mock_build_build_dependency_graph):
         ('WARNING', logging.WARNING),
     ],
 )
-def test_app_check_acyclic_log_level(mock_configure_log, arg_value, log_level):
-    sys.argv = ['cycl', 'check', '--log-level', arg_value]
+@pytest.mark.parametrize('cmd', ['check', 'topo'])
+def test_app_acyclic_log_level(mock_configure_log, arg_value, log_level, cmd):
+    sys.argv = ['cycl', cmd, '--log-level', arg_value]
 
     with pytest.raises(SystemExit) as err:
         app()
 
     assert err.value.code == 0
     mock_configure_log.assert_called_with(log_level)
+
+
+def test_app_topo_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', 'topo']
+
+    with pytest.raises(SystemExit) as err:
+        app()
+
+    assert err.value.code == 1
+    console_output = capsys.readouterr().out
+    assert 'error: graph is cyclic, topological generations can only be computed on an acyclic graph' in console_output
+
+
+def test_app_topo_acyclic(capsys, mock_build_build_dependency_graph):
+    graph = nx.MultiDiGraph()
+    graph.add_edges_from(
+        [
+            (2, 1),
+            (3, 1),
+        ]
+    )
+    mock_build_build_dependency_graph.return_value = graph
+    expected_output = json.dumps([[2, 3], [1]], indent=2)
+    sys.argv = ['cycl', 'topo']
+
+    with pytest.raises(SystemExit) as err:
+        app()
+
+    assert err.value.code == 0
+    console_output = capsys.readouterr().out
+    assert expected_output in console_output
+
+
+def test_app_topo_empty(capsys, mock_build_build_dependency_graph):
+    graph = nx.MultiDiGraph()
+    mock_build_build_dependency_graph.return_value = graph
+    expected_output = json.dumps([], indent=2)
+    sys.argv = ['cycl', 'topo']
+
+    with pytest.raises(SystemExit) as err:
+        app()
+
+    assert err.value.code == 0
+    console_output = capsys.readouterr().out
+    assert expected_output in console_output

tests/cycl_test.py

@@ -8,6 +8,18 @@
 from cycl.cycl import build_dependency_graph
 
 
+@pytest.fixture(autouse=True)
+def mock_boto3():
+    with patch.object(cycl_module, 'boto3') as mock:
+        yield mock
+
+
+@pytest.fixture(autouse=True)
+def mock_config():
+    with patch.object(cycl_module, 'Config') as mock:
+        yield mock
+
+
 @pytest.fixture(autouse=True)
 def mock_parse_name_from_id():
     with patch.object(cycl_module, 'parse_name_from_id') as mock:
@@ -123,7 +135,7 @@ def test_build_dependency_graph_returns_graph_with_multiple_exports(mock_get_all
         },
     }
 
-    def mock_get_all_imports_side_effect_func(export_name):
+    def mock_get_all_imports_side_effect_func(export_name, *_args, **_kwargs):
         if export_name == 'some-name-1':
             return [
                 'some-importing-stack-name-1',
@@ -290,3 +302,15 @@ def test_build_dependency_graph_returns_graph_with_cdk_out_path_and_no_existing_
 
     assert nx.is_directed_acyclic_graph(actual_graph)
     assert next(nx.simple_cycles(actual_graph), []) == []
+
+
+def test_config_defined_as_expected(mock_config, mock_boto3):
+    build_dependency_graph()
+
+    mock_config.assert_called_once_with(
+        retries={
+            'max_attempts': 10,
+            'mode': 'adaptive',
+        },
+    )
+    mock_boto3.client.assert_called_once_with('cloudformation', config=mock_config.return_value)