# No shebang line, this module is meant to be imported
#
# Copyright 2014 Oliver Palmer
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
from collections import namedtuple
from pprint import pprint
from random import randint
from StringIO import StringIO
from textwrap import dedent

try:
    from unittest.mock import patch
except ImportError:  # pragma: no cover
    from mock import patch

from twisted.internet.protocol import ServerFactory
from twisted.cred.portal import Portal
from twisted.conch.telnet import (
    ITelnetProtocol, TelnetBootstrapProtocol, TelnetTransport)

from pyfarm.agent.testutil import TestCase
from pyfarm.agent.manhole import (
    LoggingManhole, TransportProtocolFactory, TelnetRealm,
    manhole_factory, show)


Peer = namedtuple("Peer", ("host", "port"))


class FakeLoggingManhole(LoggingManhole):
    QUIT = False
    GET_PEER_CALLS = 0

    class terminal(object):
        RIGHT_ARROW, LEFT_ARROW = None, None

        class transport(object):
            @classmethod
            def getPeer(cls):
                FakeLoggingManhole.GET_PEER_CALLS += 1
                return Peer(os.urandom(12).encode("hex"), randint(1024, 65535))

    def handle_QUIT(self):
        self.QUIT = True


class TestManholeBase(TestCase):
    def setUp(self):
        TelnetRealm.NAMESPACE = None
        FakeLoggingManhole.GET_PEER_CALLS = 0
        FakeLoggingManhole.QUIT = False


class TestManholeFactory(TestManholeBase):
    def test_assertions(self):
        with self.assertRaises(AssertionError):
            manhole_factory(None, "", "")

        with self.assertRaises(AssertionError):
            manhole_factory({}, None, "")

        with self.assertRaises(AssertionError):
            manhole_factory({}, "", None)

    def test_instance_one(self):
        namespace = {"bob": None}
        username = os.urandom(32).encode("hex")
        password = os.urandom(32).encode("hex")
        manhole_factory(namespace, username, password)

        with self.assertRaises(AssertionError):
            manhole_factory(namespace, username, password)

    def test_instance(self):
        namespace = {"bob": None}
        username = os.urandom(32).encode("hex")
        password = os.urandom(32).encode("hex")
        manhole = manhole_factory(namespace, username, password)
        self.assertEqual(namespace, {"bob": None})
        self.assertEqual(
            TelnetRealm.NAMESPACE,
            {"bob": None, "pp": pprint, "show": show})
        self.assertIsInstance(manhole, ServerFactory)
        self.assertIsInstance(manhole.protocol, TransportProtocolFactory)
        self.assertIsInstance(manhole.protocol.portal, Portal)

        # There could be multiple password checkers, check for the one
        # we know we should have added.
        for _, instance in manhole.protocol.portal.checkers.items():
            found = False
            for user, passwd in instance.users.items():
                if user == username and passwd == password:
                    found = True
            if found:
                break
        else:
            self.fail("Failed to find correct username and password.")

    def test_request_avatar(self):
        realm = TelnetRealm()
        avatar = realm.requestAvatar(None, ITelnetProtocol)
        self.assertEqual(len(avatar), 3)
        self.assertIs(avatar[0], ITelnetProtocol)
        self.assertIsInstance(avatar[1], TelnetBootstrapProtocol)
        self.assertTrue(callable(avatar[2]))

    def test_request_avatar_error(self):
        realm = TelnetRealm()
        with self.assertRaises(NotImplementedError):
            realm.requestAvatar(None, None)

    def test_protocol_factory(self):
        factory = TransportProtocolFactory(None)
        transport = factory()
        self.assertIsInstance(transport, TelnetTransport)


