fixing tests 1 by 1

stepview/__init__.py

@@ -1,6 +1,17 @@
 import logging
 import os
 from pathlib import Path
+import warnings
+
+
+def set_logger_3rd_party_lib(logging_level=logging.CRITICAL):
+    if logging_level == logging.CRITICAL:
+        warnings.filterwarnings(action="ignore", message="unclosed", category=ResourceWarning)
+    for name in ['boto3', 'botocore', 'urllib3']:
+        logging.getLogger(name).setLevel(logging_level)
+
+
+set_logger_3rd_party_lib()
 
 root = logging.getLogger()
 log_level = os.environ.get("LOG_LEVEL", "INFO")

stepview/data.py

@@ -1,16 +1,19 @@
 import concurrent.futures
+import logging
+
 import boto3
 import botocore.client
 import pendulum
+from rich.progress import track, Progress, TextColumn, BarColumn
 
 from rich.table import Table
 from dataclasses import dataclass
-from stepview import logger
+from stepview import logger, set_logger_3rd_party_lib
 
 
 @dataclass
-class States:
-    """States that we pass to the TUI table."""
+class State:
+    """State that we pass to the TUI table."""
 
     total_executions: int
     succeeded: str
@@ -21,6 +24,30 @@ class States:
     timed_out: str
     throttled: str
 
+@dataclass
+class Row:
+    state_machine: str
+    profile_name: str
+    account: str
+    region:  str
+    state: State
+
+    def get_values(self):
+        return (
+            self.state_machine,
+            self.profile_name,
+            self.account,
+            self.region,
+            f"{self.state.total_executions:,.0f}",
+            f"{self.state.succeeded_perc:,.2f}",
+            f"{self.state.running:,.0f}",
+            f"{self.state.failed:,.0f}",
+            f"{self.state.aborted:,.0f}",
+            f"{self.state.timed_out:,.0f}",
+            f"{self.state.throttled}"
+
+        )
+
 
 @dataclass
 class Periods:
@@ -75,7 +102,10 @@ class Time:
 
 def main(aws_profiles: list, period: str):
 
-    profile_generator = run_all_profiles(aws_profiles=aws_profiles, period=period)
+    progress_viz = (TextColumn("[progress.description]{task.description}"), BarColumn())
+    with Progress(*progress_viz) as progress:
+        progress.add_task("[green]Getting Data...", start=False)
+        profile_generator = run_all_profiles(aws_profiles=aws_profiles, period=period)
 
     table = Table()
     table.add_column("StateMachine", justify="left", overflow="fold")
@@ -90,12 +120,15 @@ def main(aws_profiles: list, period: str):
     table.add_column("TimedOut")
     table.add_column("Throttled")
 
+    all_rows = []
     for profile in profile_generator:
         for row in profile:
             if row:
-                table.add_row(*row)
+                table.add_row(*row.get_values())
+            all_rows.append(row)
 
-    return table
+    # return table for viz, return all_rows for tests
+    return table, all_rows
 
 
 def run_all_profiles(aws_profiles: list, period: str):
@@ -112,7 +145,7 @@ def run_for_state_machine(
     state_machine: object, cloudwatch_resource: object, profile_name: str, period: str
 ):
     state_machine_arn = state_machine.get("stateMachineArn")
