def test_noretry(self):
        # SUCCESS
        commit = db.getLastProcessedCommit(self.session, 'python-pysaml2')
        # FAILED
        commit2 = db.getLastProcessedCommit(self.session, 'python-alembic')
        # SUCCESS, RETRY (should be ignored)
        commit3 = \
            db.getLastProcessedCommit(self.session, 'python-tripleoclient')

        mock_fp = MagicMock()
        utils.dumpshas2file(mock_fp, commit, "a", "b", commit.status, 0,
                            ['python-saml2-1.0-1.el7.src.rpm'])
        utils.dumpshas2file(mock_fp, commit2, "a", "b", commit2.status, 1,
                            ['python-alembic-1.0-2.el7.src.rpm'])
        utils.dumpshas2file(mock_fp, commit3, "a", "b", commit3.status, 2,
                            ['file1-1.2-3.el7.noarch.rpm',
                             'file2-1.2-3.el7.src.rpm'])
        expected = [
            call.write(u'python-pysaml2,a,3a9326f251b9a4162eb0dfa9f1c924ef47c'
                       '2c55a,b,024e24f0cf4366c2290c22f24e42de714d1addd1'
                       ',SUCCESS,0,python-saml2-1.0-1.el7\n'),
            call.write(u'python-alembic,a,459549c9ab7fef91b2dc8986bc0643bb2f6'
                       'ec0c8,b,885e80778edb6cbb8ee4d8909623be8062369a04'
                       ',FAILED,1,python-alembic-1.0-2.el7\n'),
            call.write(u'python-tripleoclient,a,1da7b10e55abf8c518e8f61ee7966'
                       '188f0405f59,b,0b1ce934e5b2e7d45a448f6555d24036f9aeca51'
                       ',SUCCESS,2,file2-1.2-3.el7\n')
        ]
        self.assertEqual(mock_fp.mock_calls, expected)

