<commit_before>#!/usr/bin/env python

"""
@package ion.services.mi.test.test_sbe37_driver
@file ion/services/mi/test_sbe37_driver.py
@author <NAME>
@brief Test cases for SBE37Driver
"""

__author__ = '<NAME>'
__license__ = 'Apache 2.0'

from gevent import monkey; monkey.patch_all()

import time
import unittest
import logging
from subprocess import Popen
import os
import signal

from nose.plugins.attrib import attr

from pyon.util.unit_test import PyonTestCase

from ion.services.mi.zmq_driver_client import ZmqDriverClient
from ion.services.mi.zmq_driver_process import ZmqDriverProcess
from ion.services.mi.drivers.sbe37_driver import SBE37Channel
from ion.services.mi.drivers.sbe37_driver import SBE37Command
import ion.services.mi.mi_logger

mi_logger = logging.getLogger('mi_logger')

#from pyon.public import log

# Make tests verbose and provide stdout
# bin/nosetests -s -v ion/services/mi/drivers/test/test_sbe37_driver.py

@attr('UNIT', group='mi')
class TestSBE37Driver(PyonTestCase):    
    """
    Unit tests for ZMQ driver process.
    """
    
    def setUp(self):
        """
        Setup test cases.
        """
        # Zmq parameters used by driver process and client.
        self.host = 'localhost'
        self.cmd_port = 5556
        self.evt_port = 5557
        
        # Driver module parameters.
        self.dvr_mod = 'ion.services.mi.drivers.sbe37_driver'
        self.dvr_cls = 'SBE37Driver'

        #
        self.server_addr = 'localhost'

        # Add cleanup handler functions.
        # self.addCleanup()
        
    def test_config(self):
        """
        Test driver configure.
        """
        #def wait_on_child(signum=None, frame=None):
        #    retval = os.wait()
        #signal.signal(signal.SIGCHLD, wait_on_child)


        driver_process = ZmqDriverProcess.launch_process(self.cmd_port,
            self.evt_port, self.dvr_mod,  self.dvr_cls)
        
        
        driver_client = ZmqDriverClient(self.server_addr, self.cmd_port,
                                        self.evt_port)
        driver_client.start_messaging()
        time.sleep(3)
        config = {
            'method':'ethernet',
            'device_addr': '192.168.127.12',
            'device_port': 4001,
            'server_addr': 'localhost',
            'server_port': 8888            
        }
        configs = {SBE37Channel.CTD:config}
        reply = driver_client.cmd_dvr('configure', configs)
 
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('connect', [SBE37Channel.CTD])
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('execute', [SBE37Channel.CTD], [SBE37Command.ACQUIRE_SAMPLE])
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('execute', [SBE37Channel.CTD], [SBE37Command.START_AUTO_SAMPLING])
        time.sleep(20)
        
        reply = driver_client.cmd_dvr('execute', [SBE37Channel.CTD], [SBE37Command.STOP_AUTO_SAMPLING])
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('disconnect', [SBE37Channel.CTD])
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('initialize', [SBE37Channel.CTD])
        time.sleep(2)
        
        driver_client.done()
        driver_process.wait()


    <commit_msg>fixed bug in test case<commit_after>#!/usr/bin/env python

"""
@package ion.services.mi.test.test_sbe37_driver
@file ion/services/mi/test_sbe37_driver.py
@author <NAME>
@brief Test cases for SBE37Driver
"""

__author__ = '<NAME>'
__license__ = 'Apache 2.0'

from gevent import monkey; monkey.patch_all()

import time
import unittest
import logging
from subprocess import Popen
import os
import signal

from nose.plugins.attrib import attr

from pyon.util.unit_test import PyonTestCase

from ion.services.mi.zmq_driver_client import ZmqDriverClient
from ion.services.mi.zmq_driver_process import ZmqDriverProcess
from ion.services.mi.drivers.sbe37_driver import SBE37Channel
from ion.services.mi.drivers.sbe37_driver import SBE37Command
import ion.services.mi.mi_logger

mi_logger = logging.getLogger('mi_logger')

#from pyon.public import log

# Make tests verbose and provide stdout
# bin/nosetests -s -v ion/services/mi/drivers/test/test_sbe37_driver.py

@attr('UNIT', group='mi')
class TestSBE37Driver(PyonTestCase):    
    """
    Unit tests for ZMQ driver process.
    """
    
    def setUp(self):
        """
        Setup test cases.
        """
        # Zmq parameters used by driver process and client.
        self.server_addr = 'localhost'
        self.cmd_port = 5556
        self.evt_port = 5557
        
        # Driver module parameters.
        self.dvr_mod = 'ion.services.mi.drivers.sbe37_driver'
        self.dvr_cls = 'SBE37Driver'

        # Comms config.
        self.comms_config = {
            'method':'ethernet',
            'device_addr': '192.168.127.12',
            'device_port': 4001,
            'server_addr': 'localhost',
            'server_port': 8888            
        }

        # Add cleanup handler functions.
        # self.addCleanup()
        
    def test_config(self):
        """
        Test driver configure.
        """
        #def wait_on_child(signum=None, frame=None):
        #    retval = os.wait()
        #signal.signal(signal.SIGCHLD, wait_on_child)


        driver_process = ZmqDriverProcess.launch_process(self.cmd_port,
            self.evt_port, self.dvr_mod,  self.dvr_cls)
        
        
        driver_client = ZmqDriverClient(self.server_addr, self.cmd_port,
                                        self.evt_port)
        driver_client.start_messaging()
        time.sleep(3)

        configs = {SBE37Channel.CTD:self.comms_config}
        reply = driver_client.cmd_dvr('configure', configs)
 
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('connect', [SBE37Channel.CTD])
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('execute', [SBE37Channel.CTD], [SBE37Command.ACQUIRE_SAMPLE])
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('execute', [SBE37Channel.CTD], [SBE37Command.START_AUTO_SAMPLING])
        time.sleep(20)
        
        reply = driver_client.cmd_dvr('execute', [SBE37Channel.CTD], [SBE37Command.STOP_AUTO_SAMPLING])
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('disconnect', [SBE37Channel.CTD])
        time.sleep(2)
        
        reply = driver_client.cmd_dvr('initialize', [SBE37Channel.CTD])
        time.sleep(2)
        
        driver_client.done()
        driver_process.wait()


    
{'fraction_non_alphanumeric': 0.10382262996941896, 'fraction_numerical': 0.024464831804281346, 'mean_word_length': 2.422815279958137, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5698309', 'n_tokens_mistral': 2213, 'n_tokens_neox': 2099, 'n_words': 384}