Merge latest upstream changes for RabbitMQ credential fix

Author: deepanshutiwari27

CONTRIBUTING.md

@@ -3,7 +3,7 @@
 ## Join the Community
 We have a public Slack channel in the CHAOSS workspace, as well as public meetings.
 
-We encourage all contributors to join the [CHAOSS Slack workspace](https://chaoss.community/kb-getting-started/) and participate in the `#wg-collectoss-8knot` channel. Our meeting tumes are kept up to date in the Software section of the [CHAOSS Calendar](https://chaoss.community/chaoss-calendar/). We recommend subscribing to the CHAOSS Software calendar so you can automatically stay up to date with any schedule or timezone changes. If you can't attend these meetings, they are also recorded and made available on the [CHAOSS YouTube](https://www.youtube.com/@CHAOSStube).
+We encourage all contributors to join the [CHAOSS Slack workspace](https://chaoss.community/kb-getting-started/) and participate in the `#wg-collectoss-8knot` channel. Our meeting times are kept up to date in the Software section of the [CHAOSS Calendar](https://chaoss.community/chaoss-calendar/). We recommend subscribing to the CHAOSS Software calendar so you can automatically stay up to date with any schedule or timezone changes. If you can't attend these meetings, they are also recorded and made available on the [CHAOSS YouTube](https://www.youtube.com/@CHAOSStube).
 
 These resources are a great way to meet the people behind the project, ask questions, get help, participate in discussions, and stay updated on community meetings and planning. Everyone is welcome, so feel free to introduce yourself and ask for help if you get stuck or feel frustrated with any part of this setup process!
 

collectoss/application/db/materialized_views.py

@@ -26,6 +26,27 @@
 # target_metadata = mymodel.Base.metadata
 target_metadata = Base.metadata
 
+# NOTE FOR DEVELOPERS: alembic_utils manages materialized view definitions via
+# DROP + CREATE (replace). When a view is replaced:
+#
+#   1. ALL indexes are destroyed. Manually add CREATE UNIQUE INDEX statements
+#      after the replace op, using MaterializedView.unique_index_columns from
+#      the registry (collectoss/application/db/materialized_views.py).
+#
+#   2. The view is recreated WITH NO DATA. You CANNOT run REFRESH CONCURRENTLY
+#      immediately — it requires both a unique index and pre-existing data.
+#      After recreating the index, run a non-concurrent refresh first:
+#        REFRESH MATERIALIZED VIEW augur_data.<view_name> WITH DATA;
+#      Only after that will the Celery refresh task's CONCURRENTLY succeed.
+#
+# WARNING: If MATERIALIZED_VIEWS is ever emptied, autogenerate will propose
+# dropping all registered views. Keep the list complete.
+from alembic_utils.pg_materialized_view import PGMaterializedView
+from alembic_utils.replaceable_entity import register_entities
+from collectoss.application.db.materialized_views import MATERIALIZED_VIEWS
+_materialized_view_entities = [view.to_pg_view() for view in MATERIALIZED_VIEWS]
+register_entities(_materialized_view_entities, entity_types=[PGMaterializedView])
+
 # other values from the config, defined by the needs of env.py,
 # can be acquired:
 # my_important_option = config.get_main_option("my_important_option")

collectoss/application/schema/alembic/env.py

@@ -3,197 +3,57 @@
 import sqlalchemy as s
 
 from collectoss.tasks.init.celery_app import celery_app as celery
-from collectoss.application.db.lib import execute_sql
+from collectoss.application.db.materialized_views import MATERIALIZED_VIEWS
 from collectoss.tasks.git.util.facade_worker.facade_worker.config import FacadeHelper
 from collectoss.tasks.git.util.facade_worker.facade_worker.rebuildcache import invalidate_caches, rebuild_unknown_affiliation_and_web_caches
 
 
 @celery.task(bind=True)
 def refresh_materialized_views(self):
 
-    #self.logger = SystemLogger("data_collection_jobs").get_logger()
-
-    engine = self.app.engine
-
     logger = logging.getLogger(refresh_materialized_views.__name__)
-    #self.logger = logging.getLogger(refresh_materialized_views.__name__)
-
-    mv1_refresh = s.sql.text("""    
-                REFRESH MATERIALIZED VIEW concurrently data.api_get_all_repo_prs with data;
-                COMMIT; 
-    """)
-
-    mv2_refresh = s.sql.text("""    
-                REFRESH MATERIALIZED VIEW concurrently data.api_get_all_repos_commits with data;
-                COMMIT; 
-    """)
-
-    mv3_refresh = s.sql.text("""    
-                REFRESH MATERIALIZED VIEW concurrently data.api_get_all_repos_issues with data;
-                COMMIT; 
-    """)
-
-    mv4_refresh = s.sql.text("""    
-                REFRESH MATERIALIZED VIEW concurrently data.augur_new_contributors with data;
-                COMMIT; 
-    """)
-    mv5_refresh = s.sql.text("""    
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_commits_and_committers_daily_count with data;
-                COMMIT; 
-    """)
-
-    mv6_refresh = s.sql.text("""    
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_new_contributors with data;
-                COMMIT; 
-    """)
-
-    mv7_refresh = s.sql.text("""    
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_entry_list with data;
-                COMMIT; 
-    """)
-
-    mv8_refresh = s.sql.text("""    
-
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_contributor_actions with data;
-                COMMIT; 
-    """)
-
-    mv9_refresh = s.sql.text("""    
-
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_user_repos with data;
-                COMMIT; 
-    """)
-
-    mv10_refresh = s.sql.text("""    
-
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_pr_response_times with data;
-                COMMIT; 
-    """)
-
-    mv11_refresh = s.sql.text("""    
-
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_pr_assignments with data;
-                COMMIT; 
-    """)
-
-    mv12_refresh = s.sql.text("""    
-
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_issue_assignments with data;
-                COMMIT; 
-    """)
-
-    mv13_refresh = s.sql.text("""    
-
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_pr_response with data;
-                COMMIT; 
-    """)
-
-    mv14_refresh = s.sql.text("""    
 
-                REFRESH MATERIALIZED VIEW concurrently data.explorer_repo_languages with data;
-                COMMIT; 
-    """)
-
-    try: 
-        execute_sql(mv1_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv2_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv3_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv4_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv5_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv6_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv7_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv8_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv9_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv10_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv11_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv12_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv13_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
-
-    try: 
-        execute_sql(mv14_refresh)
-    except Exception as e: 
-        logger.info(f"error is {e}")
-        pass 
+    # REFRESH MATERIALIZED VIEW CONCURRENTLY cannot run inside a transaction
+    # block, so we use an autocommit connection rather than execute_sql().
+    failed_views = []
+    with self.app.engine.connect().execution_options(isolation_level="AUTOCOMMIT") as conn:
+        for view in MATERIALIZED_VIEWS:
+            logger.info(f"Refreshing materialized view: {view.fqn}")
+            
+            if len(view.unique_index_columns) > 0:
+                try:
+                    conn.execute(s.sql.text(view.refresh_sql(concurrently=True)))
+                    continue
+                except Exception as e:
+                    logger.warning(f"Concurrent refresh failed for {view.fqn}, trying non-concurrent: {e}")
+ 
+            try:
+                conn.execute(s.sql.text(view.refresh_sql(concurrently=False)))
+            except Exception as e2:
+                logger.error(f"Non-concurrent refresh failed for {view.fqn}: {e2}")
+                failed_views.append(view.fqn)
+
+    if failed_views:
+        raise RuntimeError(
+            f"{len(failed_views)} materialized view(s) failed to refresh: {failed_views}"
+        )
 
     #Now refresh facade tables
-    #Use this class to get all the settings and 
+    #Use this class to get all the settings and
     #utility functions for facade
     facade_helper = FacadeHelper(logger)
 
     if facade_helper.nuke_stored_affiliations:
         logger.error("Nuke stored affiliations is deprecated!")
-        # deprecated because the UI component of facade where affiliations would be 
-        # nuked upon change no longer exists, and this information can easily be derived 
+        # deprecated because the UI component of facade where affiliations would be
+        # nuked upon change no longer exists, and this information can easily be derived
         # from queries and materialized views in the current version of CollectOSS.
         # This method is also a major performance bottleneck with little value.
-    
+
     if not facade_helper.limited_run or (facade_helper.limited_run and facade_helper.fix_affiliations):
         logger.error("Fill empty affiliations is deprecated!")
-        # deprecated because the UI component of facade where affiliations would need 
-        # to be fixed upon change no longer exists, and this information can easily be derived 
+        # deprecated because the UI component of facade where affiliations would need
+        # to be fixed upon change no longer exists, and this information can easily be derived
         # from queries and materialized views in the current version of CollectOSS.
         # This method is also a major performance bottleneck with little value.
 
@@ -202,13 +62,9 @@ def refresh_materialized_views(self):
             invalidate_caches(facade_helper)
         except Exception as e:
             logger.info(f"error is {e}")
-    
+
     if not facade_helper.limited_run or (facade_helper.limited_run and facade_helper.rebuild_caches):
         try:
             rebuild_unknown_affiliation_and_web_caches(facade_helper)
         except Exception as e:
             logger.info(f"error is {e}")
-
-
-
-

collectoss/tasks/db/refresh_materialized_views.py

@@ -169,7 +169,7 @@ def make_request(self, url, method="GET", timeout=100):
                 raise UrlNotFoundException(f"Could not find {url}")
 
             if response.status_code == 401:
-                raise NotAuthorizedException(f"Could not authorize with the github api")
+                raise NotAuthorizedException(f"Could not authorize with the github api using key: {mask_key(self.key)}")
 
             if response.status_code == 410:
                 response_msg = response.json().get("message")

collectoss/tasks/github/util/github_data_access.py

@@ -45,7 +45,7 @@
                 <hr>
                 <div class="dropdown">
                     <a href="{{ url_for('root') }}" class="d-flex align-items-center text-white text-decoration-none">
-                        <img src="{{ url_for('logo') }}" alt="CollectOSS logo" height="32" class="rounded-circle me-2">
+                        <img src="{{ url_for('logo') }}" alt="CollectOSS logo" height="32" class="rounded me-2">
                     </a>
                     {# Reserved for future use
                     <ul class="dropdown-menu dropdown-menu-dark text-small shadow" aria-labelledby="dropdownUser1">

collectoss/templates/admin-dashboard.j2

@@ -40,7 +40,7 @@
                     class="d-flex flex-md-column text-truncate flex-row flex-grow-1 align-items-center align-items-md-start px-3 py-2 text-white">
                     <a href="{{ url_for('root') }}"
                         class="d-flex align-items-around mw-100 text-white text-decoration-none">
-                        <img src="{{ url_for('logo') }}" alt="CollectOSS logo" height="40" class="rounded-circle me-2">
+                        <img src="{{ url_for('logo') }}" alt="CollectOSS logo" height="40" class="rounded me-2">
                     </a>
                     <ul class="nav nav-pills flex-md-column flex-row flex-nowrap flex-shrink-1 flex-md-grow-0 flex-grow-1 mb-md-auto mb-0 justify-content-center align-items-center align-items-md-start"
                         id="menu">

collectoss/templates/settings.j2

@@ -1,5 +1,13 @@
 def mask_key(key: str, first: int = 6, last: int = 3, stars: int = 6) -> str:
     """Mask key except for the first and last few characters."""
-    if not isinstance(key, str) or len(key) <= (first + last):
-        return "*" * stars
-    return f"{key[:first]}{'*' * stars}{key[-last:]}"
+    if key is None:
+        return None
+
+    if isinstance(key, str):
+        if key == "":
+            return "*" * stars + f" Type: empty string"
+        if len(key) <= (first + last):
+            return "*" * stars
+        return f"{key[:first]}{'*' * stars}{key[-last:]}"
+    else:
+        return "*" * stars + f" Type: {str(type(key))}"

collectoss/util/keys.py

@@ -19,6 +19,7 @@ classifiers = [
 ]
 dependencies = [
     "alembic>=1.17.1",
+    "alembic-utils==0.8.8",
     "Beaker==1.11.0",
     "boto3==1.17.57",
     "bs4==0.0.1",
@@ -148,6 +149,7 @@ testpaths = [
     "tests/test_application/test_cli/test_csv_utils.py",
     "tests/test_tasks/test_task_utilities/test_util/",
     "tests/test_application/test_db/test_timestamp_utils.py",
+    "tests/test_util/test_keys.py",
     # "tests/test_routes", # runs, but needs a fixture for connecting to the web interface of Augur
     # "tests/test_metrics",
     # "tests/test_tasks",

pyproject.toml

@@ -0,0 +1,56 @@
+import pytest
+
+from collectoss.util.keys import mask_key
+
+
+def test_none():
+    assert mask_key(None) is None
+
+def test_empty_string():
+    assert mask_key("") == "*" * 6 + " Type: empty string"
+
+def test_non_string():
+    result = mask_key(12345)
+    assert "*" in result
+    assert str(type(12345)) in result
+
+
+def test_non_string_list():
+    result = mask_key([1, 2, 3])
+    assert "*" in result
+    assert str(type([])) in result
+
+
+def test_short_string():
+    assert mask_key("short") == f"******"
+
+
+def test_long_string_masked_correctly():
+    key = "ghp_abcdefghij"  # 14 chars → first 6 + 6 stars + last 3
+    assert mask_key(key) == "ghp_ab******hij"
+
+
+def test_exactly_one_over_boundary():
+    # 10 chars: first 6 + 6 stars + last 3
+    key = "1234567890"
+    assert mask_key(key) == "123456******890"
+
+
+def test_default_star_count_is_six():
+    key = "abcdefghijk"  # 11 chars
+    result = mask_key(key)
+    middle = result[6:-3]
+    assert middle == "******"
+
+
+def test_custom_first_last_stars():
+    # first=2, last=2, stars=3 → boundary=4; "hello" is 5 chars → masked
+    assert mask_key("hello", first=2, last=2, stars=3) == "he***lo"
+
+
+def test_custom_stars_count_in_output():
+    key = "abcdefghijk"
+    result = mask_key(key, stars=10)
+    assert result == "abcdef**********ijk"
+
+