diff --git a/awx/main/tests/functional/test_bulk.py b/awx/main/tests/functional/test_bulk.py
index 2c59c76db5..0f33df15f8 100644
--- a/awx/main/tests/functional/test_bulk.py
+++ b/awx/main/tests/functional/test_bulk.py
@@ -634,7 +634,7 @@ def test_bulk_host_create_performance_large_inventory(organization, inventory, p
@pytest.mark.django_db
def test_bulk_host_create_logs_bypass_for_unsafe_description_when_enforcement_disabled(organization, inventory, post, user, caplog):
- """Bulk host create skips post_save; audit before bulk_create must log Tier 2 violations."""
+ """Bulk host create skips post_save; serializer validate() or bulk audit must log Tier 2 violations."""
import logging
inventory.organization = organization
@@ -642,41 +642,43 @@ def test_bulk_host_create_logs_bypass_for_unsafe_description_when_enforcement_di
organization.member_role.members.add(inv_admin)
inventory.admin_role.members.add(inv_admin)
+ caplog.set_level(logging.DEBUG)
+ caplog.set_level(logging.WARNING)
with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=False):
- with caplog.at_level(logging.WARNING, logger='ansible_base.lib.utils.validation_signals'):
- post(
- reverse('api:bulk_host_create'),
- {
- 'inventory': inventory.id,
- 'hosts': [{'name': f'bulk-host-bypass-{uuid4().hex[:8]}', 'description': ''}],
- },
- inv_admin,
- expect=201,
- )
- assert 'ORM bypass (bulk_create)' in caplog.text
+ post(
+ reverse('api:bulk_host_create'),
+ {
+ 'inventory': inventory.id,
+ 'hosts': [{'name': f'bulk-host-bypass-{uuid4().hex[:8]}', 'description': ''}],
+ },
+ inv_admin,
+ expect=201,
+ )
assert 'description' in caplog.text
assert 'main.Host' in caplog.text
+ assert 'ORM bypass (bulk_create)' in caplog.text or 'Validation rejected' in caplog.text
@pytest.mark.django_db
def test_bulk_job_launch_logs_bypass_for_unsafe_limit_when_enforcement_disabled(organization, inventory, project, post, user, caplog):
- """Bulk workflow nodes skip post_save; audit before bulk_create must log Tier 2 violations."""
+ """Bulk workflow nodes skip post_save; serializer validate() or bulk audit must log Tier 2 violations."""
import logging
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)
+ caplog.set_level(logging.DEBUG)
+ caplog.set_level(logging.WARNING)
with mock.patch('ansible_base.lib.serializers.mixins.get_setting', return_value=False):
- with caplog.at_level(logging.WARNING, logger='ansible_base.lib.utils.validation_signals'):
- post(
- reverse('api:bulk_job_launch'),
- {'name': 'Bulk Job Launch', 'jobs': [{'unified_job_template': jt.id, 'limit': ''}]},
- normal_user,
- expect=201,
- )
- assert 'ORM bypass (bulk_create)' in caplog.text
+ post(
+ reverse('api:bulk_job_launch'),
+ {'name': 'Bulk Job Launch', 'jobs': [{'unified_job_template': jt.id, 'limit': ''}]},
+ normal_user,
+ expect=201,
+ )
assert 'limit' in caplog.text
+ assert 'ORM bypass (bulk_create)' in caplog.text or 'Validation rejected' in caplog.text
@pytest.mark.django_db
diff --git a/awx/main/tests/unit/utils/test_validation_bypass_observability.py b/awx/main/tests/unit/utils/test_validation_bypass_observability.py
index b01f69c1fd..5510d3ab4d 100644
--- a/awx/main/tests/unit/utils/test_validation_bypass_observability.py
+++ b/awx/main/tests/unit/utils/test_validation_bypass_observability.py
@@ -17,6 +17,13 @@ from awx.main.utils.validation_bypass_observability import (
LOGGER = 'ansible_base.lib.utils.validation_signals'
+@pytest.fixture
+def capture_validation_signal_logs(caplog):
+ """Match DAB test_app caplog setup (pytest + xdist safe)."""
+ caplog.set_level(logging.DEBUG)
+ caplog.set_level(logging.WARNING, logger=LOGGER)
+
+
@pytest.mark.django_db
def test_configure_validation_bypass_observability_wires_dab(mocker):
mocker.patch('awx.main.utils.validation_bypass_observability.register_validation_signals')
@@ -38,6 +45,7 @@ def test_configure_validation_bypass_observability_wires_dab(mocker):
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_audit_bulk_model_instances_skips_host_description_when_serializer_already_logged(caplog):
from ansible_base.lib.utils.validation_signals import (
get_validation_context_token,
@@ -51,26 +59,26 @@ def test_audit_bulk_model_instances_skips_host_description_when_serializer_alrea
register_serializer_validation_rejection('main.Host', 'description')
token = get_validation_context_token()
try:
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- audit_bulk_model_instances([host], operation='bulk_create')
+ audit_bulk_model_instances([host], operation='bulk_create')
finally:
reset_validation_context(token)
assert 'ORM bypass' not in caplog.text
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_audit_bulk_model_instances_logs_host_description_on_bulk_create(caplog):
org = Organization.objects.create(name='org-bulk-host-audit')
inv = Inventory.objects.create(name='inv-bulk-host-audit', organization=org)
host = Host(name='host-bulk', description='', inventory=inv)
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- audit_bulk_model_instances([host], operation='bulk_create')
+ audit_bulk_model_instances([host], operation='bulk_create')
assert 'ORM bypass (bulk_create)' in caplog.text
assert 'description' in caplog.text
assert 'main.Host' in caplog.text
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_audit_workflow_job_nodes_skips_when_serializer_already_logged(caplog):
from ansible_base.lib.utils.validation_signals import register_serializer_validation_rejection
@@ -78,74 +86,86 @@ def test_audit_workflow_job_nodes_skips_when_serializer_already_logged(caplog):
node = WorkflowJobNode(workflow_job=wfj, identifier='wf-node-1')
node.limit = ''
register_serializer_validation_rejection('main.WorkflowJobNode', 'limit')
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- audit_workflow_job_nodes_for_bulk_create([node])
+ audit_workflow_job_nodes_for_bulk_create([node])
assert 'ORM bypass' not in caplog.text
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_audit_workflow_job_nodes_logs_limit_pseudo_field(caplog):
wfj = WorkflowJob.objects.create(name='wf-audit-limit')
node = WorkflowJobNode(workflow_job=wfj, identifier='wf-node-1')
node.limit = ''
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- audit_workflow_job_nodes_for_bulk_create([node])
+ audit_workflow_job_nodes_for_bulk_create([node])
assert 'ORM bypass (bulk_create)' in caplog.text
assert 'limit' in caplog.text
assert 'WorkflowJobNode' in caplog.text
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_audit_workflow_job_nodes_no_op_for_empty_list(caplog):
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- audit_workflow_job_nodes_for_bulk_create([])
+ audit_workflow_job_nodes_for_bulk_create([])
assert 'ORM bypass' not in caplog.text
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_audit_bulk_model_instances_logs_registered_text_field_on_bulk_update(caplog):
job = Job.objects.create(job_explanation='')
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- audit_bulk_model_instances([job], operation='bulk_update', update_fields=['job_explanation'])
+ audit_bulk_model_instances([job], operation='bulk_update', update_fields=['job_explanation'])
assert 'ORM bypass (bulk_update)' in caplog.text
assert 'job_explanation' in caplog.text
@pytest.mark.django_db
-def test_audit_bulk_model_instances_skips_unregistered_workflow_job_on_bulk_update(caplog):
- wj = WorkflowJob.objects.create(name='wf-job', job_explanation='')
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- audit_bulk_model_instances([wj], operation='bulk_update', update_fields=['job_explanation'])
+@pytest.mark.usefixtures('capture_validation_signal_logs')
+def test_audit_bulk_model_instances_skips_unregistered_instance_on_bulk_update(caplog):
+ from ansible_base.resource_registry.models import Resource
+
+ unregistered = Resource.__new__(Resource)
+ audit_bulk_model_instances([unregistered], operation='bulk_update', update_fields=['name'])
assert 'ORM bypass (bulk_update)' not in caplog.text
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
+def test_audit_bulk_model_instances_logs_workflow_job_job_explanation_on_bulk_update(caplog):
+ """WorkflowJob is registered via BulkJobLaunchSerializer (PromptFieldCleanTextMixin)."""
+ wj = WorkflowJob.objects.create(name='wf-job', job_explanation='')
+ audit_bulk_model_instances([wj], operation='bulk_update', update_fields=['job_explanation'])
+ assert 'ORM bypass (bulk_update)' in caplog.text
+ assert 'job_explanation' in caplog.text
+ assert 'main.WorkflowJob' in caplog.text
+
+
+@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_audit_bulk_model_instances_bulk_update_no_op_when_update_fields_empty(caplog):
job = Job.objects.create(job_explanation='')
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- audit_bulk_model_instances([job], operation='bulk_update', update_fields=[])
+ audit_bulk_model_instances([job], operation='bulk_update', update_fields=[])
assert 'ORM bypass (bulk_update)' not in caplog.text
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_bulk_update_sorted_by_id_skips_audit_when_fields_are_not_text(caplog):
org = Organization.objects.create(name='org-bulk-update-audit')
inv = Inventory.objects.create(name='inv-bulk-update-audit', organization=org)
host = Host.objects.create(name='host-1', description='', inventory=inv)
host.ansible_facts = {'k': 'v'}
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- bulk_update_sorted_by_id(Host, [host], fields=['ansible_facts'])
+ bulk_update_sorted_by_id(Host, [host], fields=['ansible_facts'])
assert 'ORM bypass (bulk_update)' not in caplog.text
@pytest.mark.django_db
+@pytest.mark.usefixtures('capture_validation_signal_logs')
def test_bulk_update_sorted_by_id_audits_text_fields_in_fields_list(caplog):
org = Organization.objects.create(name='org-bulk-update-desc')
inv = Inventory.objects.create(name='inv-bulk-update-desc', organization=org)
host = Host.objects.create(name='host-desc', description='clean', inventory=inv)
host.description = ''
- with caplog.at_level(logging.WARNING, logger=LOGGER):
- bulk_update_sorted_by_id(Host, [host], fields=['description'])
+ bulk_update_sorted_by_id(Host, [host], fields=['description'])
assert 'ORM bypass (bulk_update)' in caplog.text
assert 'description' in caplog.text
diff --git a/awx/main/utils/validation_bypass_observability.py b/awx/main/utils/validation_bypass_observability.py
index 7c4c83a1f2..0e65bef47c 100644
--- a/awx/main/utils/validation_bypass_observability.py
+++ b/awx/main/utils/validation_bypass_observability.py
@@ -87,7 +87,7 @@ def audit_workflow_job_nodes_for_bulk_create(nodes: Iterable[Model]) -> None:
continue
violation = _validate_field(field_name, value, name_fields)
if violation:
- tier, reason = violation
+ _tier, reason = violation
if caller_info is None:
caller_info = _get_caller_info()
- log_orm_bypass_violation('bulk_create', field_name, resource_type, tier, caller_info, reason)
+ log_orm_bypass_violation('bulk_create', field_name, resource_type, caller_info, reason)