class Welcome_functional_test(B3TestCase):

    def setUp(self):

        B3TestCase.setUp(self)

        with logging_disabled():
            self.adminPlugin = AdminPlugin(self.console, '@b3/conf/plugin_admin.ini')
            when(self.console).getPlugin("admin").thenReturn(self.adminPlugin)
            self.adminPlugin.onLoadConfig()
            self.adminPlugin.onStartup()

            self.conf = CfgConfigParser()
            self.p = WelcomePlugin(self.console, self.conf)

            self.joe = FakeClient(self.console, name="Joe", guid="joeguid", groupBits=1, team=b3.TEAM_RED)
            self.mike = FakeClient(self.console, name="Mike", guid="mikeguid", groupBits=1, team=b3.TEAM_RED)
            self.bill = FakeClient(self.console, name="Bill", guid="billguid", groupBits=1, team=b3.TEAM_RED)
            self.superadmin = FakeClient(self.console, name="SuperAdmin", guid="superadminguid", groupBits=128, team=b3.TEAM_RED)

    def load_config(self, config_content=None):
        """
        load the given config content, or the default config if config_content is None.
        """
        if config_content is None:
            self.conf.load(b3.getAbsolutePath('@b3/conf/plugin_welcome.ini'))
        else:
            self.conf.loadFromString(config_content)
        self.p.onLoadConfig()
        self.p.onStartup()

class Test_cmd_punkbuster(Bf3TestCase):
    def setUp(self):
        Bf3TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString(
            """[commands]
punkbuster-punk: 20
        """
        )
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()
        self.superadmin.connects("superadmin")

    def test_pb_inactive(self):
        when(self.console).write(("punkBuster.isActive",)).thenReturn(["false"])
        self.superadmin.clearMessageHistory()
        self.superadmin.says("!punkbuster test")
        self.assertEqual(["Punkbuster is not active"], self.superadmin.message_history)

    def test_pb_active(self):
        when(self.console).write(("punkBuster.isActive",)).thenReturn(["true"])
        self.superadmin.clearMessageHistory()
        self.superadmin.says("!punkbuster test")
        self.assertEqual([], self.superadmin.message_history)
        verify(self.console).write(("punkBuster.pb_sv_command", "test"))

    def test_pb_active(self):
        when(self.console).write(("punkBuster.isActive",)).thenReturn(["true"])
        self.superadmin.clearMessageHistory()
        self.superadmin.says("!punk test")
        self.assertEqual([], self.superadmin.message_history)
        verify(self.console).write(("punkBuster.pb_sv_command", "test"))

class Test_cmd_listconfig(Bf3TestCase):

    def setUp(self):
        super(Test_cmd_listconfig, self).setUp()
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
listconfig: 40

[preferences]
config_path: %(script_dir)s
            """ % {'script_dir': os.path.abspath(
            os.path.join(os.path.dirname(__file__), '../extplugins/conf/serverconfigs'))})
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()
        self.admin.connects("admin")
        self.admin.clearMessageHistory()

    def test_nominal(self):
        with patch.object(os, "listdir") as listdir_mock:
            listdir_mock.return_value = ["junk.txt", "conf1.cfg", "conf2.cfg", "hardcore.cfg"]
            self.admin.says('!listconfig')
            self.assertEqual(['Available config files: conf1, conf2, hardcore'], self.admin.message_history)

    def test_no_config(self):
        with patch.object(os, "listdir") as listdir_mock:
            listdir_mock.return_value = ["junk.txt"]
            self.admin.says('!listconfig')
            self.assertEqual(['No server config files found'], self.admin.message_history)

class mixin_cmd_nuke(object):
    def setUp(self):
        super(mixin_cmd_nuke, self).setUp()
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""
[commands]
panuke-nuke: 20
        """)
        self.p = PoweradminurtPlugin(self.console, self.conf)
        self.init_default_cvar()
        self.p.onLoadConfig()
        self.p.onStartup()

        self.sleep_patcher = patch.object(time, 'sleep')
        self.sleep_patcher.start()

        self.console.say = Mock()
        self.console.saybig = Mock()
        self.console.write = Mock()

        self.moderator.connects("2")


    def tearDown(self):
        super(mixin_cmd_nuke, self).tearDown()
        self.sleep_patcher.stop()


    def test_no_argument(self):
        self.moderator.message_history = []
        self.moderator.says("!nuke")
        self.assertEqual(['Invalid data, try !help panuke'], self.moderator.message_history)
        self.console.write.assert_has_calls([])


    def test_unknown_player(self):
        self.moderator.message_history = []
        self.moderator.says("!nuke f00")
        self.assertEqual(['No players found matching f00'], self.moderator.message_history)
        self.console.write.assert_has_calls([])

    def test_joe(self):
        self.joe.connects('3')
        self.moderator.message_history = []
        self.moderator.says("!nuke joe")
        self.assertEqual([], self.moderator.message_history)
        self.assertEqual([], self.joe.message_history)
        self.console.write.assert_has_calls([call('nuke 3')])

    def test_joe_multi(self):
        def _start_new_thread(function, args):
            function(*args)

        with patch.object(thread, 'start_new_thread', wraps=_start_new_thread):
            self.joe.connects('3')
            self.moderator.message_history = []
            self.moderator.says("!nuke joe 3")
            self.assertEqual([], self.moderator.message_history)
            self.assertEqual([], self.joe.message_history)
            self.console.write.assert_has_calls([call('nuke 3'), call('nuke 3'), call('nuke 3')])