def test_log(mock_stderr, mock_stdout):
    sshuttle.helpers.log("message")
    sshuttle.helpers.log("abc")
    sshuttle.helpers.log("message 1\n")
    sshuttle.helpers.log("message 2\nline2\nline3\n")
    sshuttle.helpers.log("message 3\nline2\nline3")
    assert mock_stdout.mock_calls == [
        call.flush(),
        call.flush(),
        call.flush(),
        call.flush(),
        call.flush(),
    ]
    assert mock_stderr.mock_calls == [
        call.write('prefix: message'),
        call.flush(),
        call.write('prefix: abc'),
        call.flush(),
        call.write('prefix: message 1\n'),
        call.flush(),
        call.write('prefix: message 2\n'),
        call.write('---> line2\n'),
        call.write('---> line3\n'),
        call.flush(),
        call.write('prefix: message 3\n'),
        call.write('---> line2\n'),
        call.write('---> line3\n'),
        call.flush(),
    ]

 def test_restore(self, fmd):
     self.driver.filenames = MagicMock(return_value=["foo", "bar"])
     self.driver.storage_only = MagicMock(return_value=["baz"])
     fmd().replace_file.return_value = True
     self.assertRaises(media.MediaRestoreException, self.driver.restore, self.archive)
     writer = self.storage.open()
     reader = self.archive.open()
     self.driver.restore(self.archive, True)
     self.assertEqual(writer.mock_calls, [
         call.write(reader.read()),
         call.write(reader.read())
     ])

    def test_lineReceived_with_response(self, m_maybeDeferred, m_getLogger):

        callback, m_transport = self._test_lineReceived(m_maybeDeferred)

        # Now test the callback with 0 response lines:
        callback(['a', 'b'])

        # Assert the responses are written:
        self.assertEqual(
            m_transport.mock_calls,
            [call.write('a\n'),
             call.write('b\n')])

	def test_begin_initializes_lcd(self):
		gpio = Mock()
		spi = Mock()
		lcd = LCD.PCD8544(1, 2, gpio=gpio, spi=spi)
		lcd.begin(40)
		# Verify RST is set low then high.
		gpio.assert_has_calls([call.set_low(2), call.set_high(2)])
		# Verify SPI calls.
		spi.assert_has_calls([call.write([0x21]), 
							  call.write([0x14]),
							  call.write([0xA8]),
							  call.write([0x20]),
							  call.write([0x0c])])

    def test_slide(self):
        for direction in ['left', 'right', 'up', 'down']:
            self.vk.return_value.press_keysym.reset_mock()
            self.vk.return_value.release_keysym.reset_mock()
            self.musca.slide(direction)
            self.assert_in_press_release((direction.capitalize(),
                                          self.musca.Mod1))
        #  Test invalid direction

        with patch("sys.stdout") as fake_stdout:
            self.musca.slide("hulahoop")
            fake_stdout.assert_has_calls([call.write("direction not in ['left', 'right', 'up', 'down']"),
                                          call.write("\n")])

    def _test_flaky_plugin_report(self, expected_stream_value):
        mock_stream = MagicMock()
        self._mock_stream.getvalue.return_value = expected_stream_value

        self._flaky_plugin.report(mock_stream)

        self.assertEqual(
            mock_stream.mock_calls,
            [
                call.write('===Flaky Test Report===\n\n'),
                call.write(expected_stream_value),
                call.write('\n===End Flaky Test Report===\n'),
            ],
        )

    def test_profile_rule_show_human_readable(self, m_print, m_client_get_profile):
        """
        Test for profile_rule_show function when human_readable=True
        """
        # Set up arguments
        profile_name = 'Profile_1'

        # Set up mock objects
        m_Rule = Mock(spec=Rule)
        m_Rule.pprint = Mock()
        m_Rules = Mock(spec=Rules, id=profile_name, inbound_rules=[m_Rule],
                       outbound_rules=[m_Rule])
        m_Profile = Mock(spec=Profile, name=profile_name, rules=m_Rules)
        m_client_get_profile.return_value = m_Profile

        # Call method under test
        profile_rule_show(profile_name, human_readable=True)

        # Assert
        m_client_get_profile.assert_called_once_with(profile_name)
        m_print.assert_has_calls([
            call.write('Inbound rules:'),
            call.write('\n'),
            call.write(' %3d %s' % (1, m_Rule.pprint())),
            call.write('\n'),
            call.write('Outbound rules:'),
            call.write('\n'),
            call.write(' %3d %s' % (1, m_Rule.pprint())),
            call.write('\n'),
        ])

    def _make_request(
            self,
            method, postpath, reqbody,
            resreadsreq, rescode, resbody):

        m_request = MagicMock(name='Request')
        m_request.method = method
        m_request.postpath = postpath
        if reqbody is None:
            readrv = ''
        elif reqbody == 'mangled JSON':
            readrv = reqbody
        else:
            readrv = json.dumps(reqbody, indent=2)

        m_request.content.read.return_value = readrv

        r = self.tar.render(m_request)

        self.assertEqual(r, server.NOT_DONE_YET)

        expected = [
            call.setResponseCode(rescode),
            call.setHeader('Content-Type', 'application/json'),
            call.write(json.dumps(resbody, indent=2)),
            call.finish(),
        ]

        if resreadsreq:
            expected.insert(0, call.content.read())

        check_mock(self, m_request, expected)

    def test_help(self, m_stdout, m_stderr, m_PythonLoggingObserver, m_basicConfig):

        self.assertRaises(SystemExit, clargs.parse_args, ['--help'])

        self.checkCalls(m_stdout, call.write(ArgStartsWith('usage: ')))
        self.checkCalls(m_stderr)
        self.checkCalls(m_basicConfig)
        self.checkCalls(m_PythonLoggingObserver)