class TestManholeShow(TestManholeBase):
    def test_uses_namespace(self):
        namespace = {"bob": None}
        username = os.urandom(32).encode("hex")
        password = os.urandom(32).encode("hex")
        manhole_factory(namespace, username, password)
        output = StringIO()
        with patch("sys.stdout", output):
            show()

        output.seek(0)
        output = output.getvalue().strip()
        self.assertEqual(output, "objects: ['bob', 'pp', 'show']")

    def test_custom_object(self):
        class Foobar(object):
            a, b, c, d, e = True, 1, "yes", {}, 0.0

        output = StringIO()
        with patch("sys.stdout", output):
            show(Foobar)

        output.seek(0)
        output = output.getvalue().strip()
        self.assertEqual(
            output,
            dedent("""
            data attributes of <class 'tests.test_agent.test_manhole.Foobar'>
                           a : True
                           b : 1
                           c : yes
                           d : {} (0 elements)
                           e : 0.0
            """).strip())

    def test_wrap_long_line(self):
        class Foobar(object):
            a = " " * 90

        output = StringIO()
        with patch("sys.stdout", output):
            show(Foobar)

        output.seek(0)
        output = output.getvalue().strip()
        self.assertEqual(
            output,
            dedent("""
            data attributes of <class 'tests.test_agent.test_manhole.Foobar'>
                           a : '                 """ +
                   """                                          '...
            """).strip())


class TestLoggingManhole(TestManholeBase):
    def test_line_received(self):
        f = FakeLoggingManhole()
        f.lineReceived("exit")
        self.assertTrue(f.QUIT)