class Test_cmd_lms(Iourt42TestCase):
    def setUp(self):
        super(Test_cmd_lms, self).setUp()
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""
[commands]
palms-lms: 20           ; change game type to Last Man Standing
        """)
        self.p = PoweradminurtPlugin(self.console, self.conf)

        when(self.console).getCvar('timelimit').thenReturn(Cvar('timelimit', value=20))
        when(self.console).getCvar('g_maxGameClients').thenReturn(Cvar('g_maxGameClients', value=16))
        when(self.console).getCvar('sv_maxclients').thenReturn(Cvar('sv_maxclients', value=16))
        when(self.console).getCvar('sv_privateClients').thenReturn(Cvar('sv_privateClients', value=0))
        when(self.console).getCvar('g_allowvote').thenReturn(Cvar('g_allowvote', value=0))
        self.p.onLoadConfig()
        self.p.onStartup()

        self.console.say = Mock()
        self.console.write = Mock()

        self.moderator.connects("2")


    def test_nominal(self):
        self.moderator.message_history = []
        self.moderator.says("!lms")
        self.console.write.assert_has_calls([call('g_gametype 1')])
        self.assertEqual(['game type changed to Last Man Standing'], self.moderator.message_history)

class SpamcontrolTestCase(B3TestCase):
    """
    Ease testcases that need an working B3 console and need to control the Spamcontrol plugin config
    """

    def setUp(self):
        self.timer_patcher = patch('threading.Timer')
        self.timer_patcher.start()

        self.log = logging.getLogger('output')
        self.log.propagate = False

        B3TestCase.setUp(self)
        self.console.startup()
        self.log.propagate = True

    def tearDown(self):
        B3TestCase.tearDown(self)
        self.timer_patcher.stop()

    def init_plugin(self, config_content):
        self.conf = CfgConfigParser()
        self.conf.loadFromString(config_content)
        self.p = SpamcontrolPlugin(self.console, self.conf)

        self.log.setLevel(logging.DEBUG)
        self.log.info("============================= Spamcontrol plugin: loading config ============================")
        self.p.onLoadConfig()
        self.log.info("============================= Spamcontrol plugin: starting  =================================")
        self.p.onStartup()

class Test_cmd_yellplayer(Bf3TestCase):

    def setUp(self):
        Bf3TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
yellplayer: 20

[preferences]
yell_duration: 2
        """)
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()


    def test_no_argument(self):
        self.moderator.connects("moderator")
        self.moderator.message_history = []
        self.moderator.says("!yellplayer")
        self.assertEqual(1, len(self.moderator.message_history))
        self.assertEqual('invalid parameters, try !help yellplayer', self.moderator.message_history[0])

    def test_nominal(self):
        self.joe.connects('joe')
        self.moderator.connects("moderator")
        self.moderator.says("!yellplayer joe changing map soon !")
        self.console.write.assert_called_once_with(('admin.yell', 'changing map soon !', '2', 'player', 'joe'))

    def setUp(self):
        # create a FakeConsole parser
        parser_ini_conf = CfgConfigParser()
        parser_ini_conf.loadFromString(r'''''')
        self.parser_main_conf = MainConfig(parser_ini_conf)

        with logging_disabled():
            from b3.fake import FakeConsole
            self.console = FakeConsole(self.parser_main_conf)

        with logging_disabled():
            self.adminPlugin = AdminPlugin(self.console, '@b3/conf/plugin_admin.ini')
            self.adminPlugin._commands = {}
            self.adminPlugin.onStartup()

        # make sure the admin plugin obtained by other plugins is our admin plugin
        when(self.console).getPlugin('admin').thenReturn(self.adminPlugin)

        # create our plugin instance
        self.p = SpreePlugin(self.console, CfgConfigParser())

        with logging_disabled():
            from b3.fake import FakeClient

        self.mike = FakeClient(console=self.console, name="Mike", guid="MIKEGUID", groupBits=1)
        self.bill = FakeClient(console=self.console, name="Bill", guid="BILLGUID", groupBits=1)

