class Test(unittest.TestCase):

    def setUp(self):
        from testfixtures import LogCapture
        self.log_capture = LogCapture()
        self.log_group  = "phidgeter.ir_temperature"
        self.lvl = "DEBUG"

    def tearDown(self):
        self.log_capture.uninstall()

    def test_log_captures(self):
        # verification of log matching functionality
        from logging import getLogger
        getLogger().info("a message")
        self.log_capture.check(("root", "INFO", "a message"))

    def test_sensor_is_available(self):
        ir_temp = IRSensor()
        assert ir_temp.open_phidget() == True
        assert ir_temp.get_temperature() >= 0.0
        assert ir_temp.close_phidget() == True

    def test_sensor_is_room_temperature(self):
        ir_temp = IRSensor()
        assert ir_temp.open_phidget() == True
        print ir_temp.get_temperature()
        assert ir_temp.get_temperature() >= 20.0
        assert ir_temp.close_phidget() == True

    def test_uninstall_all(self):
        before_handlers_root = root.handlers[:]
        before_handlers_child = child.handlers[:]

        l1 = LogCapture()
        l2 = LogCapture('one.child')

        # We can see that the LogCaptures have changed the
        # handlers, removing existing ones and installing
        # their own:

        assert len(root.handlers) == 1
        assert root.handlers != before_handlers_root
        assert len(child.handlers) == 1
        assert child.handlers != before_handlers_child

        # Now we show the function in action:

        LogCapture.uninstall_all()

        # ...and we can see the handlers are back as
        # they were beefore:

        assert before_handlers_root == root.handlers
        assert before_handlers_child == child.handlers

    def test_clear_global_state(self):
        from logging import _handlers, _handlerList

        capture = LogCapture()
        capture.uninstall()
        self.assertFalse(capture in _handlers)
        self.assertFalse(capture in _handlerList)

def connectTo():
    gearkey = "qnlgzsPUUxYeyQP"
    gearsecret = "1euJPvxybllEPQZzq2u9wpRJXDbjM7"
    appid = "testNo3"    
    client.create(gearkey, gearsecret, appid, {'debugmode': "True"})
    
    def on_connected():
        print("connect")
    def on_closed():
        print("close")  
    def on_rejected():
        print("reject")     
    def on_error():
        print("error")  
    def on_message():
        print("message")  
    def on_present():
        print("present")
    def on_absent():
        print("absent") 
    client.on_connect = on_connected
    client.on_error = on_error
    client.on_present = on_present
    client.on_absent = on_absent
    client.on_rejected = on_rejected
    client.on_closed = on_closed
    client.on_message = on_message
    logs = LogCapture()
    client.connect()
    print(logs)
    logs.check(('root', 'DEBUG', 'Check stored token.'))

 def test_simple(self):
     root.info('before')
     l = LogCapture()
     root.info('during')
     l.uninstall()
     root.info('after')
     assert str(l) == "root INFO\n  during"

class ExperimentRunningMock(test_integration_mock.GatewayCodeMock):

    """ Create environment for running experiments """

    def setUp(self):
        super(ExperimentRunningMock, self).setUp()
        # config experiment and create folder
        self.g_m._create_user_exp_folders(USER, EXP_ID)
        self.log_error = LogCapture('gateway_code', level=logging.ERROR)

    def tearDown(self):
        super(ExperimentRunningMock, self).tearDown()
        self.g_m._destroy_user_exp_folders(USER, EXP_ID)
        self.log_error.uninstall()

    @staticmethod
    def send_n_cmds(command, num_times, step=0.5):
        """ Send a command multiple times and return array of answers """
        answers = []
        cmd = command.split()
        for _itr in range(0, num_times):  # pylint:disable=unused-variable
            ans = OpenNodeConnection.send_one_command(cmd)
            ans = ' '.join(ans) if ans is not None else None
            answers.append(ans)
            time.sleep(step)
        return answers

