# 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)

[((1260, 1296), 'collections.namedtuple', 'namedtuple', (['"""Peer"""', "('host', 'port')"], {}), "('Peer', ('host', 'port'))\n", (1270, 1296), False, 'from collections import namedtuple\n'), ((2431, 2477), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', (['namespace', 'username', 'password'], {}), '(namespace, username, password)\n', (2446, 2477), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((2764, 2810), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', (['namespace', 'username', 'password'], {}), '(namespace, username, password)\n', (2779, 2810), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((3706, 3719), 'pyfarm.agent.manhole.TelnetRealm', 'TelnetRealm', ([], {}), '()\n', (3717, 3719), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((4040, 4053), 'pyfarm.agent.manhole.TelnetRealm', 'TelnetRealm', ([], {}), '()\n', (4051, 4053), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((4207, 4237), 'pyfarm.agent.manhole.TransportProtocolFactory', 'TransportProtocolFactory', (['None'], {}), '(None)\n', (4231, 4237), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((4541, 4587), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', (['namespace', 'username', 'password'], {}), '(namespace, username, password)\n', (4556, 4587), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((4605, 4615), 'StringIO.StringIO', 'StringIO', ([], {}), '()\n', (4613, 4615), False, 'from StringIO import StringIO\n'), ((4946, 4956), 'StringIO.StringIO', 'StringIO', ([], {}), '()\n', (4954, 4956), False, 'from StringIO import StringIO\n'), ((5559, 5569), 'StringIO.StringIO', 'StringIO', ([], {}), '()\n', (5567, 5569), False, 'from StringIO import StringIO\n'), ((2047, 2076), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', (['None', '""""""', '""""""'], {}), "(None, '', '')\n", (2062, 2076), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((2138, 2167), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', (['{}', 'None', '""""""'], {}), "({}, None, '')\n", (2153, 2167), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((2229, 2258), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', (['{}', '""""""', 'None'], {}), "({}, '', None)\n", (2244, 2258), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((2539, 2585), 'pyfarm.agent.manhole.manhole_factory', 'manhole_factory', (['namespace', 'username', 'password'], {}), '(namespace, username, password)\n', (2554, 2585), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((4629, 4656), 'mock.patch', 'patch', (['"""sys.stdout"""', 'output'], {}), "('sys.stdout', output)\n", (4634, 4656), False, 'from mock import patch\n'), ((4670, 4676), 'pyfarm.agent.manhole.show', 'show', ([], {}), '()\n', (4674, 4676), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((4970, 4997), 'mock.patch', 'patch', (['"""sys.stdout"""', 'output'], {}), "('sys.stdout', output)\n", (4975, 4997), False, 'from mock import patch\n'), ((5011, 5023), 'pyfarm.agent.manhole.show', 'show', (['Foobar'], {}), '(Foobar)\n', (5015, 5023), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((5583, 5610), 'mock.patch', 'patch', (['"""sys.stdout"""', 'output'], {}), "('sys.stdout', output)\n", (5588, 5610), False, 'from mock import patch\n'), ((5624, 5636), 'pyfarm.agent.manhole.show', 'show', (['Foobar'], {}), '(Foobar)\n', (5628, 5636), False, 'from pyfarm.agent.manhole import LoggingManhole, TransportProtocolFactory, TelnetRealm, manhole_factory, show\n'), ((2346, 2360), 'os.urandom', 'os.urandom', (['(32)'], {}), '(32)\n', (2356, 2360), False, 'import os\n'), ((2394, 2408), 'os.urandom', 'os.urandom', (['(32)'], {}), '(32)\n', (2404, 2408), False, 'import os\n'), ((2669, 2683), 'os.urandom', 'os.urandom', (['(32)'], {}), '(32)\n', (2679, 2683), False, 'import os\n'), ((2717, 2731), 'os.urandom', 'os.urandom', (['(32)'], {}), '(32)\n', (2727, 2731), False, 'import os\n'), ((4456, 4470), 'os.urandom', 'os.urandom', (['(32)'], {}), '(32)\n', (4466, 4470), False, 'import os\n'), ((4504, 4518), 'os.urandom', 'os.urandom', (['(32)'], {}), '(32)\n', (4514, 4518), False, 'import os\n'), ((1657, 1677), 'random.randint', 'randint', (['(1024)', '(65535)'], {}), '(1024, 65535)\n', (1664, 1677), False, 'from random import randint\n'), ((5149, 5450), 'textwrap.dedent', 'dedent', (['"""\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    )\n', (5155, 5450), False, 'from textwrap import dedent\n'), ((5762, 5982), 'textwrap.dedent', 'dedent', (['("""\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            """)\n', (5768, 5982), False, 'from textwrap import dedent\n'), ((1627, 1641), 'os.urandom', 'os.urandom', (['(12)'], {}), '(12)\n', (1637, 1641), False, 'import os\n')]