class Test_cmd_skins(Iourt42TestCase):
    def setUp(self):
        super(Test_cmd_skins, self).setUp()
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""
[commands]
pagoto-goto: 20         ; set the goto <on/off>
        """)
        self.p = PoweradminurtPlugin(self.console, self.conf)
        self.init_default_cvar()
        self.p.onLoadConfig()
        self.p.onStartup()

        self.console.say = Mock()
        self.console.write = Mock()

        self.moderator.connects("2")

    def test_missing_parameter(self):
        self.moderator.message_history = []
        self.moderator.says("!goto")
        self.assertListEqual(["invalid or missing data, try !help pagoto"], self.moderator.message_history)

    def test_junk(self):
        self.moderator.message_history = []
        self.moderator.says("!goto qsdf")
        self.assertListEqual(["invalid or missing data, try !help pagoto"], self.moderator.message_history)

    def test_on(self):
        self.moderator.says("!goto on")
        self.console.write.assert_has_calls([call('set g_allowgoto "1"')])

    def test_off(self):
        self.moderator.says("!goto off")
        self.console.write.assert_has_calls([call('set g_allowgoto "0"')])

class Test_cmd_serverreboot(Bf3TestCase):
    def setUp(self):
        super(Test_cmd_serverreboot, self).setUp()
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
serverreboot: 100
        """)
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()


    def test_nominal(self, sleep_mock):
        self.console.write.expect(('admin.shutDown',))
        self.superadmin.connects("god")
        self.superadmin.says("!serverreboot")
        self.console.write.verify_expected_calls()

    def test_frostbite_error(self, sleep_mock):
        self.console.write.expect(('admin.shutDown',)).thenRaise(CommandFailedError(['fOO']))
        self.superadmin.connects("god")
        self.superadmin.message_history = []
        self.superadmin.says("!serverreboot")
        self.console.write.verify_expected_calls()
        self.assertEqual(['Error: fOO'], self.superadmin.message_history)

class Test_other_statuses(Bf3TestCase):
    def setUp(self):
        Bf3TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""
