From 9fdfc95d374347cec95847a29839dd766a482c4f Mon Sep 17 00:00:00 2001 From: Pratyush Bhandari <112966709+prat98@users.noreply.github.com> Date: Thu, 10 Sep 2026 10:24:01 -0400 Subject: [PATCH] [AAP-78694] Add CleanTextMixin to Controller Serializers (#16635) * AAP-78694 - Add CleanTextMixin to AWX resource serializers Applies django-ansible-base's CleanTextMixin to the AWX serializers listed in AAP-78694: Organization, User, Team, Credential(+Create), CredentialType, Project, JobTemplate, WorkflowJobTemplate, Inventory, Host, Group, InventorySource, InstanceGroup, Instance, ExecutionEnvironment, NotificationTemplate, Schedule, AdHocCommand, WorkflowJobTemplateNode, WorkflowJobTemplateNodeCreateApproval, JobLaunch, and WorkflowJobLaunch. Adds a local PromptFieldCleanTextMixin for serializers whose model derives from LaunchTimeConfigBase, where scm_branch/limit/job_tags/ skip_tags are NullablePromptPseudoField descriptors invisible to the mixin's normal model-field auto-discovery. Per-serializer excluded_fields/name_fields overrides are documented inline where a field's existing behavior would otherwise conflict with Tier 1/2 validation (Jinja2-templated fields, secret fields, and existing colon-containing name/hostname syntax). Signed-off-by: Pratyush Bhandari * AAP-78694 - Cover CleanTextMixin gaps in Copy, Bulk, and Job Detail paths Three write paths were found to bypass the Tier 1/2 validation applied elsewhere for the same resources: - CopySerializer (used by the Copy button/endpoint for EE, Project, Credential, JobTemplate, WorkflowJobTemplate, NotificationTemplate) accepted a user-supplied name with zero validation. Adds a new PlainSerializerCleanTextMixin for plain serializers.Serializer subclasses with no backing model. - BulkJobNodeSerializer's validate() intentionally skips LaunchConfigurationBaseSerializer.validate() (it assumes unified_job_template/inventory/execution_environment are model instances, but bulk nodes pass raw PKs to avoid a per-node DB lookup), which meant it also skipped CleanTextMixin. Adds PromptFieldCleanTextMixin._run_clean_text_validation() so the Tier 1/2 checks still run despite that skip. BulkJobLaunchSerializer and WorkflowJobNodeSerializer are also covered. - JobSerializer (backing the narrow JobDetail PUT/PATCH window while a job's status is "new") had no coverage at all. Confirmed live against a running controller that a Copy with a malicious name previously succeeded (201) where the equivalent PATCH on the same resource was already rejected (400); both are rejected consistently after this change. Signed-off-by: Pratyush Bhandari * AAP-78694 - Fix NotificationTemplateSerializer validate() bypass validate() returned early without calling super().validate() whenever notification_configuration was absent from the request (e.g. a PATCH that only changes name/description), which skipped CleanTextMixin entirely on that path. Same bug class as the BulkJobNodeSerializer MRO-skip already fixed, caught by code review. Confirmed live: a PATCH with only a malicious name (no notification_configuration) previously succeeded; now rejected with 400, while a legitimate name-only PATCH still succeeds. Signed-off-by: Pratyush Bhandari * AAP-78694 - Add CleanTextMixin to the previously out-of-scope API-only serializers LabelSerializer, CredentialInputSourceSerializer, and WorkflowApprovalTemplateSerializer were originally scoped out ("no UI exposure") but are now in scope per updated ticket guidance. WorkflowApprovalTemplateSerializer backs the update path for the same model WorkflowJobTemplateNodeCreateApprovalSerializer already protects on create -- without this, an approval name rejected at creation could be rewritten to anything via PATCH. Confirmed live: Label create and WorkflowApprovalTemplate update both reject a malicious name (400) and accept a valid one. CredentialInputSourceSerializer verified at the code level only (plain CharField input_field_name, no custom validators; metadata is a JSONSchemaField holding vault paths/keys, not secret values) -- setting up a live external-credential-type chain in this environment wasn't done. Signed-off-by: Pratyush Bhandari * AAP-78694 - Correct HostSerializer name_fields demotion comment The comment claimed the Tier 1 -> Tier 2 demotion was needed for "hostname:port" syntax, but that colon is stripped from attrs['name'] in validate() before it reaches CleanTextMixin -- that case never needed the demotion. What actually needs it: names with any colon count other than exactly 1, which _get_host_port_from_name passes through untouched -- most notably IPv6 addresses, which its own comment says it explicitly does not parse. Caught by code review. Confirmed live: an IPv6-style host name (2 colons) is accepted unmodified (201). Signed-off-by: Pratyush Bhandari * AAP-78694 - Validate non-secret notification_configuration keys Replaced the blanket notification_configuration exclusion with a precise excluded_json_keys covering only the actual secret fields. Unlike Credential.inputs (admin-definable custom types with arbitrary secret fields), notification_type is a fixed models.CharField(choices) -- every backend class ships in awx/main/notifications/ and is known at code-authoring time, so the secret key set can be a static list rather than resolved per-instance. Enumerated every init_parameters entry with type "password" across all 8 backends (email/irc/webhook password, slack/pagerduty token, twilio account_token, grafana grafana_key, awssns aws_secret_access_key/aws_session_token) and confirmed none of those names are used for a non-secret value in a different backend. Previously-unvalidated, UI-exposed fields like webhook url and slack channels are now covered. Confirmed live: a javascript: webhook url and a ' + + +@pytest.fixture +def enforce_clean_text(): + with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True): + yield + + +@pytest.mark.django_db +def test_credential_inputs_rejects_unsafe_non_secret_key(post, admin, credentialtype_ssh, enforce_clean_text): + assert 'username' not in credentialtype_ssh.secret_fields + response = post( + reverse('api:credential_list'), + { + 'name': 'ssh-cleantext-username', + 'credential_type': credentialtype_ssh.id, + 'user': admin.id, + 'inputs': {'username': UNSAFE_INPUT, 'password': UNSAFE_INPUT}, + }, + admin, + ) + assert response.status_code == 400 + assert 'username' in response.data['inputs'] + assert 'password' not in response.data['inputs'] + + +@pytest.mark.django_db +def test_credential_inputs_skips_secret_keys(post, admin, credentialtype_ssh, enforce_clean_text): + assert 'password' in credentialtype_ssh.secret_fields + response = post( + reverse('api:credential_list'), + { + 'name': 'ssh-cleantext-password', + 'credential_type': credentialtype_ssh.id, + 'user': admin.id, + 'inputs': {'username': 'ok-user', 'password': UNSAFE_INPUT}, + }, + admin, + ) + assert response.status_code == 201 + + +@pytest.mark.django_db +def test_credential_inputs_secret_fields_follow_custom_type_schema(post, admin, organization, enforce_clean_text): + credential_type = CredentialType( + kind='cloud', + name='CustomCleanTextType', + inputs={ + 'fields': [ + {'id': 'endpoint', 'label': 'Endpoint', 'type': 'string'}, + {'id': 'token', 'label': 'Token', 'type': 'string', 'secret': True}, + ] + }, + ) + credential_type.save() + assert credential_type.secret_fields == ['token'] + + rejected = post( + reverse('api:credential_list'), + { + 'name': 'custom-cleantext-endpoint', + 'organization': organization.pk, + 'credential_type': credential_type.pk, + 'inputs': {'endpoint': UNSAFE_INPUT, 'token': 'ok-token'}, + }, + admin, + ) + assert rejected.status_code == 400 + assert 'endpoint' in rejected.data['inputs'] + assert 'token' not in rejected.data['inputs'] + + accepted = post( + reverse('api:credential_list'), + { + 'name': 'custom-cleantext-token', + 'organization': organization.pk, + 'credential_type': credential_type.pk, + 'inputs': {'endpoint': 'https://example.invalid', 'token': UNSAFE_INPUT}, + }, + admin, + ) + assert accepted.status_code == 201 diff --git a/awx/main/tests/functional/api/test_inventory.py b/awx/main/tests/functional/api/test_inventory.py index f37af494ea..d77f1a9f8c 100644 --- a/awx/main/tests/functional/api/test_inventory.py +++ b/awx/main/tests/functional/api/test_inventory.py @@ -64,6 +64,44 @@ def test_inventory_host_name_unique(scm_inventory, post, admin_user): assert "A Group with that name already exists." in json.dumps(resp.data) +@pytest.mark.django_db +def test_inventory_host_inline_port(scm_inventory, post, admin_user): + with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True): + resp = post( + reverse('api:inventory_hosts_list', kwargs={'pk': scm_inventory.id}), + {'name': 'web.example.com:2222'}, + admin_user, + expect=201, + ) + assert resp.data['name'] == 'web.example.com' + assert '2222' in resp.data['variables'] + + +@pytest.mark.django_db +def test_inventory_host_ipv6_name(scm_inventory, post, admin_user): + """HostSerializer demotes name from Tier 1 so IPv6 (multiple colons) + is not rejected once enforcement is on (AAP-78694).""" + with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True): + resp = post( + reverse('api:inventory_hosts_list', kwargs={'pk': scm_inventory.id}), + {'name': '2001:db8::1'}, + admin_user, + expect=201, + ) + assert resp.data['name'] == '2001:db8::1' + + +@pytest.mark.django_db +def test_inventory_host_invalid_inline_port(scm_inventory, post, admin_user): + resp = post( + reverse('api:inventory_hosts_list', kwargs={'pk': scm_inventory.id}), + {'name': 'web.example.com:99999'}, + admin_user, + expect=400, + ) + assert 'Invalid port' in json.dumps(resp.data) + + @pytest.mark.django_db def test_inventory_host_list_ordering(scm_inventory, get, admin_user): # create 3 hosts, hit the inventory host list view 3 times and get the order visible there each time and compare diff --git a/awx/main/tests/functional/test_bulk.py b/awx/main/tests/functional/test_bulk.py index 65eda300e1..a75a06a798 100644 --- a/awx/main/tests/functional/test_bulk.py +++ b/awx/main/tests/functional/test_bulk.py @@ -1,6 +1,7 @@ import pytest from uuid import uuid4 +from unittest import mock from awx.api.versioning import reverse @@ -629,3 +630,44 @@ def test_bulk_host_create_performance_large_inventory(organization, inventory, p assert len(response.data['hosts']) == 10 assert Host.objects.filter(inventory=inventory).count() == 10010 + + +@pytest.mark.django_db +def test_bulk_job_launch_rejects_unsafe_limit(organization, inventory, project, post, user): + """PromptFieldCleanTextMixin._run_clean_text_validation must raise when + enforcement is on -- BulkJobNodeSerializer skips the normal validate() + chain that would otherwise do this (AAP-78694).""" + normal_user = user('normal_user', False) + organization.member_role.members.add(normal_user) + jt = JobTemplate.objects.create(name='my-jt', inventory=inventory, project=project, playbook='helloworld.yml') + jt.execute_role.members.add(normal_user) + with mock.patch('awx.api.serializers.get_setting', return_value=True): + response = post( + reverse('api:bulk_job_launch'), + {'name': 'Bulk Job Launch', 'jobs': [{'unified_job_template': jt.id, 'limit': ''}]}, + normal_user, + expect=400, + ) + assert 'limit' in str(response.data) + + +@pytest.mark.django_db +def test_bulk_job_launch_allows_jinja_extra_vars(organization, inventory, project, post, user): + """BulkJobLaunchSerializer stringifies extra_vars before CleanTextMixin + runs; excluded_fields must keep legitimate Jinja from being rejected + (AAP-78694).""" + normal_user = user('normal_user', False) + organization.member_role.members.add(normal_user) + jt = JobTemplate.objects.create(name='my-jt', inventory=inventory, project=project, playbook='helloworld.yml') + jt.execute_role.members.add(normal_user) + with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True): + post( + reverse('api:bulk_job_launch'), + { + 'name': 'Bulk Job Launch', + 'jobs': [{'unified_job_template': jt.id}], + 'extra_vars': {'foo': '{{ bar }}'}, + }, + normal_user, + expect=201, + ) diff --git a/awx/main/tests/functional/test_copy.py b/awx/main/tests/functional/test_copy.py index 7c156979a8..adb4aead4c 100644 --- a/awx/main/tests/functional/test_copy.py +++ b/awx/main/tests/functional/test_copy.py @@ -271,3 +271,30 @@ def test_notification_template_copy(post, get, notification_template_with_encryp assert decrypt_field(notification_template_with_encrypt, 'notification_configuration', 'token') == decrypt_field( notification_template_copy, 'notification_configuration', 'token' ) + + +@pytest.mark.django_db +def test_job_template_copy_rejects_same_name(post, job_template, admin): + response = post( + reverse('api:job_template_copy', kwargs={'pk': job_template.pk}), + {'name': job_template.name}, + admin, + expect=400, + ) + assert 'already named' in str(response.data) + + +@pytest.mark.django_db +def test_job_template_copy_rejects_unsafe_name(post, job_template, admin): + """CopySerializer.super().validate() must run CleanTextMixin — the same-name + check above never reaches it (AAP-78694).""" + unsafe_name = '' + assert job_template.name != unsafe_name + with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True): + response = post( + reverse('api:job_template_copy', kwargs={'pk': job_template.pk}), + {'name': unsafe_name}, + admin, + expect=400, + ) + assert 'name' in response.data diff --git a/awx/main/tests/functional/test_notifications.py b/awx/main/tests/functional/test_notifications.py index b66ddc319d..0ee99e03ee 100644 --- a/awx/main/tests/functional/test_notifications.py +++ b/awx/main/tests/functional/test_notifications.py @@ -111,6 +111,20 @@ def test_notification_template_simple_patch(patch, notification_template, admin) patch(reverse('api:notification_template_detail', kwargs={'pk': notification_template.id}), {'name': 'foo'}, admin, expect=200) +@pytest.mark.django_db +def test_notification_template_patch_rejects_unsafe_name(patch, notification_template, admin): + """Name-only PATCH omits notification_configuration; that branch must still + chain into CleanTextMixin.validate() (AAP-78694).""" + with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=True): + response = patch( + reverse('api:notification_template_detail', kwargs={'pk': notification_template.id}), + {'name': ''}, + admin, + expect=400, + ) + assert 'name' in response.data + + @pytest.mark.django_db def test_notification_template_invalid_notification_type(patch, notification_template, admin): patch(reverse('api:notification_template_detail', kwargs={'pk': notification_template.id}), {'notification_type': 'invalid'}, admin, expect=400) diff --git a/awx/main/tests/unit/api/serializers/test_cleantext_mixin.py b/awx/main/tests/unit/api/serializers/test_cleantext_mixin.py new file mode 100644 index 0000000000..aa3ede8491 --- /dev/null +++ b/awx/main/tests/unit/api/serializers/test_cleantext_mixin.py @@ -0,0 +1,116 @@ +from unittest import mock + +import pytest +from rest_framework.exceptions import PermissionDenied, ValidationError + +from ansible_base.lib.serializers.mixins import CleanTextMixin + +from awx.api.serializers import ( + BulkJobNodeSerializer, + CopySerializer, + CredentialSerializer, + ExecutionEnvironmentSerializer, + HostSerializer, +) + + +UNSAFE_INPUT = '' + + +def _passthrough_validate(self, attrs): + return attrs + + +class TestCopySerializerValidate: + def test_rejects_copy_with_the_same_name(self): + original = mock.Mock() + original.name = 'keep-me' + view = mock.Mock() + view.get_object.return_value = original + serializer = CopySerializer(context={'view': view}) + + with pytest.raises(ValidationError) as exc: + serializer.validate({'name': 'keep-me'}) + + assert 'already named' in str(exc.value.detail) + + +@pytest.mark.django_db +class TestHostSerializerPortParsing: + def test_extracts_inline_ssh_port(self): + serializer = HostSerializer() + assert serializer._get_host_port_from_name('web.example.com:2222') == ('web.example.com', 2222) + + @pytest.mark.parametrize('name', ['web.example.com:0', 'web.example.com:65536', 'web.example.com:notaport']) + def test_rejects_invalid_inline_port(self, name): + serializer = HostSerializer() + with pytest.raises(ValidationError): + serializer._get_host_port_from_name(name) + + def test_validate_moves_port_into_ansible_ssh_port(self): + serializer = HostSerializer() + with mock.patch.object(CleanTextMixin, 'validate', _passthrough_validate): + attrs = serializer.validate({'name': 'web.example.com:2222'}) + + assert attrs['name'] == 'web.example.com' + assert '"ansible_ssh_port": 2222' in attrs['variables'] + + +@pytest.mark.django_db +class TestCredentialSerializerValidate: + def test_rejects_updates_to_managed_credentials(self): + serializer = CredentialSerializer() + serializer.instance = mock.Mock(managed=True) + + with pytest.raises(PermissionDenied): + serializer.validate({'name': 'nope'}) + + def test_fail_closed_excludes_inputs_when_type_is_unknown(self): + serializer = CredentialSerializer() + with mock.patch.object(CleanTextMixin, 'validate', _passthrough_validate): + serializer.validate({'name': 'no-type', 'inputs': {'username': UNSAFE_INPUT}}) + + assert 'inputs' in serializer.excluded_fields + + +@pytest.mark.django_db +class TestBulkJobNodeCleanText: + def test_run_clean_text_validation_raises_when_enforced(self): + serializer = BulkJobNodeSerializer() + with mock.patch('awx.api.serializers.get_setting', return_value=True): + with pytest.raises(ValidationError) as exc: + serializer._run_clean_text_validation({'limit': UNSAFE_INPUT}) + + assert 'limit' in exc.value.detail + + +@pytest.mark.django_db +class TestExecutionEnvironmentSerializerValidate: + def test_rejects_non_registry_credentials(self): + serializer = ExecutionEnvironmentSerializer() + credential = mock.Mock(kind='ssh') + with pytest.raises(ValidationError): + serializer.validate_credential(credential) + + def test_accepts_registry_credentials(self): + credential = mock.Mock(kind='registry') + assert ExecutionEnvironmentSerializer().validate_credential(credential) is credential + + def test_rejects_organization_change(self): + serializer = ExecutionEnvironmentSerializer() + serializer.instance = mock.Mock(organization_id=1) + organization = mock.Mock(pk=2) + + with pytest.raises(ValidationError) as exc: + serializer.validate({'organization': organization}) + + assert 'organization' in exc.value.detail + + def test_allows_same_organization(self): + serializer = ExecutionEnvironmentSerializer() + serializer.instance = mock.Mock(organization_id=1) + org = mock.Mock(pk=1) + with mock.patch.object(CleanTextMixin, 'validate', _passthrough_validate): + attrs = serializer.validate({'organization': org}) + + assert attrs['organization'] is org diff --git a/awx/main/tests/unit/api/serializers/test_notification_template_serializers.py b/awx/main/tests/unit/api/serializers/test_notification_template_serializers.py index 22376c5d72..a6a6c44e15 100644 --- a/awx/main/tests/unit/api/serializers/test_notification_template_serializers.py +++ b/awx/main/tests/unit/api/serializers/test_notification_template_serializers.py @@ -30,6 +30,14 @@ class TestNotificationTemplateSerializer: {'started': {'body': '{{ job_metadata }}'}}, {'started': {'body': '{{ job.summary_fields.inventory.total_hosts }}'}}, {'started': {'body': 'Iñtërnâtiônàlizætiøn'}}, + { + 'workflow_approval': { + 'running': {'message': 'running'}, + 'approved': None, + 'timed_out': {'message': 'timed out'}, + 'denied': {'body': 'denied'}, + } + }, ], ) def test_valid_messages(self, valid_messages): @@ -55,6 +63,8 @@ class TestNotificationTemplateSerializer: {'started': {'message': '{{ job.id | bad_filter }}'}}, {'started': {'message': '{{ job.__class__ }}'}}, {'started': {'message': 'Newlines \n not allowed\n'}}, + {'workflow_approval': {'invalid_subevent': {}}}, + {'workflow_approval': {'running': 'should_be_dict'}}, ], ) def test_invalid__messages(self, invalid_messages): diff --git a/awx/main/tests/unit/api/serializers/test_uncovered_paths.py b/awx/main/tests/unit/api/serializers/test_uncovered_paths.py new file mode 100644 index 0000000000..0542c5fa31 --- /dev/null +++ b/awx/main/tests/unit/api/serializers/test_uncovered_paths.py @@ -0,0 +1,351 @@ +from unittest import mock + +import pytest +from rest_framework.exceptions import ValidationError + +from awx.api.serializers import ( + ActivityStreamSerializer, + AdHocCommandRelaunchSerializer, + AnsibleFactsSerializer, + GroupTreeSerializer, + HostSerializer, + JobCreateScheduleSerializer, + JobHostSummarySerializer, + JobRelaunchSerializer, + NotificationSerializer, + OpaQueryPathMixin, + ProjectUpdateEventSerializer, + UnifiedJobSerializer, + UnifiedJobStdoutSerializer, + WorkflowApprovalSerializer, + WorkflowJobLaunchSerializer, + reverse_gfk, +) +from awx.main.constants import ACTIVE_STATES +from awx.main.models import JobLaunchConfig, WorkflowApproval, WorkflowJobTemplate + + +def _mock(**kwargs): + """Build a Mock with a real `.name` attribute. + + unittest.mock.Mock(name=...) only sets the mock's repr name, so + getattr(obj, 'name') would return another Mock instead of the string. + """ + name = kwargs.pop('name', None) + obj = mock.Mock(**kwargs) + if name is not None: + obj.name = name + return obj + + +def test_reverse_gfk_empty_when_object_missing(): + assert reverse_gfk(None, None) == {} + assert reverse_gfk(object(), None) == {} + + +def test_reverse_gfk_uses_absolute_url(): + class Organization: + def get_absolute_url(self, request=None): + return '/api/v2/organizations/1/' + + assert reverse_gfk(Organization(), None) == {'organization': '/api/v2/organizations/1/'} + + +def test_opa_query_path_rejects_unencoded_url(): + serializer = OpaQueryPathMixin() + with pytest.raises(ValidationError): + serializer.validate_opa_query_path('not a valid path') + + +def test_unified_job_stdout_types(): + assert UnifiedJobStdoutSerializer().get_types() == [ + 'project_update', + 'inventory_update', + 'job', + 'ad_hoc_command', + 'system_job', + ] + + +def test_unified_job_launched_by(): + serializer = UnifiedJobSerializer() + assert serializer.get_launched_by(None) is None + obj = mock.Mock(launched_by={'id': 7}) + assert serializer.get_launched_by(obj) == {'id': 7} + + +def test_unified_job_sub_serializer_workflow_approval(): + obj = mock.MagicMock(spec=WorkflowApproval) + assert UnifiedJobSerializer().get_sub_serializer(obj) is WorkflowApprovalSerializer + + +def test_ansible_facts_to_representation(): + obj = mock.Mock(ansible_facts={'os': 'rhel'}) + assert AnsibleFactsSerializer().to_representation(obj) == {'os': 'rhel'} + + +def test_group_tree_children_none(): + assert GroupTreeSerializer().get_children(None) == {} + + +def test_adhoc_relaunch_to_representation(): + serializer = AdHocCommandRelaunchSerializer() + assert serializer.to_representation(None) == {} + obj = mock.Mock(passwords_needed_to_start=['ssh_password']) + assert serializer.to_representation(obj) == {'ssh_password': ''} + + +class TestHostToRepresentation: + def test_related_includes_last_job_links(self): + serializer = HostSerializer() + serializer.reverse = mock.Mock(return_value='/x/') + obj = mock.Mock(pk=1, instance_id=9) + obj.inventory.kind = 'constructed' + obj.inventory.pk = 2 + obj.latest_summary = mock.Mock(pk=3, job_id=4) + with mock.patch('awx.api.serializers.BaseSerializer.get_related', return_value={}): + related = serializer.get_related(obj) + assert 'last_job_host_summary' in related + assert 'last_job' in related + + def test_empty_obj_returns_parent_payload(self): + serializer = HostSerializer() + with mock.patch('awx.api.serializers.BaseSerializer.to_representation', return_value={'inventory': 1}): + assert serializer.to_representation(None) == {'inventory': 1} + + def test_nulls_missing_inventory(self): + serializer = HostSerializer() + obj = mock.Mock(inventory=None) + with mock.patch('awx.api.serializers.BaseSerializer.to_representation', return_value={'inventory': 9}): + assert serializer.to_representation(obj)['inventory'] is None + + +@pytest.mark.django_db +class TestJobRelaunchSerializerPaths: + def test_missing_credential_passwords(self): + serializer = JobRelaunchSerializer() + serializer.instance = mock.Mock(passwords_needed_to_start=['ssh_password', 'become_password']) + with pytest.raises(ValidationError): + serializer.validate_credential_passwords({'ssh_password': ''}) + + def test_to_representation_injects_password_keys(self): + view = mock.Mock(_raw_data_form_marker=True) + serializer = JobRelaunchSerializer(context={'view': view}) + obj = mock.Mock(passwords_needed_to_start=['ssh_password']) + with mock.patch('awx.api.serializers.BaseSerializer.to_representation', return_value={}): + assert serializer.to_representation(obj)['ssh_password'] == '' + + def test_passwords_needed_to_start(self): + serializer = JobRelaunchSerializer() + assert serializer.get_passwords_needed_to_start(None) == '' + obj = mock.Mock(passwords_needed_to_start=['ssh_password']) + assert serializer.get_passwords_needed_to_start(obj) == ['ssh_password'] + + def test_retry_counts_while_running(self): + serializer = JobRelaunchSerializer() + obj = mock.Mock(status=next(iter(ACTIVE_STATES))) + assert 'not available' in str(serializer.get_retry_counts(obj)) + + def test_retry_counts_when_finished(self): + serializer = JobRelaunchSerializer() + obj = mock.Mock(status='successful') + obj.retry_qs.return_value.count.return_value = 2 + counts = serializer.get_retry_counts(obj) + assert counts['all'] == 2 + assert counts['failed'] == 2 + + def test_validate_missing_project(self): + serializer = JobRelaunchSerializer() + serializer.instance = mock.Mock(project=None, inventory=mock.Mock(pending_deletion=False)) + with pytest.raises(ValidationError): + serializer.validate({}) + + def test_validate_missing_inventory(self): + serializer = JobRelaunchSerializer() + serializer.instance = mock.Mock(project=mock.Mock(), inventory=None) + with pytest.raises(ValidationError): + serializer.validate({}) + + def test_validate_inventory_pending_deletion(self): + serializer = JobRelaunchSerializer() + serializer.instance = mock.Mock(project=mock.Mock(), inventory=mock.Mock(pending_deletion=True)) + with pytest.raises(ValidationError): + serializer.validate({}) + + +class TestJobCreateScheduleSerializerPaths: + def test_can_schedule(self): + obj = mock.Mock(can_schedule=True) + assert JobCreateScheduleSerializer().get_can_schedule(obj) is True + + def test_summarize_copies_fk_fields(self): + obj = _mock(id=3, name='inv', description='d') + summary = JobCreateScheduleSerializer._summarize('host', obj) + assert summary['id'] == 3 + assert summary['name'] == 'inv' + + def test_get_prompts_summarizes_related_objects(self): + inventory = _mock(id=1, name='inv', description='', has_active_failures=False) + ee = _mock(id=2, name='ee', description='', image='img') + cred = _mock(id=3, name='cred', description='', kind='ssh', cloud=False, kubernetes=False, credential_type_id=1) + ig = _mock(id=4, name='ig', is_container_group=False) + config = mock.Mock() + config.prompts_dict.return_value = { + 'inventory': inventory, + 'execution_environment': ee, + 'credentials': [cred], + 'instance_groups': [ig], + 'labels': True, + } + job = mock.Mock(launch_config=config) + serializer = JobCreateScheduleSerializer() + serializer._summary_field_labels = mock.Mock(return_value=[{'id': 9, 'name': 'l'}]) + prompts = serializer.get_prompts(job) + assert prompts['inventory']['id'] == 1 + assert prompts['execution_environment']['id'] == 2 + assert prompts['credentials'][0]['id'] == 3 + assert prompts['instance_groups'][0]['id'] == 4 + assert prompts['labels'] == [{'id': 9, 'name': 'l'}] + + def test_get_prompts_without_launch_config(self): + class JobWithoutConfig: + @property + def launch_config(self): + raise JobLaunchConfig.DoesNotExist() + + prompts = JobCreateScheduleSerializer().get_prompts(JobWithoutConfig()) + assert 'all' in prompts + + +class TestNotificationSerializerBody: + def test_webhook_dict_body(self): + obj = mock.Mock(notification_type='webhook', body={'body': 'hello'}) + assert NotificationSerializer().get_body(obj) == 'hello' + + def test_webhook_json_string_body(self): + obj = mock.Mock(notification_type='pagerduty', body='{"ok": true}') + assert NotificationSerializer().get_body(obj) == {'ok': True} + + def test_webhook_invalid_json_string_body(self): + obj = mock.Mock(notification_type='awssns', body='not-json') + assert NotificationSerializer().get_body(obj) == 'not-json' + + def test_plain_body_passthrough(self): + obj = mock.Mock(notification_type='email', body='text') + assert NotificationSerializer().get_body(obj) == 'text' + + +class TestJobHostSummarySerializerPaths: + def test_get_related_includes_host(self): + serializer = JobHostSummarySerializer() + serializer.reverse = mock.Mock(side_effect=lambda *a, **k: '/x/') + obj = mock.Mock(job=mock.Mock(pk=1), host=mock.Mock(pk=2)) + with mock.patch('awx.api.serializers.BaseSerializer.get_related', return_value={}): + related = serializer.get_related(obj) + assert 'job' in related + assert 'host' in related + + def test_summary_fields_adds_job_template(self): + serializer = JobHostSummarySerializer() + obj = mock.Mock() + obj.job.job_template.id = 11 + obj.job.job_template.name = 'jt' + with mock.patch('awx.api.serializers.BaseSerializer.get_summary_fields', return_value={'job': {}}): + summary = serializer.get_summary_fields(obj) + assert summary['job']['job_template_id'] == 11 + assert summary['job']['job_template_name'] == 'jt' + + def test_summary_fields_tolerates_missing_job(self): + serializer = JobHostSummarySerializer() + obj = mock.Mock() + with mock.patch('awx.api.serializers.BaseSerializer.get_summary_fields', return_value={}): + assert serializer.get_summary_fields(obj) == {} + + +class TestWorkflowJobLaunchSerializerPaths: + def test_survey_enabled(self): + serializer = WorkflowJobLaunchSerializer() + assert serializer.get_survey_enabled(None) is False + obj = mock.Mock(survey_enabled=True, survey_spec={'spec': []}) + assert serializer.get_survey_enabled(obj) is True + + def test_workflow_job_template_data(self): + obj = _mock(name='wf', id=5, description='d') + assert WorkflowJobLaunchSerializer().get_workflow_job_template_data(obj) == {'name': 'wf', 'id': 5, 'description': 'd'} + + def test_defaults_inventory_and_labels(self): + mapping = {'inventory': None, 'labels': None, 'limit': None} + obj = mock.Mock() + obj.inventory.name = 'inv' + obj.inventory.pk = 8 + obj.limit = 'webservers' + label = _mock(id=1, name='prod') + obj.labels.all.return_value = [label] + with mock.patch.object(WorkflowJobTemplate, 'get_ask_mapping', return_value=mapping): + defaults = WorkflowJobLaunchSerializer().get_defaults(obj) + assert defaults['inventory'] == {'name': 'inv', 'id': 8} + assert defaults['labels'] == [{'id': 1, 'name': 'prod'}] + assert defaults['limit'] == 'webservers' + + +class TestActivityStreamSerializerPaths: + def test_get_changes(self): + serializer = ActivityStreamSerializer() + assert serializer.get_changes(None) == {} + assert serializer.get_changes(mock.Mock(changes='{"a": 1}')) == {'a': 1} + assert serializer.get_changes(mock.Mock(changes='not-json')) == {} + + def test_object_association(self): + serializer = ActivityStreamSerializer() + assert serializer.get_object_association(mock.Mock(object_relationship_type='')) == '' + assert serializer.get_object_association(mock.Mock(object_relationship_type='awx.main.models.inventory.Inventory.admin_role')) == 'role' + rel = 'awx.main.models.organization.Organization_notification_templates_success' + assert 'notification' in serializer.get_object_association(mock.Mock(object_relationship_type=rel)) + + def test_object_type_role(self): + serializer = ActivityStreamSerializer() + assert serializer.get_object_type(mock.Mock(object_relationship_type='')) == '' + rel = 'awx.main.models.inventory.Inventory.admin_role' + assert serializer.get_object_type(mock.Mock(object_relationship_type=rel)) == 'inventory' + + +class TestProjectUpdateEventSanitization: + def test_sanitizes_git_event_data(self): + serializer = ProjectUpdateEventSerializer() + obj = mock.Mock(event_data={'task_action': 'git', 'msg': 'ok'}) + assert serializer.get_event_data(obj)['task_action'] == 'git' + + def test_returns_empty_on_sanitize_failure(self): + serializer = ProjectUpdateEventSerializer() + obj = mock.Mock(event_data={'task_action': 'svn'}) + with mock.patch('awx.api.serializers.json.dumps', side_effect=ValueError('boom')): + assert serializer.get_event_data(obj) == {} + + def test_passthrough_for_other_actions(self): + serializer = ProjectUpdateEventSerializer() + obj = mock.Mock(event_data={'task_action': 'debug', 'msg': 'hi'}) + assert serializer.get_event_data(obj) == obj.event_data + + +class TestUnifiedJobSummaryFields: + def test_spawned_by_workflow_copies_job_summary(self): + serializer = UnifiedJobSerializer() + workflow_job = _mock(id=10, name='wf', description='', status='successful', failed=False, elapsed=1.5, type='workflow_job', canceled_on=None) + obj = mock.Mock(spawned_by_workflow=True, ancestor_job=None) + obj.unified_job_node.workflow_job = workflow_job + with mock.patch('awx.api.serializers.BaseSerializer.get_summary_fields', return_value={}): + summary = serializer.get_summary_fields(obj) + assert summary['source_workflow_job']['id'] == 10 + assert summary['source_workflow_job']['name'] == 'wf' + + +def test_unified_job_elapsed_for_running_job(): + serializer = UnifiedJobSerializer() + started = mock.Mock() + td = mock.Mock(microseconds=0, seconds=5, days=0) + obj = mock.Mock(pk=1, started=started, finished=None, job_explanation='') + with mock.patch.object(UnifiedJobSerializer, 'get_sub_serializer', return_value=None): + with mock.patch('awx.api.serializers.BaseSerializer.to_representation', return_value={'elapsed': 0, 'job_explanation': ''}): + with mock.patch('awx.api.serializers.now', return_value=mock.Mock(__sub__=lambda self, other: td)): + ret = serializer.to_representation(obj) + assert ret['elapsed'] == 5.0