Bugfix/security-data flattened checkpoint on v2 events (#399)

* fix v2 checkpoint bug when format is json

* fix v2 checkpoint bug when format is json

* fix some new flake8 errors

* style

Author: timabrmsn

CHANGELOG.md

@@ -8,6 +8,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 The intended audience of this file is for py42 consumers -- as such, changes that don't affect
 how a consumer would use the library (e.g. adding unit tests, updating documentation, etc) are not captured here.
 
+## 1.16.4 - 2023-01-27
+
+### Fixed
+
+- Bug in `security-data search|send-to` where using `--format json` and a checkpoint raised an error when configured for V2 file events.
+
 ## 1.16.3 - 2022-12-08
 
 ### Fixed

setup.cfg

@@ -29,5 +29,7 @@ ignore =
     W503
     # exception chaining
     B904
+    # manual quoting
+    B907
 # up to 88 allowed by bugbear B950
 max-line-length = 80

src/code42cli/__version__.py

@@ -1 +1 @@
-__version__ = "1.16.3"
+__version__ = "1.16.4"

src/code42cli/cmds/securitydata.py

@@ -459,14 +459,17 @@ def search(
                     "riskSeverity",
                 ]
 
+    flatten = format in (OutputFormat.TABLE, OutputFormat.CSV)
+
     if use_checkpoint:
         cursor = _get_file_event_cursor_store(state.profile.name)
         checkpoint = _handle_timestamp_checkpoint(cursor.get(use_checkpoint), state)
 
         if state.profile.use_v2_file_events == "True":
 
             def checkpoint_func(event):
-                cursor.replace(use_checkpoint, event["event.id"])
+                event_id = event["event.id"] if flatten else event["event"]["id"]
+                cursor.replace(use_checkpoint, event_id)
 
         else:
 
@@ -477,7 +480,6 @@ def checkpoint_func(event):
         checkpoint = checkpoint_func = None
 
     query = _construct_query(state, begin, end, saved_search, advanced_query, or_query)
-    flatten = format in (OutputFormat.TABLE, OutputFormat.CSV)
     dfs = _get_all_file_events(state, query, checkpoint, flatten)
     formatter = FileEventsOutputFormatter(format, checkpoint_func=checkpoint_func)
     # sending to pager when checkpointing can be inaccurate due to pager buffering, so disallow pager

tests/test_config.py

@@ -151,7 +151,7 @@ def test_create_profile_when_given_default_name_does_not_create(
         self, config_parser_for_create
     ):
         accessor = ConfigAccessor(config_parser_for_create)
-        with pytest.raises(Exception):
+        with pytest.raises(NoConfigProfileError):
             accessor.create_profile(
                 ConfigAccessor.DEFAULT_VALUE, "foo", "bar", False, False, False
             )
@@ -201,7 +201,7 @@ def test_update_profile_when_no_profile_exists_raises_exception(
         self, config_parser_for_multiple_profiles
     ):
         accessor = ConfigAccessor(config_parser_for_multiple_profiles)
-        with pytest.raises(Exception):
+        with pytest.raises(NoConfigProfileError):
             accessor.update_profile("Non-existent Profile")
 
     def test_update_profile_updates_profile(self, config_parser_for_multiple_profiles):