def test_debug3(mock_stderr, mock_stdout):
    sshuttle.helpers.debug3("message")
    assert mock_stdout.mock_calls == [
        call.flush(),
    ]
    assert mock_stderr.mock_calls == [
        call.write('prefix: message'),
        call.flush(),
    ]

    def test_debug(self, mock_time):
        mock_time.return_value = time.mktime(datetime(2016, 12, 18).timetuple())
        self.base_collector.logfile = Mock()

        self.base_collector.debug('test')

        self.assertEqual(
            self.base_collector.logfile.mock_calls[0],
            call.write('2016-12-18:00:00:00 test\n')
        )

def test_main(mock_get_method, mock_setup_daemon):
    stdin, stdout = setup_daemon()
    mock_setup_daemon.return_value = stdin, stdout

    if not os.path.isdir("tmp"):
        os.mkdir("tmp")

    sshuttle.firewall.main("test", False)

    with open("tmp/hosts") as f:
        line = f.readline()
        s = line.split()
        assert s == ['1.2.3.3', 'existing']

        line = f.readline()
        assert line == ""

    stdout.mock_calls == [
        call.write('READY test\n'),
        call.flush(),
        call.write('STARTED\n'),
        call.flush()
    ]
    mock_setup_daemon.mock_calls == [call()]
    mock_get_method.mock_calls == [
        call('test'),
        call().setup_firewall(
            1024, 1026,
            [(10, u'2404:6800:4004:80c::33')],
            10,
            [(10, 64, False, u'2404:6800:4004:80c::'),
                (10, 128, True, u'2404:6800:4004:80c::101f')],
            True),
        call().setup_firewall(
            1025, 1027,
            [(2, u'1.2.3.33')],
            2,
            [(2, 24, False, u'1.2.3.0'), (2, 32, True, u'1.2.3.66')],
            True),
        call().setup_firewall()(),
        call().setup_firewall(1024, 0, [], 10, [], True),
        call().setup_firewall(1025, 0, [], 2, [], True),
    ]