class UserSignupTests(TestCase):

    def setUp(self):
        self.handler = LogCapture()
        self.formatter = JsonLogFormatter(logger_name='testpilot.newuser')

        self.username = 'newuserdoe2'
        self.password = 'trustno1'
        self.email = '%[email protected]' % self.username

        self.user = User.objects.create_user(
            username=self.username,
            email=self.email,
            password=self.password)

        UserProfile.objects.filter(user=self.user).delete()

    def tearDown(self):
        self.handler.uninstall()

    def test_newuser_log_event(self):
        """testpilot.newuser log event should be emitted on signup"""
        self.user.is_active = True
        user_signed_up.send(sender=self.user.__class__,
                            request=None,
                            user=self.user)

        self.assertEquals(len(self.handler.records), 1)
        record = self.handler.records[0]

        details = json.loads(self.formatter.format(record))
        self.assertTrue('Fields' in details)

        fields = details['Fields']
        self.assertEqual(fields['uid'], self.user.id)

class TopLevelFormatterTest(unittest.TestCase):

    def setUp(self):
        self.handler = LogCapture()
        self.handler.addFilter(TopLevelFormatter(['test']))

    def test_top_level_logger(self):
        logger = logging.getLogger('test')
        with self.handler as l:
            logger.warning('test log msg')

        l.check(('test', 'WARNING', 'test log msg'))

    def test_children_logger(self):
        logger = logging.getLogger('test.test1')
        with self.handler as l:
            logger.warning('test log msg')

        l.check(('test', 'WARNING', 'test log msg'))

    def test_overlapping_name_logger(self):
        logger = logging.getLogger('test2')
        with self.handler as l:
            logger.warning('test log msg')

        l.check(('test2', 'WARNING', 'test log msg'))

    def test_different_name_logger(self):
        logger = logging.getLogger('different')
        with self.handler as l:
            logger.warning('test log msg')

        l.check(('different', 'WARNING', 'test log msg'))

 def test_simple_manual_install(self):
     l = LogCapture(install=False)
     root.info('before')
     l.install()
     root.info('during')
     l.uninstall()
     root.info('after')
     assert str(l) == "root INFO\n  during"

class TestPickleSerializer(unittest.TestCase):

    layer = ZAMQP_FUNCTIONAL_TESTING

    def setUp(self):
        from testfixtures import LogCapture
        self.l = LogCapture("c.zamqp.tests")

    def tearDown(self):
        self.l.uninstall()

    def _testDeclareQueue(self):
        rabbitctl = self.layer['rabbitctl']
        self.assertIn("my.picklequeue\t0",
                      rabbitctl('list_queues')[0].split("\n"))

    def testDeclareQueue(self):
        runAsyncTest(self._testDeclareQueue)

    def _testDeclareQueueAgain(self):
        rabbitctl = self.layer['rabbitctl']
        self.assertIn("my.picklequeue\t0",
                      rabbitctl('list_queues')[0].split("\n"))

    def testDeclareQueueAgain(self):
        runAsyncTest(self._testDeclareQueueAgain)

    def _testPublishToQueue(self):
        rabbitctl = self.layer['rabbitctl']
        self.assertIn("my.picklequeue\t1",
                      rabbitctl('list_queues')[0].split("\n"))

    def _testPublishToQueueAndConsumeIt(self):
        rabbitctl = self.layer['rabbitctl']
        self.assertIn("my.picklequeue\t0",
                      rabbitctl('list_queues')[0].split("\n"))

    def testPublishToQueueAndConsumeIt(self):
        runAsyncTest(self._testDeclareQueue)

        from zope.component import getUtility
        from collective.zamqp.interfaces import IProducer
        producer = getUtility(IProducer, name="my.picklequeue")
        producer.publish({"key": "value"})

        runAsyncTest(self._testPublishToQueue)
        runAsyncTest(self._testPublishToQueueAndConsumeIt)
        self.l.check(
            ('c.zamqp.tests', 'INFO',
             "<BasicProperties(['delivery_mode=2', "
             "'content_type=application/x-python-serialize'])>"),
            ('c.zamqp.tests', 'INFO', "{'key': 'value'}"),
            ('c.zamqp.tests', 'INFO', "<type 'dict'>")
        )

 def test_multiple_loggers(self):
     l = LogCapture(('one.child','two'))
     root.info('1')
     one.info('2')
     two.info('3')
     child.info('4')
     l.uninstall()
     assert str(l) == (
         "two INFO\n  3\n"
         "one.child INFO\n  4"
     )

 def test_specific_logger(self):
     l = LogCapture('one')
     root.info('1')
     one.info('2')
     two.info('3')
     child.info('4')
     l.uninstall()
     assert str(l) == (
         "one INFO\n  2\n"
         "one.child INFO\n  4"
     )