[commands]
bf3stats: 0
""")
        self.p = Bf3StatsPlugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()

    @patch("urllib.urlopen", new=urlopen_not_found_mock)
    def test_not_found(self):
        self.joe.connects('Joe')
        self.joe.says("!bf3stats")
        self.assertEqual(["Error while querying bf3stats.com. Player 'Joe' not found"], self.joe.message_history)

    @patch("urllib.urlopen", new=urlopen_invalid_name_mock)
    def test_invalid_name(self):
        self.joe.connects('Joe')
        self.joe.says("!bf3stats")
        self.assertEqual(["Error while querying bf3stats.com. Error while querying 'Joe' : invalid_name"],
                         self.joe.message_history)

    @patch("urllib.urlopen", new=urlopen_pifound_mock)
    def test_pifound(self):
        self.joe.connects('Joe')
        self.joe.says("!bf3stats")
        self.assertEqual(['bf3stats.com has no stats for Joe'], self.joe.message_history)

class Test_cmd_yellteam(Bf4TestCase):

    def setUp(self):
        Bf4TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
yellteam: 20

[preferences]
yell_duration: 2
        """)
        self.p = Poweradminbf4Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()


    def test_no_argument(self):
        self.moderator.connects("moderator")
        self.moderator.message_history = []
        self.moderator.says("!yellteam")
        self.assertEqual(1, len(self.moderator.message_history))
        self.assertEqual('missing parameter, try !help yellteam', self.moderator.message_history[0])

    def test_nominal(self):
        self.moderator.connects("moderator")
        self.moderator.teamId = 3
        with patch.object(self.console, "write") as write_mock:
            self.moderator.says("!yellteam changing map soon !")
        write_mock.assert_called_once_with(('admin.yell', 'changing map soon !', '2', 'team', '3'))

class Test_cmd_scramble(Bf3TestCase):

    def setUp(self):
        Bf3TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
scramble: 20
        """)
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()
        self.p._scrambler = Mock()
        self.superadmin.connects('superadmin')
        self.superadmin.clearMessageHistory()

    def test_none(self):
        self.p._scrambling_planned = None
        self.superadmin.says('!scramble')
        self.assertEqual(['Teams will be scrambled at next round start'], self.superadmin.message_history)
        self.assertTrue(self.p._scrambling_planned)

    def test_true(self):
        self.p._scrambling_planned = True
        self.superadmin.says('!scramble')
        self.assertEqual(['Teams scrambling canceled for next round'], self.superadmin.message_history)
        self.assertFalse(self.p._scrambling_planned)

    def test_false(self):
        self.p._scrambling_planned = False
        self.superadmin.says('!scramble')
        self.assertEqual(['Teams will be scrambled at next round start'], self.superadmin.message_history)
        self.assertTrue(self.p._scrambling_planned)

    def test_nominal(self):
        # GIVEN
        conf = CfgConfigParser()
        conf.loadFromString(dedent(r"""
            [commands]
            mapstats-stats: 2
            testscore-ts: 2
            topstats-top: 20
            topxp: 20

            [settings]
            startPoints: 150
            resetscore: yes
            resetxp: yes
            show_awards: yes
            show_awards_xp: yes
        """))
        self.p = StatsPlugin(self.console, conf)
        # WHEN
        self.p.onLoadConfig()
        # THEN
        self.assertEqual(2, self.p.mapstatslevel)
        self.assertEqual(2, self.p.testscorelevel)
        self.assertEqual(20, self.p.topstatslevel)
        self.assertEqual(20, self.p.topxplevel)
        self.assertEqual(150, self.p.startPoints)
        self.assertTrue(self.p.resetscore)
        self.assertTrue(self.p.resetxp)
        self.assertTrue(self.p.show_awards)
        self.assertTrue(self.p.show_awards_xp)

class Test_cmd_yellsquad(Bf3TestCase):

    def setUp(self):
        Bf3TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
yellsquad: 20

[preferences]
yell_duration: 2
        """)
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()


    def test_no_argument(self):
        self.moderator.connects("moderator")
        self.moderator.message_history = []
        self.moderator.says("!yellsquad")
        self.assertEqual(1, len(self.moderator.message_history))
        self.assertEqual('missing parameter, try !help yellsquad', self.moderator.message_history[0])

    def test_nominal(self):
        self.moderator.connects("moderator")
        self.moderator.teamId = 3
        self.moderator.squad = 4
        self.moderator.says("!yellsquad changing map soon !")
        self.console.write.assert_called_once_with(('admin.yell', 'changing map soon !', '2', 'squad', '3', '4'))

