feat(shareable-events): Add filtered list

event-schema/aws_lambda_publish_shared_event/__main__.py

@@ -6,7 +6,7 @@
 import boto3
 from pick import pick
 
-from aws_lambda_publish_shared_event.util import build_test_event, list_of_test_events
+from aws_lambda_publish_shared_event.util import build_test_event, handle_list_arguments, list_of_test_events
 
 registry_name = "lambda-testevent-schemas"
 
@@ -31,6 +31,8 @@ def get_lambda_name(session: boto3.session.Session) -> str:
     """Prompt to select which lambda event to add a shareable test event."""
     lambda_client = session.client("lambda")
     response = lambda_client.list_functions()
+    # sourcery skip: use-named-expression
+    # as we support python 3.7
     functions = [f["FunctionName"] for f in response["Functions"]]
     if functions:
         return pick(functions, f"Select Lambda function ({session.region_name}):")[0]
@@ -113,21 +115,12 @@ def generate_new_schema_content(event_name: str, event: Dict) -> str:
 
 def main():
     args = parse_args(sys.argv[1:])
-    list_of_events = list_of_test_events()
-
-    if args.list:
-        print("List of supported event sources:")
-        print(*list_of_events, sep="\n")
-        return
-    if args.filtered_list:
-        filtered_list = list(filter(lambda x: x.startswith(args.filtered_list), list_of_events))
-        print("Filtered list of supported event sources:")
-        print(*filtered_list, sep="\n")
+    if handle_list_arguments(args):
         return
 
     session = get_session(args.region)
     lambda_name = args.lambda_name or get_lambda_name(session)
-    event_path = args.event or get_test_event(list_of_events)
+    event_path = args.event or get_test_event(list_of_test_events())
     schemas_client = session.client("schemas")
     create_registry(schemas_client)
     update_schema(schemas_client, lambda_name, event_path, args.event_name)

event-schema/aws_lambda_publish_shared_event/generate_test_event/__main__.py

@@ -3,27 +3,22 @@
 import sys
 from typing import List
 
-from aws_lambda_publish_shared_event.util import build_test_event, list_of_test_events
+from aws_lambda_publish_shared_event.util import build_test_event, handle_list_arguments
 
 
 def parse_args(args: List[str]) -> argparse.Namespace:
     """Parse arguments from the cli"""
-    parser = argparse.ArgumentParser(
-        prog="generate-test-event", description="Generated a local test event to the system out"
-    )
+    parser = argparse.ArgumentParser(prog="generate-test-event", description="Generated a local test event")
     parser.add_argument("-l", "--list", dest="list", help="List of supported event sources", action="store_true")
+    parser.add_argument("--filtered-list", dest="filtered_list", help="Filtered list")
     parser.add_argument("event", help="Event source", nargs="?", default=None)
     return parser.parse_args(args)
 
 
 def main():
     args = parse_args(sys.argv[1:])
-
-    if args.list:
-        print("List of supported event sources:")
-        print(*list_of_test_events(), sep="\n")
-        return
-
+    if handle_list_arguments(args):
+        return True
     event = args.event
     if event is None:
         raise SystemExit("No event source specified")

event-schema/aws_lambda_publish_shared_event/util.py

@@ -1,3 +1,4 @@
+import argparse
 import json
 import os
 from fnmatch import fnmatch
@@ -37,3 +38,17 @@ def build_test_event(event_name: Optional[str], event_path: str) -> Tuple[str, D
     event_name = event_name or path.name.replace(".json", "")
     event = json.loads(path.read_text())
     return event_name, event
+
+
+def handle_list_arguments(args: argparse.Namespace) -> bool:
+    """Handle any of the list arguments (-l, --filter-list) and return True if processed"""
+    if args.list:
+        print("List of supported event sources:")
+        print(*list_of_test_events(), sep="\n")
+        return True
+    if args.filtered_list:
+        filtered_list = list(filter(lambda x: x.startswith(args.filtered_list), list_of_test_events()))
+        print("Filtered list of supported event sources:")
+        print(*filtered_list, sep="\n")
+        return True
+    return False

event-schema/tests/test_generate_test_event.py

@@ -19,20 +19,6 @@ def test_parse_args_no_args():
     assert str(e.value) == "No event source specified"
 
 
-@pytest.mark.parametrize("argument", ["-h", "--help"])
-def test_parse_args_help(capsys, argument: str):
-    # GIVEN
-    sys.argv = [SCRIPT_NAME, argument]
-
-    # WHEN
-    with pytest.raises(SystemExit):
-        __main__.main()
-
-    # THEN
-    captured = capsys.readouterr()
-    assert captured.out.startswith("usage:")
-
-
 @pytest.mark.parametrize("argument", ["-l", "--list"])
 def test_parse_args_list(capsys, argument: str):
     # GIVEN

event-schema/tests/test_main.py

@@ -25,11 +25,29 @@ def test_parse_args_help(capsys):
     assert captured.out.startswith("usage:")
 
 
-def test_parse_args_list(capsys):
-    sys.argv = [SCRIPT_NAME, "--list"]
+@pytest.mark.parametrize("argument", ["-h", "--help"])
+def test_parse_args_help(capsys, argument: str):
+    # GIVEN
+    sys.argv = [SCRIPT_NAME, argument]
 
+    # WHEN
+    with pytest.raises(SystemExit):
+        __main__.main()
+
+    # THEN
+    captured = capsys.readouterr()
+    assert captured.out.startswith("usage:")
+
+
+@pytest.mark.parametrize("argument", ["-l", "--list"])
+def test_parse_args_list(capsys, argument: str):
+    # GIVEN
+    sys.argv = [SCRIPT_NAME, argument]
+
+    # WHEN
     __main__.main()
 
+    # THEN
     captured = capsys.readouterr()
     assert captured.out.startswith("List of supported event sources:")