<commit_before>'''

@author: rsherman
'''
import unittest
import mock
import os

from ncclient.manager import Manager, make_device_handler
from ncclient.transport import SSHSession


facts = {'domain': None, 'hostname': 'firefly', 'ifd_style': 'CLASSIC',
         'version_info': 'junos.version_info(major=(12, 1), type=X,'
                         ' minor=(46, \'D\', 10), build=2)',
         '2RE': False, 'serialnumber': '123456789102', 'fqdn': 'firefly',
         'virtual': True, 'switch_style': 'NONE', 'version': '12.1X46-D10.2',
         'HOME': '/cf/var/home/rick', 'srx_cluster': False,
         'model': 'FIREFLY-PERIMETER',
         'RE0': {
             'status': 'Testing',
             'last_reboot_reason': 'Router rebooted after a normal shutdown.',
             'model': 'FIREFLY-PERIMETER RE', 'up_time': '1 minute,'
                                                         ' 10 seconds'},
         'personality': 'SRX_BRANCH'}


class TestDevice(unittest.TestCase):

    def setUp(self):
        self.device_params = {'name': 'junos'}
        self.device_handler = make_device_handler(self.device_params)
        self.session = SSHSession(self.device_handler)

        from jnpr.junos import Device
        self.dev = Device(host='1.1.1.1', user='rick', password='<PASSWORD>')

    def tearDown(self):
        self.session._connected = False

    @mock.patch('ncclient.manager.connect')
    @mock.patch('jnpr.junos.Device.execute')
    def test_device_open(self, mock_connect, mock_execute):
        mock_connect.side_effect = self.mock_manager
        mock_execute.side_effect = self.mock_manager
        self.dev.open()
        self.assertEqual(self.dev.connected, True)

#     def test_device_facts(self):
#         print self.dev.facts
#         assert self.dev.facts == facts

    def test_device_hostname(self):
        assert self.dev.hostname == '1.1.1.1'

    def test_device_user(self):
        assert self.dev.user == 'rick'

    def test_device_get_password(self):
        assert self.dev.password is None

    def test_device_set_password(self):
        self.dev.password = '<PASSWORD>'
        assert self.dev._password == '<PASSWORD>'

#     def test_device_get_timeout(self):
#         assert self.dev.timeout == 30

#     def test_device_set_timeout(self):
#         self.dev.timeout = 10
#         assert self.dev.timeout == 10

#     def test_device_cli(self):
#         print self.dev.cli('show cli directory', format='xml')
#         assert self.dev.cli('show cli directory').findtext('./working-directory') == '/cf/var/home/rick</working-directory'

#     def test_device_execute(self):
#         self.dev.execute = mock.MagicMock(name='execute')
#         self.dev.execute('<get-software-information/>')
#         self.assertEqual(self.dev.execute.call_args[0][
#             0], '<get-software-information/>')
# 
#     def test_device_rpcmeta(self):
#         assert self.dev.rpc.get_software_information.func_doc ==\
#             'get-software-information'

    def test_device_close(self):
        def close_conn():
            self.dev.connected = False
        self.dev.close = mock.MagicMock(name='close')
        self.dev.close.side_effect = close_conn
        self.dev.close()
        self.assertEqual(self.dev.connected, False)


    def read_file(self, fname):
        from ncclient.xml_ import NCElement

        fpath = os.path.join(os.path.dirname(__file__), fname)
        foo = open(fpath).read()

        rpc_reply = NCElement(foo, self.device_handler.transform_reply())._NCElement__doc[0]
        return rpc_reply

 
    def mock_manager(self, *args, **kwargs):
        from jnpr.junos.exception import RpcError
        if kwargs:
            device_params = kwargs['device_params']
            device_handler = make_device_handler(device_params)
            session = SSHSession(device_handler)
            return Manager(session, device_handler)
 
        elif args:
            if args[0].tag == 'command':
                if args[0].text == 'show cli directory':
                    return self.read_file('show-cli-directory.xml')
                else:
                    raise RpcError
 
            else:
                return self.read_file(args[0].tag + '.xml')

if __name__ == "__main__":
    #import sys;sys.argv = ['', 'TestDevice.testName']
    unittest.main()<commit_msg>Updated Device Test to proper Mocking<commit_after>'''

@author: rsherman
'''
import unittest
from mock import MagicMock, patch
import os

from ncclient.manager import Manager, make_device_handler
from ncclient.transport import SSHSession

from jnpr.junos.facts.swver import version_info

facts = {'domain': None, 'hostname': 'firefly', 'ifd_style': 'CLASSIC',
          'version_info': version_info('12.1X46-D15.3'),
          '2RE': False, 'serialnumber': 'aaf5fe5f9b88', 'fqdn': 'firefly',
          'virtual': True, 'switch_style': 'NONE', 'version': '12.1X46-D15.3',
          'HOME': '/cf/var/home/rick', 'srx_cluster': False,
          'model': 'FIREFLY-PERIMETER',
          'RE0': {'status': 'Testing',
                    'last_reboot_reason': 'Router rebooted after a '
                                        'normal shutdown.',
                    'model': 'FIREFLY-PERIMETER RE',
                    'up_time': '6 hours, 29 minutes, 30 seconds'},
          'vc_capable': False, 'personality': 'SRX_BRANCH'}