class Test_commands(PluginTestCase):
    def setUp(self):
        super(Test_commands, self).setUp()
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
vaccheck: 20
        """)
        self.p = VacbanPlugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()

        when(self.p)._query_service(anything()).thenReturn(vac_response_not_banned)
        self.moderator.connects("2")



    def test_vaccheck(self):
        # GIVEN
        self.moderator.message_history = []
        # WHEN
        with patch.object(self.p, '_checkConnectedPlayers') as mock_checkConnectedPlayers:
            self.moderator.says("!vaccheck")
        # THEN
        self.assertEqual(['checking players ...', 'done'], self.moderator.message_history)
        self.assertEqual(1, mock_checkConnectedPlayers.call_count)

class Test_cmd_punkbuster(Bf4TestCase):

    def setUp(self):
        Bf4TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
punkbuster-punk: 20
        """)
        self.p = Poweradminbf4Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()
        self.superadmin.connects('superadmin')


    def test_pb_inactive(self):
        when(self.console).write(('punkBuster.isActive',)).thenReturn(['false'])
        self.superadmin.clearMessageHistory()
        self.superadmin.says('!punkbuster test')
        self.assertEqual(['Punkbuster is not active'], self.superadmin.message_history)

    def test_pb_active(self):
        when(self.console).write(('punkBuster.isActive',)).thenReturn(['true'])
        self.superadmin.clearMessageHistory()
        self.superadmin.says('!punkbuster test')
        self.assertEqual([], self.superadmin.message_history)
        verify(self.console).write(('punkBuster.pb_sv_command', 'test'))

    def test_pb_active(self):
        when(self.console).write(('punkBuster.isActive',)).thenReturn(['true'])
        self.superadmin.clearMessageHistory()
        self.superadmin.says('!punk test')
        self.assertEqual([], self.superadmin.message_history)
        verify(self.console).write(('punkBuster.pb_sv_command', 'test'))

class Test_config(Bf3TestCase):
    default_value = 3
    minimum_value = 2

    def assert_config_value(self, expected, conf_value):
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""
[preferences]
team_swap_threshold: %s
                    """ % conf_value)
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.assertEqual(expected, self.p._team_swap_threshold)

    def test_default_value(self):
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[foo]""")
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.assertEqual(self.default_value, self.p._team_swap_threshold)

    def test_nominal(self):
        self.assert_config_value(6, '6')

    def test_value_too_low(self):
        self.assert_config_value(self.minimum_value, '1')

    def test_negative_value(self):
        self.assert_config_value(self.minimum_value, '-2')

    def test_float(self):
        self.assert_config_value(self.default_value, '3.54')

    def test_junk(self):
        self.assert_config_value(self.default_value, 'junk')

class Test_cmd_roundrestart(Bf3TestCase):

    @classmethod
    def setUpClass(cls):
        Bf3TestCase.setUpClass()
        cls.sleep_patcher = patch.object(time, "sleep")
        cls.sleep_patcher.start()

    @classmethod
    def tearDownClass(cls):
        cls.sleep_patcher.stop()

    def setUp(self):
        Bf3TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
roundrestart: 20
        """)
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()
        self.superadmin.connects('superadmin')


    def test_nominal(self):
        self.superadmin.clearMessageHistory()
        self.superadmin.says('!roundrestart')
        self.assertEqual([], self.superadmin.message_history)
        self.console.write.assert_has_calls([call(('mapList.restartRound',))])

class Test_cmd_vipclear(Bf3TestCase):
    def setUp(self):
        Bf3TestCase.setUp(self)
        self.conf = CfgConfigParser()
        self.conf.loadFromString("""[commands]
vipclear: 20
        """)
        self.p = Poweradminbf3Plugin(self.console, self.conf)
        self.p.onLoadConfig()
        self.p.onStartup()

        self.moderator.connects("moderator")


    def test_nominal(self):
        when(self.console).write(('reservedSlotsList.clear',)).thenReturn([])
        self.moderator.connects("moderator")
        self.moderator.message_history = []
        self.moderator.says("!vipclear")
        self.assertEqual(1, len(self.moderator.message_history))
        self.assertEqual('VIP list is now empty', self.moderator.message_history[0])

    def test_frostbite_error(self):
        when(self.console).write(('reservedSlotsList.clear',)).thenRaise(CommandFailedError(['f00']))
        self.moderator.connects("moderator")
        self.moderator.message_history = []
        self.moderator.says("!vipclear")
        self.assertEqual(["Error: f00"], self.moderator.message_history)