Tooling for improved local dev experience (#1423)

* First commit of the devtools work.

This will configure and launch a stack. It isn't grabbing some things correctly, because of some more env-specific configuration that needs to be done. However, it's a solid start.

* Updating; looks like it builds/runs.

* Stack talks to BFD

This worked for test. There is now more cleanup.

There's a notional split between "setting up to run" and "getting the variables into the container."

This moves all the work out of the container. It could, perhaps, all be moved into the container if the correct env was passed for secrets gathering?

* A cleaner/quieter boot sequence.

* Improving local build.

* Fix where check happens

* adding extra around kion setup

* Add more checking, change params.

* Updated/all tests passing

* Removing script, out of scope

* Incremental/bug fixes.

* Fixes.

* Fixed sandbox FHIRv3 URL.

* Removing comments, unneeded prints.

---------

Co-authored-by: Brandon Wang <[email protected]>
Co-authored-by: James Demery <[email protected]>

Author: 3 people

Makefile

@@ -15,3 +15,9 @@ reqs-install:
 
 reqs-install-dev:
 	pip install -r requirements/requirements.dev.txt --no-index --find-links ./vendor/
+
+build-local:
+	cd dev-local ; make build-local ; cd ..
+
+run-local:
+	cd dev-local ; make run-local ; cd ..

apps/authorization/views.py

@@ -32,7 +32,6 @@ class Meta:
         fields = ('id', 'name', 'logo_uri', 'tos_uri', 'policy_uri', 'contacts')
 
     def get_contacts(self, obj):
-        print(obj)
         application = Application.objects.get(id=obj.id)
         return application.support_email or ""
 

apps/core/management/commands/create_test_feature_switches.py

@@ -34,7 +34,6 @@ def handle(self, *args, **options):
         for switch in WAFFLE_FEATURE_SWITCHES:
             try:
                 Switch.objects.get(name=switch[0])
-                self._log("Feature switch already exists: %s" % (str(switch)))
             except Switch.DoesNotExist:
                 Switch.objects.create(name=switch[0], active=switch[1], note=switch[2])
                 self._log("Feature switch created: %s" % (str(switch)))
@@ -46,7 +45,6 @@ def handle(self, *args, **options):
 
             try:
                 flag_obj = Flag.objects.get(name=flag[0])
-                self._log("Feature flag already exists: %s" % (str(flag_obj)))
             except Flag.DoesNotExist:
                 flag_obj = Flag.objects.create(name=flag[0])
                 self._log("Feature flag created: %s" % (str(flag[0])))
@@ -62,7 +60,6 @@ def handle(self, *args, **options):
                                 flag_obj.save()
                                 self._log("User {} added to feature flag: {}".format(u, flag))
                             except Exception as e:
-                                print(e)
                                 self._log("Exception when adding user {} to feature flag: {}".format(u, flag))
                         except User.DoesNotExist:
                             # assuming test users exist before creating flags associated with them

apps/dot_ext/tests/test_views.py

@@ -24,6 +24,8 @@
     SCOPES_TO_URL_BASE_PATH,
 )
 
+import os
+
 from hhs_oauth_server.settings.base import MOCK_FHIR_ENDPOINT_HOSTNAME
 
 
@@ -576,16 +578,15 @@ def test_delete_token_success(self):
 
         # This assertion is incorrectly crafted - it actually requires a local server started
         # so that the fhir fetch data is called and hence generate cert file not found error.
