Fallback to use subscription cred for analytic upload (#15479) (#6668)

* Fallback to use subscription cred for analytic

Fall back to use SUBSCRIPTION_USERNAME/PASSWORD to upload analytic to if REDHAT_USERNAME/PASSWORD are not set

* Improve error message

* Guard against request with no query or data

* Add test for _send_to_analytics

Focus on credentials

* Supress sonarcloud warning about password

* Add test for analytic ship

Co-authored-by: Hao Liu <44379968+TheRealHaoLiu@users.noreply.github.com>

awx/main/tests/functional/analytics/test_core.py

@@ -2,11 +2,13 @@ import importlib
 import json
 import os
 import tarfile
+import tempfile
 from unittest import mock
 import pytest
 
 from django.conf import settings
-from awx.main.analytics import gather, register
+from django.test.utils import override_settings
+from awx.main.analytics import gather, register, ship
 
 
 @register('example', '1.0')
@@ -57,3 +59,83 @@ def test_gather(mock_valid_license):
             os.remove(tgz)
     except Exception:
         pass
+
+
+@pytest.fixture
+def temp_analytic_tar():
+    # Create a temporary file and yield its path
+    with tempfile.NamedTemporaryFile(delete=False) as temp_file:
+        temp_file.write(b"data")
+        temp_file_path = temp_file.name
+    yield temp_file_path
+    # Clean up the temporary file after the test
+    os.remove(temp_file_path)
+
+
+@pytest.fixture
+def mock_analytic_post():
+    # Patch the Session.post method to return a mock response with status_code 200
+    with mock.patch('awx.main.analytics.core.requests.Session.post', return_value=mock.Mock(status_code=200)) as mock_post:
+        yield mock_post
+
+
+@pytest.mark.parametrize(
+    "setting_map, expected_result, expected_auth",
+    [
+        # Test case 1: Valid Red Hat credentials
+        (
+            {
+                'REDHAT_USERNAME': 'redhat_user',
+                'REDHAT_PASSWORD': 'redhat_pass',  # NOSONAR
+                'SUBSCRIPTION_USERNAME': None,
+                'SUBSCRIPTION_PASSWORD': None,
+            },
+            True,
+            ('redhat_user', 'redhat_pass'),
+        ),
+        # Test case 2: Valid Subscription credentials
+        (
+            {
+                'REDHAT_USERNAME': None,
+                'REDHAT_PASSWORD': None,
+                'SUBSCRIPTION_USERNAME': 'subs_user',
+                'SUBSCRIPTION_PASSWORD': 'subs_pass',  # NOSONAR
+            },
+            True,
+            ('subs_user', 'subs_pass'),
+        ),
+        # Test case 3: No credentials
+        (
+            {
+                'REDHAT_USERNAME': None,
+                'REDHAT_PASSWORD': None,
+                'SUBSCRIPTION_USERNAME': None,
+                'SUBSCRIPTION_PASSWORD': None,
+            },
+            False,
+            None,  # No request should be made
+        ),
+        # Test case 4: Mixed credentials
+        (
+            {
+                'REDHAT_USERNAME': None,
+                'REDHAT_PASSWORD': 'redhat_pass',  # NOSONAR
+                'SUBSCRIPTION_USERNAME': 'subs_user',
+                'SUBSCRIPTION_PASSWORD': None,
+            },
+            False,
+            None,  # Invalid, no request should be made
+        ),
+    ],
+)
+@pytest.mark.django_db
+def test_ship_credential(setting_map, expected_result, expected_auth, temp_analytic_tar, mock_analytic_post):
+    with override_settings(**setting_map):
+        result = ship(temp_analytic_tar)
+
+        assert result == expected_result
+        if expected_auth:
+            mock_analytic_post.assert_called_once()
+            assert mock_analytic_post.call_args[1]['auth'] == expected_auth
+        else:
+            mock_analytic_post.assert_not_called()

awx/main/tests/functional/api/test_analytics.py

@@ -1,7 +1,10 @@
 import pytest
 import requests
-from awx.api.views.analytics import AnalyticsGenericView, MissingSettings, AUTOMATION_ANALYTICS_API_URL_PATH
+from unittest import mock
+from awx.api.views.analytics import AnalyticsGenericView, MissingSettings, AUTOMATION_ANALYTICS_API_URL_PATH, ERROR_MISSING_USER, ERROR_MISSING_PASSWORD
 from django.test.utils import override_settings
+from django.test import RequestFactory
+from rest_framework import status
 
 from awx.main.utils import get_awx_version
 from django.utils import translation
@@ -84,3 +87,102 @@ class TestAnalyticsGenericView:
                     AnalyticsGenericView._get_setting(setting_name, False, None)
             else:
                 assert AnalyticsGenericView._get_setting(setting_name, False, None) == setting_value
+
+    @pytest.mark.parametrize(
+        "settings_map, expected_auth, expected_error_keyword",
+        [
+            # Test case 1: Valid Red Hat credentials
+            (
+                {
+                    'INSIGHTS_TRACKING_STATE': True,
+                    'REDHAT_USERNAME': 'redhat_user',
+                    'REDHAT_PASSWORD': 'redhat_pass',  # NOSONAR
+                    'SUBSCRIPTIONS_USERNAME': '',
+                    'SUBSCRIPTIONS_PASSWORD': '',
+                },
+                ('redhat_user', 'redhat_pass'),
+                None,
+            ),
+            # Test case 2: Valid Subscription credentials
+            (
+                {
+                    'INSIGHTS_TRACKING_STATE': True,
+                    'REDHAT_USERNAME': '',
+                    'REDHAT_PASSWORD': '',
+                    'SUBSCRIPTIONS_USERNAME': 'subs_user',
+                    'SUBSCRIPTIONS_PASSWORD': 'subs_pass',  # NOSONAR
+                },
+                ('subs_user', 'subs_pass'),
+                None,
+            ),
+            # Test case 3: No credentials
+            (
+                {
+                    'INSIGHTS_TRACKING_STATE': True,
+                    'REDHAT_USERNAME': '',
+                    'REDHAT_PASSWORD': '',
+                    'SUBSCRIPTIONS_USERNAME': '',
+                    'SUBSCRIPTIONS_PASSWORD': '',
+                },
+                None,
+                ERROR_MISSING_USER,
+            ),
+            # Test case 4: Both credentials
+            (
+                {
+                    'INSIGHTS_TRACKING_STATE': True,
+                    'REDHAT_USERNAME': 'redhat_user',
+                    'REDHAT_PASSWORD': 'redhat_pass',  # NOSONAR
+                    'SUBSCRIPTIONS_USERNAME': 'subs_user',
+                    'SUBSCRIPTIONS_PASSWORD': 'subs_pass',  # NOSONAR
+                },
+                ('redhat_user', 'redhat_pass'),
+                None,
+            ),
+            # Test case 5: Missing password
+            (
+                {
+                    'INSIGHTS_TRACKING_STATE': True,
+                    'REDHAT_USERNAME': '',
+                    'REDHAT_PASSWORD': '',
+                    'SUBSCRIPTIONS_USERNAME': 'subs_user',  # NOSONAR
+                    'SUBSCRIPTIONS_PASSWORD': '',
+                },
+                None,
+                ERROR_MISSING_PASSWORD,
+            ),
+        ],
+    )
+    @pytest.mark.django_db
+    def test__send_to_analytics_credentials(self, settings_map, expected_auth, expected_error_keyword):
+        with override_settings(**settings_map):
+            request = RequestFactory().post('/some/path')
+            view = AnalyticsGenericView()
+
+            if expected_auth:
+                with mock.patch('requests.request') as mock_request:
+                    mock_request.return_value = mock.Mock(status_code=200)
+
+                    analytic_url = view._get_analytics_url(request.path)
+                    response = view._send_to_analytics(request, 'POST')
+
+                    # Assertions
+                    mock_request.assert_called_once_with(
+                        'POST',
+                        analytic_url,
+                        auth=expected_auth,
+                        verify=mock.ANY,
+                        headers=mock.ANY,
+                        json=mock.ANY,
+                        params=mock.ANY,
+                        timeout=mock.ANY,
+                    )
+                    assert response.status_code == 200
+            else:
+                # Test when settings are missing and MissingSettings is raised
+                response = view._send_to_analytics(request, 'POST')
+
+                # # Assert that _error_response is called when MissingSettings is raised
+                # mock_error_response.assert_called_once_with(expected_error_keyword, remote=False)
+                assert response.status_code == status.HTTP_403_FORBIDDEN
+                assert response.data['error']['keyword'] == expected_error_keyword