class TestElfTargetIsCompatibleWithNode(unittest.TestCase):
    """Test elftarget.is_compatible_with_node."""

    def setUp(self):
        self.m3_class = mock.Mock()
        self.m3_class.ELF_TARGET = ('ELFCLASS32', 'EM_ARM')
        self.log = LogCapture('gateway_code', level=logging.DEBUG)

    def tearDown(self):
        self.log.uninstall()

    def test_m3_like_elf_check(self):
        """Test elftarget for an m3 like node."""
        ret = elftarget.is_compatible_with_node(firmware('idle_m3.elf'),
                                                self.m3_class)
        self.assertTrue(ret)
        self.log.check()

        # invalid target
        ret = elftarget.is_compatible_with_node(firmware('node.z1'),
                                                self.m3_class)
        self.assertFalse(ret)
        self.log.check()

        # invalid, not elf file
        ret = elftarget.is_compatible_with_node(
            firmware('wsn430_print_uids.hex'), self.m3_class)
        self.assertFalse(ret)
        self.log.check(('gateway_code', 'WARNING',
                        'Invalid firmware: Not a valid elf file'))

    def test_socat_needs_sigkill(self):
        """Test cases where send_signal must be called multiple times."""
        log = LogCapture('gateway_code', level=logging.WARNING)
        self.addCleanup(log.uninstall)

        only_sigkill = os.path.join(CURRENT_DIR, 'only_sigkill.py')
        only_sigkill = 'python %s' % only_sigkill

        with mock.patch.object(SerialRedirection, 'SOCAT', only_sigkill):
            self.redirect = SerialRedirection(self.tty, self.baud)
            self.redirect.start()
            time.sleep(1)
            self.redirect.stop()

        log.check(('gateway_code', 'WARNING',
                   'SerialRedirection signal: escalading to SIGKILL'))

    def setUp(self):
        super(BaseTestCase, self).setUp()

        self.handler = LogCapture()

        self.username = 'johndoe2'
        self.password = 'trustno1'
        self.email = '%[email protected]' % self.username

        self.user = User.objects.create_user(
            username=self.username,
            email=self.email,
            password=self.password)

        self.users = dict((obj.username, obj) for obj in (
            User.objects.create_user(
                username='experimenttest-%s' % idx,
                email='experimenttest-%[email protected]' % idx,
                password='trustno%s' % idx
            ) for idx in range(0, 5)))

        self.experiments = dict((obj.slug, obj) for (obj, created) in (
            Experiment.objects.get_or_create(
                slug="test-%s" % idx, defaults=dict(
                    order=idx,
                    title="Longer Test Title %s" % idx,
                    short_title="Test %s" % idx,
                    description="This is a test",
                    introduction="<h1>Hello, Test!</h1>",
                    addon_id="addon-%[email protected]" % idx
                )) for idx in range(1, 4)))

 def setUp(self):
   # webtest
   if self.use_cookie:
     cookiejar = cookielib.CookieJar()
   else:
     cookiejar = None
   self.app = TestApp(self.application, domain=self.domain, cookiejar=cookiejar)
   # os.environ
   self.origin_environ = dict()
   if "HTTP_HOST" not in self.environ.viewkeys():
     self.environ["HTTP_HOST"] = "localhost"
   for key, value in self.environ.viewitems():
     self.origin_environ[key], os.environ[key] = os.environ.get(key), value
   # testbed
   self.testbed = testbed.Testbed()
   self.testbed.activate()
   self.testbed.init_datastore_v3_stub(
     consistency_policy=datastore_stub_util.PseudoRandomHRConsistencyPolicy(probability=0),
     root_path=self.root_path,
   )
   self.testbed.init_blobstore_stub()
   self.testbed.init_files_stub()
   self.testbed.init_memcache_stub()
   self.testbed.init_taskqueue_stub(root_path=self.root_path)
   self.testbed.init_urlfetch_stub()
   self.testbed.init_user_stub()
   # logging
   self.log = LogCapture(level=logging.WARNING)
   self.log.install()

 def setUp(self):
     self.dir = TempDirectory()
     self.db_path = self.dir.getpath('test.db')
     self.conn = sqlite3.connect(self.db_path)
     self.conn.execute('create table notes '
                       '(filename varchar, text varchar)')
     self.conn.commit()
     self.log = LogCapture()

