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
17 changes: 16 additions & 1 deletion stacklet/client/sinistral/commands/dump.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,20 +2,35 @@
# SPDX-License-Identifier: Apache-2.0
import sys

from pathlib import Path

import click

from c7n_left.cli import dump as left_dump
from c7n_left.core import get_provider
from c7n_left.entry import initialize_iac


class LeftWrapper(click.core.Command):
def make_parser(self, ctx):
existing = {param.name for param in self.params}
for param in left_dump.params:
self.params.append(param)
# make_parser runs on every invocation; the dump command is a
# module-level singleton, so skip params already appended.
if param.name not in existing:
self.params.append(param)
return super().make_parser(ctx)


@click.command(name="dump", cls=LeftWrapper)
@click.pass_context
def dump(ctx, *args, **kwargs):
"""Dump the IaC resource graph and input variables"""
# c7n-left registers its IaC providers in its own cli group callback,
# which is bypassed when invoking the dump subcommand directly.
initialize_iac()
if not ctx.params.get("directory"):
ctx.params["directory"] = "."
if get_provider(Path(ctx.params["directory"])) is None:
raise click.UsageError(f"No supported IaC files found in {ctx.params['directory']}")
sys.exit(left_dump.invoke(ctx))
46 changes: 46 additions & 0 deletions tests/test_dump.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,46 @@
# Copyright Stacklet, Inc.
# SPDX-License-Identifier: Apache-2.0
import json
import pathlib
import subprocess
import sys


TERRAFORM_DIR = pathlib.Path(__file__).parent.resolve() / "terraform" / "good"


def run_dump(*args, cwd=None):
# Run in a fresh interpreter: c7n-left registers its IaC providers via
# import side effects, so in-process tests could pass only because an
# earlier test happened to register them.
return subprocess.run(
[
sys.executable,
"-c",
"from stacklet.client.sinistral.cli import cli; cli()",
"dump",
*args,
],
check=False,
capture_output=True,
text=True,
cwd=cwd,
)


def test_dump_directory():
result = run_dump("-d", str(TERRAFORM_DIR))
assert result.returncode == 0, result.stderr
assert "aws_sqs_queue" in json.loads(result.stdout)["graph"]


def test_dump_defaults_to_current_directory():
result = run_dump(cwd=TERRAFORM_DIR)
assert result.returncode == 0, result.stderr
assert "aws_sqs_queue" in json.loads(result.stdout)["graph"]


def test_dump_without_iac_files(tmp_path):
result = run_dump("-d", str(tmp_path))
assert result.returncode != 0
assert "No supported IaC files found" in result.stderr + result.stdout
Loading