[((40, 7, 40, 43), 'collections.namedtuple', 'namedtuple', ({(40, 18, 40, 24): '"""Peer"""', (40, 26, 40, 42): "('host', 'port')"}, {}), "('Peer', ('host', 'port'))", False, 'from collections import namedtuple\n'), ((82, 8, 82, 54), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', ({(82, 24, 82, 33): 'namespace', (82, 35, 82, 43): 'username', (82, 45, 82, 53): 'password'}, {}), '(namespace, username, password)', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((91, 18, 91, 64), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', ({(91, 34, 91, 43): 'namespace', (91, 45, 91, 53): 'username', (91, 55, 91, 63): 'password'}, {}), '(namespace, username, password)', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((113, 16, 113, 29), 'pyfarm.agent.manhole.TelnetRealm', 'TelnetRealm', ({}, {}), '()', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((121, 16, 121, 29), 'pyfarm.agent.manhole.TelnetRealm', 'TelnetRealm', ({}, {}), '()', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((126, 18, 126, 48), 'pyfarm.agent.manhole.TransportProtocolFactory', 'TransportProtocolFactory', ({(126, 43, 126, 47): 'None'}, {}), '(None)', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((136, 8, 136, 54), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', ({(136, 24, 136, 33): 'namespace', (136, 35, 136, 43): 'username', (136, 45, 136, 53): 'password'}, {}), '(namespace, username, password)', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((137, 17, 137, 27), 'StringIO.StringIO', 'StringIO', ({}, {}), '()', False, 'from StringIO import StringIO\n'), ((149, 17, 149, 27), 'StringIO.StringIO', 'StringIO', ({}, {}), '()', False, 'from StringIO import StringIO\n'), ((170, 17, 170, 27), 'StringIO.StringIO', 'StringIO', ({}, {}), '()', False, 'from StringIO import StringIO\n'), ((70, 12, 70, 41), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', ({(70, 28, 70, 32): 'None', (70, 34, 70, 36): '""""""', (70, 38, 70, 40): '""""""'}, {}), "(None, '', '')", False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((73, 12, 73, 41), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', ({(73, 28, 73, 30): '{}', (73, 32, 73, 36): 'None', (73, 38, 73, 40): '""""""'}, {}), "({}, None, '')", False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((76, 12, 76, 41), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', ({(76, 28, 76, 30): '{}', (76, 32, 76, 34): '""""""', (76, 36, 76, 40): 'None'}, {}), "({}, '', None)", False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((85, 12, 85, 58), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', ({(85, 28, 85, 37): 'namespace', (85, 39, 85, 47): 'username', (85, 49, 85, 57): 'password'}, {}), '(namespace, username, password)', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((138, 13, 138, 40), 'mock.patch', 'patch', ({(138, 19, 138, 31): '"""sys.stdout"""', (138, 33, 138, 39): 'output'}, {}), "('sys.stdout', output)", False, 'from mock import patch\n'), ((139, 12, 139, 18), 'pyfarm.agent.manhole.show', 'show', ({}, {}), '()', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((150, 13, 150, 40), 'mock.patch', 'patch', ({(150, 19, 150, 31): '"""sys.stdout"""', (150, 33, 150, 39): 'output'}, {}), "('sys.stdout', output)", False, 'from mock import patch\n'), ((151, 12, 151, 24), 'pyfarm.agent.manhole.show', 'show', ({(151, 17, 151, 23): 'Foobar'}, {}), '(Foobar)', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((171, 13, 171, 40), 'mock.patch', 'patch', ({(171, 19, 171, 31): '"""sys.stdout"""', (171, 33, 171, 39): 'output'}, {}), "('sys.stdout', output)", False, 'from mock import patch\n'), ((172, 12, 172, 24), 'pyfarm.agent.manhole.show', 'show', ({(172, 17, 172, 23): 'Foobar'}, {}), '(Foobar)', False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((80, 19, 80, 33), 'os.urandom', 'os.urandom', ({(80, 30, 80, 32): '32'}, {}), '(32)', False, 'import os\n'), ((81, 19, 81, 33), 'os.urandom', 'os.urandom', ({(81, 30, 81, 32): '32'}, {}), '(32)', False, 'import os\n'), ((89, 19, 89, 33), 'os.urandom', 'os.urandom', ({(89, 30, 89, 32): '32'}, {}), '(32)', False, 'import os\n'), ((90, 19, 90, 33), 'os.urandom', 'os.urandom', ({(90, 30, 90, 32): '32'}, {}), '(32)', False, 'import os\n'), ((134, 19, 134, 33), 'os.urandom', 'os.urandom', ({(134, 30, 134, 32): '32'}, {}), '(32)', False, 'import os\n'), ((135, 19, 135, 33), 'os.urandom', 'os.urandom', ({(135, 30, 135, 32): '32'}, {}), '(32)', False, 'import os\n'), ((54, 58, 54, 78), 'random.randint', 'randint', ({(54, 66, 54, 70): '(1024)', (54, 72, 54, 77): '(65535)'}, {}), '(1024, 65535)', False, 'from random import randint\n'), ((157, 12, 164, 16), 'textwrap.dedent', 'dedent', ({(157, 19, 164, 15): '"""\n            data attributes of <class \'tests.test_agent.test_manhole.Foobar\'>\n                           a : True\n                           b : 1\n                           c : yes\n                           d : {} (0 elements)\n                           e : 0.0\n            """'}, {}), '(\n    """\n            data attributes of <class \'tests.test_agent.test_manhole.Foobar\'>\n                           a : True\n                           b : 1\n                           c : yes\n                           d : {} (0 elements)\n                           e : 0.0\n            """\n    )', False, 'from textwrap import dedent\n'), ((178, 12, 182, 16), 'textwrap.dedent', 'dedent', ({(178, 19, 182, 15): '("""\n            data attributes of <class \'tests.test_agent.test_manhole.Foobar\'>\n                           a : \'                 """\n     + """                                          \'...\n            """)'}, {}), '(\n    """\n            data attributes of <class \'tests.test_agent.test_manhole.Foobar\'>\n                           a : \'                 """\n     + """                                          \'...\n            """)', False, 'from textwrap import dedent\n'), ((54, 28, 54, 42), 'os.urandom', 'os.urandom', ({(54, 39, 54, 41): '(12)'}, {}), '(12)', False, 'import os\n')]