External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-17 20:51:21 -03:30
Code Issues Packages Projects Releases Wiki Activity

Disable socketio notifications for unit tests.

Browse Source
This commit is contained in:
Chris Church 2014-04-19 02:12:33 -04:00
parent 2177d6feab
commit 29349d0a5a
3 changed files with 12 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
awx/main/tests/base.py
View File

@ -66,6 +66,8 @@ class BaseTestMixin(object):
self._temp_paths.append(callback_queue_path)
settings.CALLBACK_QUEUE_PORT = 'ipc://%s' % callback_queue_path
settings.TASK_COMMAND_PORT = 'ipc:///tmp/task_command_receiver_%d.ipc' % callback_port
# Disable socket notifications for unit tests.
settings.SOCKETIO_NOTIFICATION_PORT = None
# Make temp job status directory for unit tests.
job_status_dir = tempfile.mkdtemp()
self._temp_paths.append(job_status_dir)

3
awx/main/tests/commands.py
View File

@ -323,7 +323,8 @@ class CleanupJobsTest(BaseCommandMixin, BaseLiveServerTest):
self.project = None
self.credential = None
settings.INTERNAL_API_URL = self.live_server_url
self.start_queue(settings.CALLBACK_CONSUMER_PORT, settings.CALLBACK_QUEUE_PORT)
if settings.CALLBACK_CONSUMER_PORT:
self.start_queue(settings.CALLBACK_CONSUMER_PORT, settings.CALLBACK_QUEUE_PORT)
def tearDown(self):
super(CleanupJobsTest, self).tearDown()

17
awx/main/utils.py
View File

@ -19,9 +19,6 @@ from Crypto.Cipher import AES
# ZeroMQ
import zmq
# Django
from django.conf import settings
__all__ = ['get_object_or_400', 'get_object_or_403', 'camelcase_to_underscore',
'get_ansible_version', 'get_awx_version', 'update_scm_url',
'get_type_for_model', 'get_model_for_type']
@ -345,9 +342,11 @@ def get_system_task_capacity():
return 50 + ((int(total_mem_value) / 1024) - 2) * 75
def emit_websocket_notification(endpoint, event, payload):
emit_context = zmq.Context()
emit_socket = emit_context.socket(zmq.PUSH)
emit_socket.connect(settings.SOCKETIO_NOTIFICATION_PORT)
payload['event'] = event
payload['endpoint'] = endpoint
emit_socket.send_json(payload);
from django.conf import settings
if getattr(settings, 'SOCKETIO_NOTIFICATION_PORT', None):
emit_context = zmq.Context()
emit_socket = emit_context.socket(zmq.PUSH)
emit_socket.connect(settings.SOCKETIO_NOTIFICATION_PORT)
payload['event'] = event
payload['endpoint'] = endpoint
emit_socket.send_json(payload);