# Path: matl_online/tasks.py
# class OctaveTask(Task):
#     """Custom Task type for interacting with octave."""
# 
#     abstract = True
#     _octave = None
#     _tempfolder = None
#     session_id = None
#     _handler = None
# 
#     def __init__(self, *args, **kwargs):
#         """Initialize task."""
#         global octave
#         super(OctaveTask, self).__init__(*args, **kwargs)
# 
#     @property
#     def octave(self):
#         """Dependent property which automatically spawns octave if needed."""
#         if self._octave is None:
#             global octave
#             self._octave = octave
# 
#         return self._octave
# 
#     @property
#     def handler(self):
#         """Dependent property that creates an OutputHandler instance."""
#         if self._handler is None:
#             self._handler = OutputHandler(self)
# 
#         return self._handler
# 
#     @property
#     def folder(self):
#         """Dependent property that creates a session-specific folder if needed."""
#         if self._tempfolder is None:
#             # Generate the temporary folder
#             if self.session_id:
#                 self._tempfolder = os.path.join(tempfile.gettempdir(),
#                                                 self.session_id)
#             else:
#                 self._tempfolder = tempfile.mkdtemp()
# 
#         # Make directory if it doesn't exist
#         if not os.path.isdir(self._tempfolder):
#             os.makedirs(self._tempfolder)
# 
#         return self._tempfolder
# 
#     def emit(self, *args, **kwargs):
#         """Send an event to any listening clients."""
#         socket.emit(*args, room=self.session_id, **kwargs)
# 
#     def on_term(self):
#         """Clean up after termination event."""
#         # Restart octave so we're ready to go with future calls
#         self.octave.restart()
#         _initialize_process()
# 
#     def on_success(self, *args, **kwargs):
#         """Send a completion messages upon successful completion."""
#         self.emit('complete', {'success': True, 'message': ''})
# 
#     def on_kill(self):
#         """Clean up after a task is killed.
# 
#         This is a hard kill by celery so this worker will be destroyed. No
#         need to restart octave
#         """
#         self.octave._engine.repl.terminate()
#         self.on_failure()
# 
#     def send_results(self):
#         """Local forwarder for all send events."""
#         return self.handler.send()
# 
#     def after_return(self, *args, **kwargs):
#         """Fire after task completion."""
#         self.handler.clear()
# 
#         if os.path.isdir(self.folder):
#             shutil.rmtree(self.folder)
# 
#     def on_failure(self, *args, **kwargs):
#         """Send a message if the task failed for any reason."""
#         self.send_results()
#         self.emit('complete', {'success': False})
# 
# @celery.task(base=OctaveTask, bind=True)
# def matl_task(self, *args, **kwargs):
#     """Celery task for processing a MATL command and returning the result."""
#     self.session_id = kwargs.pop('session', '')
#     self.handler.clear()
# 
#     is_test = config.ENV == 'test'
# 
#     try:
#         matl(self.octave, *args, folder=self.folder,
#              line_handler=self.handler.process_message, **kwargs)
#         result = self.send_results()
# 
#     # In the case of an interrupt (either through a time limit or a
#     # revoke() event, we will still clean things up
#     except (KeyboardInterrupt, SystemExit):
#         self.handler.process_message('[STDERR]Job cancelled')
#         self.on_kill()
#         raise
#     except SoftTimeLimitExceeded:
#         # Propagate the term event up the chain to actually kill the worker
#         self.handler.process_message('[STDERR]Operation timed out')
#         self.on_term()
# 
#         if is_test:
#             self.on_failure()
#             self.after_return()
# 
#         raise
# 
#     if is_test:
#         self.on_success()
#         self.after_return()
# 
#     return result
# 
# Path: tests/helpers.py
# def session_id_for_client(client: SocketIOTestClient, namespace: str = '/') -> Optional[str]:
#     """Retrieve the session id for a SocketIOTestClient."""
#     return client.socketio.server.manager.sid_from_eio_sid(client.eio_sid, namespace)


import logging
import os
import pytest
from celery.exceptions import SoftTimeLimitExceeded
from matl_online.tasks import OctaveTask, matl_task
from .helpers import session_id_for_client
"""Unit tests for basic celery task functionality."""




def prepare_folder_testcase(mocker, moctave, tmpdir):
    """Create the necessary mocks."""
    mocker.patch('matl_online.tasks.Task')

    mktmp = mocker.patch('matl_online.tasks.tempfile.mkdtemp')
    mktmp.return_value = tmpdir.strpath

    gettmp = mocker.patch('matl_online.tasks.tempfile.gettempdir')
    gettmp.return_value = tmpdir.strpath


class TestOctaveTask:
    """Series of tests for the OctaveTask class."""

    def test_octave_property(self, mocker, moctave, logger):
        """Make sure that an instance is created only when requested."""
        mocker.patch('matl_online.tasks.Task')

        logger.setLevel(logging.ERROR)
        moctave.logger = logger