-        # TODO: refactor test to not depend on a server up and running.
-
-        # Post Django 2.2:  An OSError exception is expected when trying to reach the
-        #                   backend FHIR server and proves authentication worked.
-        with self.assertRaisesRegexp(
-            OSError, 'Could not find the TLS certificate file'
-        ):
-            response = self.client.get(
-                '/v1/fhir/Patient', headers={'authorization': 'Bearer ' + anna_token.token}
-            )
+        # 20251120 This test is now gated on a variable; if the variable does not exist, or
+        # is not set, the test will run. This is the desired behavior.
+        if os.getenv("RUNNING_IN_LOCAL_STACK", None) != "true":
+            with self.assertRaisesRegexp(
+                OSError, 'Could not find the TLS certificate file'
+            ):
+                response = self.client.get(
+                    '/v1/fhir/Patient', headers={'authorization': 'Bearer ' + anna_token.token}
+                )
 
         bob_tkn = self._create_test_token(bob, bob_application)
         self.assertTrue(
@@ -638,24 +639,26 @@ def test_delete_token_success(self):
 
         # Post Django 2.2:  An OSError exception is expected when trying to reach the
         #                   backend FHIR server and proves authentication worked.
-        with self.assertRaisesRegexp(
-            OSError, 'Could not find the TLS certificate file'
-        ):
-            response = self.client.get(
-                '/v1/fhir/Patient', headers={'authorization': 'Bearer ' + bob_tkn.token}
-            )
+        if os.getenv("RUNNING_IN_LOCAL_STACK", None) != "true":
+            with self.assertRaisesRegexp(
+                OSError, 'Could not find the TLS certificate file'
+            ):
+                response = self.client.get(
+                    '/v1/fhir/Patient', headers={'authorization': 'Bearer ' + bob_tkn.token}
+                )
 
         next_tkn = self._create_test_token(anna, anna_application)
 
         # Post Django 2.2:  An OSError exception is expected when trying to reach the
         #                   backend FHIR server and proves authentication worked.
-        with self.assertRaisesRegexp(
-            OSError, 'Could not find the TLS certificate file'
-        ):
-            response = self.client.get(
-                '/v1/fhir/Patient',
-                headers={'authorization': 'Bearer ' + next_tkn.token},
-            )
+        if os.getenv("RUNNING_IN_LOCAL_STACK", None) != "true":
+            with self.assertRaisesRegexp(
+                OSError, 'Could not find the TLS certificate file'
+            ):
+                response = self.client.get(
+                    '/v1/fhir/Patient',
+                    headers={'authorization': 'Bearer ' + next_tkn.token},
+                )
 
         # self.assertEqual(next_tkn.token, tkn.token)
         self.assertTrue(

apps/fhir/bluebutton/tests/test_wellknown_endpoints.py

@@ -147,11 +147,10 @@ def test_smart_configuration_missing_fields_in_v3(self):
     # is commented above for reference.
 
     @skipIf((not settings.RUN_ONLINE_TESTS), "Can't reach external sites.")
-    # This overrides the switch and sets it to true, always.
-    # We should only run the test if we have v3 enabled.
     @override_switch('v3_endpoints', active=True)
     def test_fhir_metadata_extensions_have_v3(self):
-        response = self.client.get(f'{BASEURL}/v3/fhir/metadata')
+        the_url = f'{BASEURL}/v3/fhir/metadata'
+        response = self.client.get(the_url)
         self.assertEqual(response.status_code, 200)
         json = response.json()
         self.assertIn('v3', json['implementation']['url'])

apps/testclient/management/commands/create_test_user_and_application.py

@@ -12,10 +12,15 @@
 from datetime import timedelta, datetime
 from django.conf import settings
 from apps.authorization.models import update_grants
+from apps.authorization.models import ArchivedDataAccessGrant, DataAccessGrant
 
+# Imports for quieting things during startup.
+from waffle.models import Switch
+
+from uuid import uuid4
 
-def create_group(name="BlueButton"):
 
+def create_group(name="BlueButton"):
     g, created = Group.objects.get_or_create(name=name)
     if created:
         print("%s group created" % (name))
@@ -24,42 +29,29 @@ def create_group(name="BlueButton"):
     return g
 
 
-def create_user(group, usr):
-    u_name = "fred"
-    first_name = "Fred" 
-    last_name = "Flinstone"
-    email = "[email protected]"
-    password = "foobarfoobarfoobar"
+def create_user(the_group):
+    username = "rogersf"
+    first_name = "Fred"
+    last_name = "Rogers"
+    email = "[email protected]"
+    password = uuid4()
     user_type = "BEN"
-    
-    if usr is not None:
-        u_name = usr
-        first_name = "{}{}".format(usr, "First") 
-        last_name = "{}{}".format(usr, "Last")
-        email = "{}.{}@example.com".format(first_name, last_name)
-        user_type = "DEV"
-
-
-    if User.objects.filter(username=u_name).exists():
-        User.objects.filter(username=u_name).delete()
-
-    u = None
-
-    if usr is not None:
-        u = User.objects.create_user(username=u_name,
-                                    first_name=first_name,
-                                    last_name=last_name,
-                                    email=email)
-        u.set_unusable_password()
-    else:
-        # create a sample user 'fred' for dev local that has a usable password
-        u = User.objects.create_user(username=u_name,
-                                    first_name=first_name,
-                                    last_name=last_name,
-                                    email=email,
-                                    password=password,)
-
-    UserProfile.objects.create(user=u,
+
+    # We will do this over-and-over.
+    # If we don't already exist, then create the user.
+    if User.objects.filter(username=username).exists():
+        print(f"👟 {username} already exists. Skipping test user creation.")
+        return User.objects.get(username=username)
+
+    # If the user didn't exist, it is our first time through.
+    # Create the user.
+    user_obj = User.objects.create(username=username,
+                                   first_name=first_name,
+                                   last_name=last_name,
+                                   email=email,
+                                   password=password,)
+    user_obj.set_unusable_password()
+    UserProfile.objects.create(user=user_obj,
                                user_type=user_type,
                                create_applications=True,
                                password_reset_question_1='1',
@@ -68,33 +60,35 @@ def create_user(group, usr):
                                password_reset_answer_2='Frank',
                                password_reset_question_3='3',
                                password_reset_answer_3='Bentley')
+    user_obj.groups.add(the_group)
 
-    u.groups.add(group)
+    # CROSSWALK
+    # Removing any existing crosswalks for this artificial user.
+    # Why? Just in case.
+    user_id_hash = "ee78989d1d9ba0b98f3cfbd52479f10c7631679c17563186f70fbef038cc9536"
+    Crosswalk.objects.filter(_user_id_hash=user_id_hash).delete()
+    Crosswalk.objects.get_or_create(user=user_obj,
+                                    fhir_id_v2=settings.DEFAULT_SAMPLE_FHIR_ID_V2,
+                                    _user_id_hash=user_id_hash)
+    return user_obj
 
-    if usr is None:
-        c, g_o_c = Crosswalk.objects.get_or_create(user=u,
-                                                   fhir_id_v2=settings.DEFAULT_SAMPLE_FHIR_ID_V2,
-                                                   _user_id_hash="ee78989d1d9ba0b98f3cfbd52479f10c7631679c17563186f70fbef038cc9536")
-    return u
 
+def create_application(user):
+    app_name = "TestApp"
+    if Application.objects.filter(name=app_name).exists():
+        return Application.objects.get(name=app_name)
+
+    # If the app doesn't exist, create the test app.
 
-def create_application(user, group, app, redirect):
-    app_name = "TestApp" if app is None else app
     Application.objects.filter(name=app_name).delete()
     redirect_uri = "{}{}".format(settings.HOSTNAME_URL, settings.TESTCLIENT_REDIRECT_URI)
 
-    if redirect:
-        redirect_uri = redirect
-
-    if not(redirect_uri.startswith("http://") or redirect_uri.startswith("https://")):
-        redirect_uri = "https://" + redirect_uri
-
-    a = Application.objects.create(name=app_name,
-                                redirect_uris=redirect_uri,
-                                user=user,
-                                data_access_type="THIRTEEN_MONTH",
-                                client_type="confidential",
-                                authorization_grant_type="authorization-code")
+    the_app = Application.objects.create(name=app_name,
+                                         redirect_uris=redirect_uri,
+                                         user=user,
+                                         data_access_type="THIRTEEN_MONTH",
+                                         client_type="confidential",
+                                         authorization_grant_type="authorization-code",)
 
     titles = ["My Medicare and supplemental coverage information.",
               "My Medicare claim information.",
@@ -104,49 +98,64 @@ def create_application(user, group, app, redirect):
 
     for t in titles:
         c = ProtectedCapability.objects.get(title=t)
-        a.scope.add(c)
-    return a
+        the_app.scope.add(c)
+
+    return the_app
 
 
-def create_test_token(user, application):
+def create_test_token(the_user, the_app):
 
+    # Set expiration one day from now.
     now = timezone.now()
     expires = now + timedelta(days=1)
 
-    scopes = application.scope.all()
+    scopes = the_app.scope.all()
     scope = []
     for s in scopes:
         scope.append(s.slug)
 
-    t = AccessToken.objects.create(user=user, application=application,
+    # We have to have a tokent with token="sample-token-string", because we
+    # rely on it existing for unit tests. Which are actually integration tests.
+    if AccessToken.objects.filter(token="sample-token-string").exists():
+        t = AccessToken.objects.get(token="sample-token-string")
+        t.expires = expires
+        t.save()
+    else:
+        AccessToken.objects.create(user=the_user,
+                                   application=the_app,
+                                   # This needs to be "sample-token-string", because
+                                   # we have tests that rely on it.
                                    token="sample-token-string",
                                    expires=expires,
-                                   scope=' '.join(scope))
-    return t
+                                   scope=' '.join(scope),)
+
+
+def get_switch(name):
+    try:
+        sw = Switch.objects.get(name=name)
+        return sw.active
+    except Exception as e:
+        print(f"Could not get switch {name}: {e}")
+
+
+def set_switch(name, b):
+    sw, _ = Switch.objects.get_or_create(name=name)
+    sw.active = b
+    sw.save()
 
 
 class Command(BaseCommand):
     help = 'Create a test user and application for the test client'
 
-    def add_arguments(self, parser):
-        parser.add_argument("-u", "--user", help="Name of the user to be created (unique).")
-        parser.add_argument("-a", "--app", help="Name of the application to be created (unique).")
-        parser.add_argument("-r", "--redirect", help="Redirect url of the application.")
-
     def handle(self, *args, **options):
-        usr = options["user"]
-        app = options["app"]
-        redirect = options["redirect"]
-
-        g = create_group()
-        u = create_user(g, usr)
-        a = create_application(u, g, app, redirect)
-        t = None
-        if usr is None and app is None:
-            t = create_test_token(u, a)
-            update_grants()
-        print("Name:", a.name)
-        print("client_id:", a.client_id)
-        print("client_secret:", a.client_secret)
-        print("access_token:", t.token if t else "None")
-        print("redirect_uri:", a.redirect_uris)
+
+        set_switch('outreach_email', False)
+
+        the_group = create_group()
+        the_user = create_user(the_group)
+        the_app = create_application(the_user)
+        create_test_token(the_user, the_app)
+        update_grants()
+
+        # Restore switch to whatever it was.
+        set_switch('outreach_email', True)