class TestDevice(unittest.TestCase):

    @patch('ncclient.manager.connect')
    def setUp(self, mock_connect):
        mock_connect.side_effect = self.mock_manager
        from jnpr.junos import Device
        self.dev = Device(host='1.1.1.1', user='rick', password='<PASSWORD>',
                          gather_facts=False)
        self.dev.open()

    @patch('ncclient.operations.session.CloseSession.request')
    def tearDown(self, mock_session):
        self.dev.close()

    def test_device_property_logfile_isinstance(self):
        mock = MagicMock()
        with patch('__builtin__.open', mock):
            with patch('__builtin__.file', MagicMock):
                handle = open('filename', 'r')
                self.dev.logfile = handle
                self.assertEqual(self.dev.logfile, handle)

    def test_device_property_logfile_close(self):
        self.dev._logfile = MagicMock()
        self.dev._logfile.close.return_value = 0
        self.dev.logfile = None
        self.assertFalse(self.dev._logfile)

    def test_device_open(self):
        self.assertEqual(self.dev.connected, True)

    @patch('jnpr.junos.Device.execute')
    def test_device_facts(self, mock_execute):
        mock_execute.side_effect = self.mock_manager
        self.dev.facts_refresh()
        assert self.dev.facts['version'] == facts['version']

    def test_device_hostname(self):
        assert self.dev.hostname == '1.1.1.1'

    def test_device_user(self):
        assert self.dev.user == 'rick'

    def test_device_get_password(self):
        assert self.dev.password is None

    def test_device_set_password(self):
        self.dev.password = '<PASSWORD>'
        assert self.dev._password == '<PASSWORD>'

    def test_device_get_timeout(self):
        assert self.dev.timeout == 30

    def test_device_set_timeout(self):
        self.dev.timeout = 10
        assert self.dev.timeout == 10

    def test_device_cli(self):
        self.dev.execute = MagicMock(name='execute')
        self.dev.cli('show version')
        assert self.dev.execute.call_args[0][0].text == 'show version'

    def test_device_cli_exception(self):
        self.dev.rpc.cli = MagicMock(side_effect=AttributeError)
        val = self.dev.cli('show version')
        self.assertEqual(val, 'invalid command: show version')

    def test_device_execute(self):
        self.dev.execute = MagicMock(name='execute')
        self.dev.execute('<get-software-information/>')
        self.assertEqual(self.dev.execute.call_args[0][
            0], '<get-software-information/>')

    def test_device_rpcmeta(self):
        assert self.dev.rpc.get_software_information.func_doc ==\
            'get-software-information'

    def test_device_probe_timeout_zero(self):
        with patch('jnpr.junos.device.socket') as mock_socket:
            self.assertFalse(self.dev.probe(0))

    def test_device_probe_timeout_gt_zero(self):
        with patch('jnpr.junos.device.socket') as mock_socket:
            self.assertTrue(self.dev.probe(1),
                            'probe fn is not working for'
                            ' timeout greater than zero')

    def test_device_probe_timeout_exception(self):
        with patch('jnpr.junos.device.socket') as mock_socket:
            with patch('jnpr.junos.device.time.sleep') as mock_time:
                mock_socket.socket.return_value.close.side_effect \
                    = RuntimeError
                mock_time.return_value = None
                self.assertFalse(self.dev.probe(.01))

    def test_device_bind(self):
        self.dev.bind()
        mock = MagicMock()
        mock.__name__ = 'magic mock'
        self.dev.bind(kw=mock)

    def test_device_template(self):
        self.dev._j2ldr = MagicMock()
        self.dev.Template('test')

    def test_device_close(self):
        def close_conn():
            self.dev.connected = False
        self.dev.close = MagicMock(name='close')
        self.dev.close.side_effect = close_conn
        self.dev.close()
        self.assertEqual(self.dev.connected, False)


    def read_file(self, fname):
        from ncclient.xml_ import NCElement

        fpath = os.path.join(os.path.dirname(__file__), fname)
        foo = open(fpath).read()

        rpc_reply = NCElement(foo, self.dev._conn._device_handler.transform_reply())._NCElement__doc[0]
        return rpc_reply

 
    def mock_manager(self, *args, **kwargs):
        from jnpr.junos.exception import RpcError
        if kwargs:
            device_params = kwargs['device_params']
            device_handler = make_device_handler(device_params)
            session = SSHSession(device_handler)
            return Manager(session, device_handler)

        elif args:
            if args[0].tag == 'command':
                if args[0].text == 'show cli directory':
                    return self.read_file('show-cli-directory.xml')
                else:
                    raise RpcError

            else:
                return self.read_file(args[0].tag + '.xml')

if __name__ == "__main__":
    #import sys;sys.argv = ['', 'TestDevice.testName']
    unittest.main()
{'fraction_non_alphanumeric': 0.12497614958977295, 'fraction_numerical': 0.010398778858996375, 'mean_word_length': 2.448355263157895, 'pattern_counts': {'":': 2, '<': 14, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 7}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4367731', 'n_tokens_mistral': 3278, 'n_tokens_neox': 3138, 'n_words': 590}