# Path: matl_online/tasks.py
# class OctaveTask(Task):
#     """Custom Task type for interacting with octave."""
# 
#     abstract = True
#     _octave = None
#     _tempfolder = None
#     session_id = None
#     _handler = None
# 
#     def __init__(self, *args, **kwargs):
#         """Initialize task."""
#         global octave
#         super(OctaveTask, self).__init__(*args, **kwargs)
# 
#     @property
#     def octave(self):
#         """Dependent property which automatically spawns octave if needed."""
#         if self._octave is None:
#             global octave
#             self._octave = octave
# 
#         return self._octave
# 
#     @property
#     def handler(self):
#         """Dependent property that creates an OutputHandler instance."""
#         if self._handler is None:
#             self._handler = OutputHandler(self)
# 
#         return self._handler
# 
#     @property
#     def folder(self):
#         """Dependent property that creates a session-specific folder if needed."""
#         if self._tempfolder is None:
#             # Generate the temporary folder
#             if self.session_id:
#                 self._tempfolder = os.path.join(tempfile.gettempdir(),
#                                                 self.session_id)
#             else:
#                 self._tempfolder = tempfile.mkdtemp()
# 
#         # Make directory if it doesn't exist
#         if not os.path.isdir(self._tempfolder):
#             os.makedirs(self._tempfolder)
# 
#         return self._tempfolder
# 
#     def emit(self, *args, **kwargs):
#         """Send an event to any listening clients."""
#         socket.emit(*args, room=self.session_id, **kwargs)
# 
#     def on_term(self):
#         """Clean up after termination event."""
#         # Restart octave so we're ready to go with future calls
#         self.octave.restart()
#         _initialize_process()
# 
#     def on_success(self, *args, **kwargs):
#         """Send a completion messages upon successful completion."""
#         self.emit('complete', {'success': True, 'message': ''})
# 
#     def on_kill(self):
#         """Clean up after a task is killed.
# 
#         This is a hard kill by celery so this worker will be destroyed. No
#         need to restart octave
#         """
#         self.octave._engine.repl.terminate()
#         self.on_failure()
# 
#     def send_results(self):
#         """Local forwarder for all send events."""
#         return self.handler.send()
# 
#     def after_return(self, *args, **kwargs):
#         """Fire after task completion."""
#         self.handler.clear()
# 
#         if os.path.isdir(self.folder):
#             shutil.rmtree(self.folder)
# 
#     def on_failure(self, *args, **kwargs):
#         """Send a message if the task failed for any reason."""
#         self.send_results()
#         self.emit('complete', {'success': False})
# 
# @celery.task(base=OctaveTask, bind=True)
# def matl_task(self, *args, **kwargs):
#     """Celery task for processing a MATL command and returning the result."""
#     self.session_id = kwargs.pop('session', '')
#     self.handler.clear()
# 
#     is_test = config.ENV == 'test'
# 
#     try:
#         matl(self.octave, *args, folder=self.folder,
#              line_handler=self.handler.process_message, **kwargs)
#         result = self.send_results()
# 
#     # In the case of an interrupt (either through a time limit or a
#     # revoke() event, we will still clean things up
#     except (KeyboardInterrupt, SystemExit):
#         self.handler.process_message('[STDERR]Job cancelled')
#         self.on_kill()
#         raise
#     except SoftTimeLimitExceeded:
#         # Propagate the term event up the chain to actually kill the worker
#         self.handler.process_message('[STDERR]Operation timed out')
#         self.on_term()
# 
#         if is_test:
#             self.on_failure()
#             self.after_return()
# 
#         raise
# 
#     if is_test:
#         self.on_success()
#         self.after_return()
# 
#     return result
# 
# Path: tests/helpers.py
# def session_id_for_client(client: SocketIOTestClient, namespace: str = '/') -> Optional[str]:
#     """Retrieve the session id for a SocketIOTestClient."""
#     return client.socketio.server.manager.sid_from_eio_sid(client.eio_sid, namespace)


# Path: tests/test_tasks.py
import logging
import os
import pytest
from celery.exceptions import SoftTimeLimitExceeded
from matl_online.tasks import OctaveTask, matl_task
from .helpers import session_id_for_client

"""Unit tests for basic celery task functionality."""




def prepare_folder_testcase(mocker, moctave, tmpdir):
    """Create the necessary mocks."""
    mocker.patch('matl_online.tasks.Task')

    mktmp = mocker.patch('matl_online.tasks.tempfile.mkdtemp')
    mktmp.return_value = tmpdir.strpath

    gettmp = mocker.patch('matl_online.tasks.tempfile.gettempdir')
    gettmp.return_value = tmpdir.strpath


class TestOctaveTask:
    """Series of tests for the OctaveTask class."""

    def test_octave_property(self, mocker, moctave, logger):
        """Make sure that an instance is created only when requested."""
        mocker.patch('matl_online.tasks.Task')

        logger.setLevel(logging.ERROR)
        moctave.logger = logger


        task = OctaveTask()