-    states = get_data_from_cloudwatch(
+    state = get_data_from_cloudwatch(
         cloudwatch_resource=cloudwatch_resource,
         state_machine_arn=state_machine_arn,
         period=period,
@@ -124,24 +157,20 @@ def run_for_state_machine(
     state_machine_url = get_statemachine_url(
         state_machine_arn=state_machine_arn, region=region
     )
-    state_machine_name_url = f"[link={state_machine_url}]{state_machine_name}[/link]"
-    return_object = (
-        state_machine_name_url,
-        profile_name,
-        account,
-        region,
-        f"{states.total_executions:,.0f}",
-        f"{states.succeeded_perc:,.2f}",
-        f"{states.running:,.0f}",
-        f"{states.failed:,.0f}",
-        f"{states.aborted:,.0f}",
-        f"{states.timed_out:,.0f}",
-        f"{states.throttled:,.0f}",
+    state_machine_name_with_url = f"[link={state_machine_url}]{state_machine_name}[/link]"
+    row = Row(
+        state_machine=state_machine_name_with_url,
+        profile_name=profile_name,
+        account=account,
+        region=region,
+        state=state
+
     )
-    return return_object
+    return row
 
 
 def run_for_profile(profile_name: str, period: str) -> Table:
+
     sfn_client = boto3.Session(
         profile_name=profile_name
     ).client(
@@ -179,7 +208,7 @@ def call_metric_endpoint(
     """
     https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/cloudwatch.html#metric
     """
-    metric = cloudwatch_resource.Metric("AWS/States", metric_name).get_statistics(
+    metric = cloudwatch_resource.Metric("AWS/State", metric_name).get_statistics(
         Dimensions=[
             {
                 "Name": "StateMachineArn",
@@ -199,7 +228,7 @@ def call_metric_endpoint(
 
 def get_data_from_cloudwatch(
     cloudwatch_resource: object, state_machine_arn: str, period: str
-) -> States:
+) -> State:
     """
     check the docs for more info
     https://docs.aws.amazon.com/step-functions/latest/dg/procedure-cw-metrics.html
@@ -243,7 +272,7 @@ def _call_metric_endpoint(metric_name):
 
     succeeded_perc = (succeeded / started) * 100 if started > 0 else 0
 
-    return States(
+    return State(
         total_executions=started,
         succeeded=succeeded,
         succeeded_perc=succeeded_perc,

stepview/entrypoint.py

@@ -1,7 +1,9 @@
+import logging
 from typing import List
 
 import typer
 
+from stepview import set_logger_3rd_party_lib
 from stepview.data import PERIODS_MAPPING, Time
 from stepview.tui import StepViewTUI
 
@@ -37,7 +39,13 @@ def stepview(
         help="specify the time period for which you wish to look back. "
         f"""You can choose from the values: {', '.join(PERIODS_MAPPING.keys())}""",
     ),
+    verbose: bool = typer.Option(
+        False, "--verbose",
+        help="use --verbose to set verbose logging."),
+
 ):
+    if verbose:
+        set_logger_3rd_party_lib(logging_level=logging.DEBUG)
     StepViewTUI.run(
         title=f"STEPVIEW (period: {period})", aws_profiles=profile, period=period
     )

stepview/tui.py

@@ -31,7 +31,7 @@ async def on_mount(self, event: events.Mount) -> None:
         await self.view.dock(body, edge="right")
 
         async def get_stepfunction_data():
-            table = main(aws_profiles=self.aws_profiles, period=self.period)
+            table, _ = main(aws_profiles=self.aws_profiles, period=self.period)
             await body.update(table)
 
         await self.call_later(get_stepfunction_data)

stepview_tests/test_stepview.py

@@ -76,7 +76,7 @@ def create_statemachine(name, profile):
     return client, role, state_machine
 
 
-def create_metric(metric_name, profile, state_machine):
+def create_metric(metric_name, profile, state_machine, timestamp=NOW.subtract(minutes=1)):
     """
     Add a metric to cloudwatch
     check how to add MetricData from the documentation
@@ -92,7 +92,7 @@ def create_metric(metric_name, profile, state_machine):
 
     client = boto3.Session(profile_name=profile).client("cloudwatch")
     client.put_metric_data(
-        Namespace="AWS/States",
+        Namespace="AWS/State",
         MetricData=[
             {
                 "MetricName": metric_name,
@@ -110,7 +110,7 @@ def create_metric(metric_name, profile, state_machine):
                 },
                 # we substract 1 minute so that we are sure we are
                 # before the init of NOW in data module.
-                "Timestamp": NOW.subtract(minutes=1),
+                "Timestamp": timestamp,
                 "Values": [1],
                 "Value": 1,
             }
@@ -138,61 +138,38 @@ def test_get_stepfunctions_status_happy_flow(self):
 
         self.assertIsNone(self.exception_)
 
+    @mock_cloudwatch
     @mock_stepfunctions
-    @patch("stepview.data.list_executions_for_state_machine")
-    def test_get_stepfunctions_with_next_token(self, m_list_executions):
-
-        sfn_client, role, statemachine = create_statemachine("sm1", "profile1")
-
-        m_list_executions.side_effect = [
-            {
-                # we add a token so that we call the function
-                # list_executions_for_state_machine two times.
-                "nextToken": "some-token",
-                **list_executions(["RUNNING", "FAILED", "SUCCEEDED", "SUCCEEDED"]),
-            },
-            list_executions(["RUNNING", "FAILED", "SUCCEEDED", "SUCCEEDED"]),
-        ]
-        self.exception_ = None
-        try:
-            states = stepview.data.get_all_states_of_executions(
-                sfn_client=sfn_client,
-                state_machine_arn=statemachine.get("stateMachineArn"),
-                period="day",
-            )
-
-        except Exception as e:
-            self.exception_ = e
-
-        self.assertIsNone(self.exception_)
-        self.assertEqual(m_list_executions.call_count, 2)
-        self.assertEqual(states.failed, 2)
-        self.assertEqual(states.running, 2)
-        self.assertEqual(states.succeeded, 4)
-        self.assertEqual(states.succeeded_perc, 50.0)
-        self.assertEqual(states.total_executions, 8)
-
-    @mock_stepfunctions
-    @patch("stepview.data.list_executions_for_state_machine")
-    def test_stepview_on_time_period_minute(self, m_list_executions):
-        sfn_client, role, statemachine = create_statemachine("sm1", "profile1")
+    def test_stepview_on_time_period_minute(self):
+        sfn_client, role, state_machine = create_statemachine("sm1", "profile1")
 
-        last_minute = datetime.datetime.fromisoformat(
+        time_started = datetime.datetime.fromisoformat(
             pendulum.now().subtract(minutes=1, seconds=2).to_iso8601_string()
         )
-        m_list_executions.side_effect = [
-            {
-                "nextToken": "some-token",
-                **list_executions(["SUCCEEDED"]),
-            },
-            list_executions(["FAILED"], start_date=last_minute),
-        ]
+        time_succeeded = datetime.datetime.fromisoformat(
+            pendulum.now().subtract(minutes=1, seconds=1).to_iso8601_string()
+        )
 
-        states = stepview.data.get_all_states_of_executions(
-            sfn_client=sfn_client,
-            state_machine_arn=statemachine.get("stateMachineArn"),
-            period=stepview.data.MINUTE,
+
+        create_metric(
+            MetricNames.EXECUTIONS_STARTED,
+            profile="profile1",
+            state_machine=state_machine,
+            timestamp=time_started
+        )
+        create_metric(
+            MetricNames.EXECUTIONS_SUCCEEDED,
+            profile="profile1",
+            state_machine=state_machine,
+            timestamp=time_succeeded
         )
+        _, result = stepview.data.main(aws_profiles=["profile1"], period="day")
+
+        # states = stepview.data.get_all_states_of_executions(
+        #     sfn_client=sfn_client,
+        #     state_machine_arn=state_machine.get("stateMachineArn"),
+        #     period=stepview.data.MINUTE,
+        # )
 
         self.assertEqual(states.succeeded, 1)
         self.assertEqual(states.succeeded_perc, 100.0)
@@ -391,3 +368,6 @@ def test_cli(self, m_textual_run):
             stepview.entrypoint.app, ["--profile", "profile1 profile2 profile3"]
         )
         self.assertEqual(result.exit_code, 0)
+
+    def test_verbose(self):
+        pass