def test_main(mock_get_method, mock_setup_daemon, mock_rewrite_etc_hosts):
    stdin, stdout = setup_daemon()
    mock_setup_daemon.return_value = stdin, stdout

    mock_get_method("not_auto").name = "test"
    mock_get_method.reset_mock()

    sshuttle.firewall.main("not_auto", False)

    assert mock_rewrite_etc_hosts.mock_calls == [
        call({'1.2.3.3': 'existing'}, 1024),
        call({}, 1024),
    ]

    assert stdout.mock_calls == [
        call.write('READY test\n'),
        call.flush(),
        call.write('STARTED\n'),
        call.flush()
    ]
    assert mock_setup_daemon.mock_calls == [call()]
    assert mock_get_method.mock_calls == [
        call('not_auto'),
        call().setup_firewall(
            1024, 1026,
            [(AF_INET6, u'2404:6800:4004:80c::33')],
            AF_INET6,
            [(AF_INET6, 64, False, u'2404:6800:4004:80c::', 0, 0),
                (AF_INET6, 128, True, u'2404:6800:4004:80c::101f', 80, 80)],
            True,
            None),
        call().setup_firewall(
            1025, 1027,
            [(AF_INET, u'1.2.3.33')],
            AF_INET,
            [(AF_INET, 24, False, u'1.2.3.0', 8000, 9000),
                (AF_INET, 32, True, u'1.2.3.66', 8080, 8080)],
            True,
            None),
        call().restore_firewall(1024, AF_INET6, True, None),
        call().restore_firewall(1025, AF_INET, True, None),
    ]

 def test_write(self):
     """
     Does it write the configuration to a file?
     """
     open_file = MagicMock()
     for key in self.parser.defaults():
         del(self.parser.defaults()[key])
     calls = [call.write(line + '\n') for line in SAMPLE.split('\n')]
     self.adapter.write(open_file)
     self.assertEqual(calls, open_file.mock_calls)
     return

    def test_update_nrpe_config(self, nrpe, install_nrpe_scripts):
        nrpe_compat = MagicMock()
        nrpe_compat.checks = [MagicMock(shortname="haproxy"),
                              MagicMock(shortname="haproxy_queue")]
        nrpe.return_value = nrpe_compat

        hooks.update_nrpe_config()

        self.assertEqual(
            nrpe_compat.mock_calls,
            [call.add_check('haproxy', 'Check HAProxy', 'check_haproxy.sh'),
             call.add_check('haproxy_queue', 'Check HAProxy queue depth',
                            'check_haproxy_queue_depth.sh'),
             call.write()])

    def test_write_disclaimer(self, mock_workbook):
        setting = Setting.objects.first() or SettingFactory()
        download = DownloadFactory()

        line_1 = 'line_1'
        line_2 = 'line_2'
        setting.export_excel_disclaimer = '{line_1}\n{line_2}'.format(line_1=line_1, line_2=line_2)
        setting.save()

        mock_worksheet = MagicMock()
        mock_workbook().add_worksheet.return_value = mock_worksheet

        with patch('allegation.services.download_allegations.os'):
            allegation_download = AllegationsDownload(download.id)
            allegation_download.init_workbook()
            allegation_download.write_disclaimer()

            expected_calls = [
                call.write('A1', line_1),
                call.write('A2', line_2)
            ]

            mock_worksheet.assert_has_calls(expected_calls)

    def _test_flaky_plugin_handles_success(
        self,
        current_passes=0,
        current_runs=0,
        is_test_method=True,
        max_runs=2,
        min_passes=1
    ):
        self._expect_test_flaky(is_test_method, max_runs, min_passes)
        self._set_flaky_attribute(
            is_test_method,
            FlakyNames.CURRENT_PASSES,
            current_passes
        )
        self._set_flaky_attribute(
            is_test_method,
            FlakyNames.CURRENT_RUNS,
            current_runs
        )

        too_few_passes = current_passes + 1 < min_passes
        retries_remaining = current_runs + 1 < max_runs
        expected_plugin_handles_success = too_few_passes and retries_remaining

        self._flaky_plugin.prepareTestCase(self._mock_test_case)
        actual_plugin_handles_success = self._flaky_plugin.addSuccess(
            self._mock_test_case
        )

        self.assertEqual(
            expected_plugin_handles_success,
            actual_plugin_handles_success,
            'Expected plugin{} to handle the test run, but it did{}.'.format(
                ' to' if expected_plugin_handles_success else '',
                '' if actual_plugin_handles_success else ' not'
            )
        )
        self._assert_flaky_attributes_contains(
            {
                FlakyNames.CURRENT_PASSES: current_passes + 1,
                FlakyNames.CURRENT_RUNS: current_runs + 1,
            }
        )
        expected_test_case_calls = [call.address(), call.address()]
        expected_stream_calls = [call.writelines([
            self._mock_test_method_name,
            " passed {} out of the required {} times. ".format(
                current_passes + 1, min_passes,
            ),
        ])]
        if expected_plugin_handles_success:
            expected_test_case_calls.append(call.run(self._mock_test_result))
            expected_stream_calls.append(
                call.write(
                    'Running test again until it passes {} times.\n'.format(
                        min_passes,
                    ),
                ),
            )
        else:
            expected_stream_calls.append(call.write('Success!\n'))
        self.assertEqual(
            self._mock_test_case.mock_calls,
            expected_test_case_calls,
            'Unexpected TestCase calls = {} vs {}'.format(
                self._mock_test_case.mock_calls,
                expected_test_case_calls,
            ),
        )
        self.assertEqual(self._mock_stream.mock_calls, expected_stream_calls)

 def test_handle_unknown_git_command(self, m_stop, m_exit, m_stderr):
     self.assertIsNone(self._hgcs('flub'))
     self.assertEqual(m_stderr.mock_calls, [call.write("Unknown Command 'flub'\n")])
     self.assertEqual(m_stop.mock_calls, [call()])
     self.assertEqual(m_exit.mock_calls, [call(UnknownCommandExitStatus)])