Merge pull request #347 from opencivicdata/add-clean-failsafe

Add failsafe to clean command

pupa/cli/commands/clean.py

@@ -88,11 +88,21 @@ def handle(self, args, other):
             )
             self.report_stale_objects()
         else:
-            if not args.noinput:
+            num_stale_objects = len(list(self.get_stale_objects(args.window)))
+
+            if args.noinput:
+                # Fail-safe to avoid deleting a large amount of objects
+                # without explicit confimation
+                if num_stale_objects > 10:
+                    print(
+                        "This command would delete more than 10 objects."
+                        "If you're sure, re-run without --noinput to provide confirmation."
+                    )
+                    sys.exit(1)
+            else:
                 print(
                     f"This will permanently delete"
-                    f" {len(list(self.get_stale_objects(args.window)))}"
-                    " objects from your database"
+                    f" {num_stale_objects} objects from your database"
                     f" that have not been scraped within the last {args.window}"
                     " days. Are you sure? (Y/N)"
                 )

pupa/tests/clean/test_clean.py

@@ -97,3 +97,25 @@ def test_clean_command(subparsers):
         for obj in expected_not_stale_objects:
             was_not_deleted = type(obj).objects.filter(id=obj.id).exists()
             assert was_not_deleted
+
+
+@pytest.mark.django_db
+def test_clean_command_failsafe(subparsers):
+    _ = create_jurisdiction()
+    o = Organization.objects.create(name="WWE", jurisdiction_id="jid")
+
+    stale_people = [
+        Person.objects.create(name="George Washington", family_name="Washington")
+        for i in range(20)
+    ]
+    stale_memberships = [ # noqa
+        p.memberships.create(organization=o) for p in stale_people
+    ]
+
+    a_week_from_now = datetime.now(tz=timezone.utc) + timedelta(days=7)
+    with freeze_time(a_week_from_now):
+        with pytest.raises(SystemExit):
+            # Should trigger failsafe exist when deleting more than 10 objects
+            Command(subparsers).handle(
+                argparse.Namespace(noinput=True, report=False, window=7), []
+            )