Merge pull request #1321 from cisagov/nmb/1294-slow-roll

#1294: Send transition domain invitations to particular email addresses

src/registrar/management/commands/send_domain_invitations.py

@@ -37,14 +37,24 @@ class Command(BaseCommand):
             help="Send emails ",
         )
 
+        parser.add_argument("emails", nargs="*", help="Email addresses to send invitations to")
+
     def handle(self, **options):
         """Process the objects in TransitionDomain."""
 
         logger.info("checking domains and preparing emails")
-        # Get all TransitionDomain objects
+
-        self.transition_domains = TransitionDomain.objects.filter(
+        if options["emails"]:
-            email_sent=False,
+            # this option is a list of email addresses
-        ).order_by("username")
+            self.transition_domains = TransitionDomain.objects.filter(
+                username__in=options["emails"],
+                email_sent=False,
+            ).order_by("username")
+        else:
+            # Get all TransitionDomain objects
+            self.transition_domains = TransitionDomain.objects.filter(
+                email_sent=False,
+            ).order_by("username")
         logger.info("Found %d transition domains", len(self.transition_domains))
 
         self.build_emails_to_send_array()

src/registrar/tests/common.py

@@ -58,7 +58,7 @@ def get_handlers():
 
 
 @contextmanager
-def less_console_noise():
+def less_console_noise(output_stream=None):
     """
     Context manager to use in tests to silence console logging.
 
@@ -66,14 +66,19 @@ def less_console_noise():
     (such as errors) which are normal and expected.
 
     It can easily be removed to debug a failing test.
+
+    Arguments:
+        `output_stream`: a stream to redirect every handler to. If it's
+        not provided, use /dev/null.
     """
     restore = {}
     handlers = get_handlers()
-    devnull = open(os.devnull, "w")
+    if output_stream is None:
+        output_stream = open(os.devnull, "w")
 
     # redirect all the streams
     for handler in handlers.values():
-        prior = handler.setStream(devnull)
+        prior = handler.setStream(output_stream)
         restore[handler.name] = prior
     try:
         # run the test
@@ -82,8 +87,9 @@ def less_console_noise():
         # restore the streams
         for handler in handlers.values():
             handler.setStream(restore[handler.name])
-        # close the file we opened
+        if output_stream is None:
-        devnull.close()
+            # we opened output_stream so we have to close it
+            output_stream.close()
 
 
 class MockUserLogin:

src/registrar/tests/test_transition_domain_migrations.py

@@ -1,4 +1,7 @@
 import datetime
+
+from io import StringIO
+
 from django.test import TestCase
 
 from registrar.models import (
@@ -13,6 +16,8 @@ from registrar.models import (
 from django.core.management import call_command
 from unittest.mock import patch
 
+from .common import less_console_noise
+
 
 class TestMigrations(TestCase):
     def setUp(self):
@@ -369,3 +374,42 @@ class TestMigrations(TestCase):
             expected_missing_domain_informations,
             expected_missing_domain_invitations,
         )
+
+    def test_send_domain_invitations_email(self):
+        """Can send only a single domain invitation email."""
+        with less_console_noise():
+            self.run_load_domains()
+            self.run_transfer_domains()
+
+        # this is one of the email addresses in data/test_contacts.txt
+        output_stream = StringIO()
+        # also have to re-point the logging handlers to output_stream
+        with less_console_noise(output_stream):
+            call_command("send_domain_invitations", "testuser@gmail.com", stdout=output_stream)
+
+        # Check that we had the right numbers in our output
+        output = output_stream.getvalue()
+        # should only be one domain we send email for
+        self.assertIn("Found 1 transition domains", output)
+        self.assertTrue("would send email to testuser@gmail.com", output)
+
+    def test_send_domain_invitations_two_emails(self):
+        """Can send only a single domain invitation email."""
+        with less_console_noise():
+            self.run_load_domains()
+            self.run_transfer_domains()
+
+        # these are two email addresses in data/test_contacts.txt
+        output_stream = StringIO()
+        # also have to re-point the logging handlers to output_stream
+        with less_console_noise(output_stream):
+            call_command(
+                "send_domain_invitations", "testuser@gmail.com", "agustina.wyman7@test.com", stdout=output_stream
+            )
+
+        # Check that we had the right numbers in our output
+        output = output_stream.getvalue()
+        # should only be one domain we send email for
+        self.assertIn("Found 2 transition domains", output)
+        self.assertTrue("would send email to testuser@gmail.com", output)
+        self.assertTrue("would send email to agustina.wyman7@test.com", output)