def test_query_sanitazion(query_sanitazion):
    app_client = query_sanitazion.app.test_client()
    l = LogCapture()

    url = '/v1.0/greeting'
    response = app_client.post(url, data={'name': 'Jane Doe'})
    # This is ugly. The reason for asserting the logging in this way
    # is that in order to use LogCapture().check, we'd have to assert that
    # a specific sequence of logging has occurred. This is too restricting
    # for future development, and we are really only interested in the fact
    # a single message is logged.
    messages = [x.strip() for x in str(l).split("\n")]
    assert "FormData parameter 'name' in function arguments" in messages
    assert "Query Parameter 'name' in function arguments" not in messages
    assert "Function argument 'name' not defined in specification" not in messages
    assert response.status_code == 200
    l.uninstall()

    def test_atexit(self):
        m = Mock()
        with Replacer() as r:
            # make sure the marker is false, other tests will
            # probably have set it
            r.replace('testfixtures.LogCapture.atexit_setup', False)
            r.replace('atexit.register', m.register)

            l = LogCapture()

            expected = [call.register(l.atexit)]

            compare(expected, m.mock_calls)

            with catch_warnings(record=True) as w:
                l.atexit()
                self.assertTrue(len(w), 1)
                compare(str(w[0].message), ( # pragma: no branch
                    "LogCapture instances not uninstalled by shutdown, "
                    "loggers captured:\n"
                    "(None,)"
                    ))
                
            l.uninstall()

            compare(set(), LogCapture.instances)
            
            # check re-running has no ill effects
            l.atexit()

class TestJsonLogFormatter(unittest2.TestCase):

    def setUp(self):
        self.handler = LogCapture()
        self.formatter = JsonLogFormatter()

    def tearDown(self):
        self.handler.uninstall()

    def test_basic_operation(self):
        logging.debug("simple test")
        self.assertEquals(len(self.handler.records), 1)
        details = json.loads(self.formatter.format(self.handler.records[0]))
        self.assertEquals(details["message"], "simple test")
        self.assertEquals(details["name"], "root")
        self.assertEquals(details["pid"], os.getpid())
        self.assertEquals(details["op"], "root")
        self.assertEquals(details["v"], 1)
        self.assertTrue("time" in details)

    def test_custom_paramters(self):
        logger = logging.getLogger("mozsvc.test.test_logging")
        logger.warn("custom test %s", "one", extra={
            "more": "stuff",
            "op": "mytest",
        })
        self.assertEquals(len(self.handler.records), 1)
        details = json.loads(self.formatter.format(self.handler.records[0]))
        self.assertEquals(details["message"], "custom test one")
        self.assertEquals(details["name"], "mozsvc.test.test_logging")
        self.assertEquals(details["op"], "mytest")
        self.assertEquals(details["more"], "stuff")

    def test_logging_error_tracebacks(self):
        try:
            raise ValueError("\n")
        except Exception:
            logging.exception("there was an error")
        self.assertEquals(len(self.handler.records), 1)
        details = json.loads(self.formatter.format(self.handler.records[0]))
        self.assertEquals(details["message"], "there was an error")
        self.assertEquals(details["error"], "ValueError('\\n',)")
        tblines = details["traceback"].strip().split("\n")
        self.assertEquals(tblines[-1], details["error"])
        self.assertEquals(tblines[-2], "<type 'exceptions.ValueError'>")