applied bouncing mechanism

Author: qcdyx

functions-python/refresh_materialized_view/src/main.py

@@ -2,8 +2,9 @@
 import os
 import json
 from google.cloud import tasks_v2
-from datetime import datetime
+from datetime import datetime, timedelta
 import functions_framework
+from shared.helpers import timezone
 from shared.helpers.logger import init_logger
 from shared.database.database import with_db_session
 
@@ -21,9 +22,16 @@ def refresh_materialized_view_function(request):
     """
     try:
         logging.info("Starting materialized view refresh function.")
+        now = datetime.now(timezone.utc)
+        delay_minutes = 5
+        # Round up to the next 5-minute mark (bounce window)
+        delay_target = now + timedelta(minutes=delay_minutes)
+        bucket_time = delay_target.replace(
+            minute=(delay_target.minute // 5) * 5, second=0, microsecond=0
+        )
 
         # Generate deduplication key based on current timestamp
-        timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M")
+        timestamp = bucket_time.strftime("%Y-%m-%d-%H-%M")
         task_name = f"refresh-materialized-view-{timestamp}"
 
         # Cloud Tasks client setup
@@ -46,6 +54,7 @@ def refresh_materialized_view_function(request):
                 "headers": {"Content-Type": "application/json"},
                 "body": json.dumps(payload).encode(),
             },
+            "schedule_time": timestamp,
         }
 
         # Enqueue the task
@@ -70,7 +79,7 @@ def refresh_materialized_view_task(request, db_session):
         dict: Response message and status code.
     """
     try:
-        logging.info("Starting materialized view refresh task.")
+        logging.info("Materialized view refresh task initiated.")
 
         data = request.json()
         view_name = data.get("view_name")
@@ -81,7 +90,6 @@ def refresh_materialized_view_task(request, db_session):
             f"{view_name} with key: {deduplication_key}"
         )
 
-        # Call the refresh function
         success = refresh_materialized_view(db_session, view_name)
 
         if success:

functions-python/refresh_materialized_view/tests/test_main.py

@@ -1,77 +1,35 @@
 import pytest
-from unittest.mock import Mock, patch
-import sys
-import os
+from fastapi.testclient import TestClient
+from src.main import app
 
-# Add the src directory to the path
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src"))
+client = TestClient(app)
 
-from main import refresh_materialized_view_function  # noqa: E402
 
+@pytest.fixture
+def mock_environment_variables(monkeypatch):
+    monkeypatch.setenv("PROJECT_ID", "test-project")
+    monkeypatch.setenv("QUEUE_NAME", "test-queue")
+    monkeypatch.setenv("LOCATION", "us-central1")
+    monkeypatch.setenv("FUNCTION_URL", "http://localhost")
 
-class TestRefreshMaterializedView:
-    """Test class for the refresh materialized view function."""
 
-    @patch("main.refresh_materialized_view")
-    def test_refresh_view_success(self, mock_refresh):
-        """Test successful view refresh."""
-        mock_refresh.return_value = True
+@pytest.fixture
+def mock_cloud_tasks_client(mocker):
+    mock_client = mocker.Mock()
+    mocker.patch("src.main.tasks_v2.CloudTasksClient", return_value=mock_client)
+    return mock_client
 
-        # Mock request object
-        request = Mock()
-        request.method = "GET"
 
-        # Mock database session
-        db_session = Mock()
+@pytest.mark.parametrize("endpoint", ["/refresh-materialized-view"])
+def test_refresh_materialized_view_function(endpoint):
+    response = client.post(endpoint, json={})
+    assert response.status_code == 200
+    assert "Task" in response.json()["message"]
 
-        # Call the function
-        response, status_code = refresh_materialized_view_function(request, db_session)
 
-        # Assertions
-        assert "Successfully refreshed materialized view" in response["message"]
-        mock_refresh.assert_called_once_with(db_session, "my_materialized_view")
-
-    @patch("main.refresh_materialized_view")
-    def test_refresh_view_failure(self, mock_refresh):
-        """Test failure to refresh the view."""
-        mock_refresh.return_value = False
-
-        # Mock request object
-        request = Mock()
-        request.method = "GET"
-
-        # Mock database session
-        db_session = Mock()
-
-        # Call the function
-        response, status_code = refresh_materialized_view_function(request, db_session)
-
-        # Assertions
-        assert status_code == 500
-        assert "Failed to refresh materialized view" in response["error"]
-        mock_refresh.assert_called_once_with(db_session, "my_materialized_view")
-
-    @patch("main.refresh_materialized_view")
-    def test_refresh_view_exception(self, mock_refresh):
-        """Test exception during view refresh."""
-        mock_refresh.side_effect = Exception("Database connection failed")
-
-        # Mock request object
-        request = Mock()
-        request.method = "GET"
-
-        # Mock database session
-        db_session = Mock()
-
-        # Call the function
-        response, status_code = refresh_materialized_view_function(request, db_session)
-
-        # Assertions
-        assert status_code == 500
-        assert "Error refreshing materialized view" in response["error"]
-        assert "Database connection failed" in response["error"]
-        mock_refresh.assert_called_once_with(db_session, "my_materialized_view")
-
-
-if __name__ == "__main__":
-    pytest.main([__file__])
+@pytest.mark.parametrize("endpoint", ["/refresh-materialized-view-task"])
+def test_refresh_materialized_view_task(endpoint):
+    payload = {"view_name": "test_view", "deduplication_key": "test_key"}
+    response = client.post(endpoint, json=payload)
+    assert response.status_code == 200
+    assert "Successfully refreshed materialized view" in response.json()["message"]

infra/functions-python/main.tf

@@ -1518,6 +1518,7 @@ resource "google_cloudfunctions2_function" "refresh_materialized_view" {
       PROJECT_ID  = var.project_id
       QUEUE_NAME  = google_cloud_tasks_queue.refresh_materialized_view_queue.name
       LOCATION    = var.gcp_region
+      FUNCTION_URL = "https://${var.gcp_region}-${var.project_id}.cloudfunctions.net/refresh_materialized_view"
     }
     available_memory                 = "512Mi"
     timeout_seconds                  = 60