twgs-proxy/test_1.py

from proxy import *
from twisted.trial import unittest
import sys
from twisted.python import log

log.startLogging(sys.stdout)
from pprint import pprint

from unittest.mock import Mock, patch

from shutil import copyfile

# import twisted.internet.base
# twisted.internet.base.DelayedCall.debug = True


class GameTestCase(unittest.TestCase):
    def setUp(self):
        self.game = Game()
        # Fake game init
        self.game.queue_player = defer.DeferredQueue()
        self.game.queue_game = defer.DeferredQueue()
        self.game.observer = Observer()
        self.game.setPlayerReceived()
        self.game.observer.connect("user-game", self.game.show_game)

        # This should silence trying to send data when we can't.
        # Well, it doesn't.  >:(
        self.game.playerDataReceived = Mock()
        print("Set!")

    def setRecv(self, f):
        self.game.queue_game.get().addCallback(f)

    def sent(self, line):
        print("Sent: [{0}]".format(line))
        self.setRecv(self.sent)

    def show(self, line):
        print("Recv: [{0}]".format(line))

    def inject(self, data):
        line = data.decode("latin-1", "ignore")
        self.assertIn(
            "TWGS Proxy build v", line, "Proxy activation text is in the data."
        )
        self.fired = True

    def test_inject(self):
        """ Verify that our injection is working. 
        
        This doesn't need a defer, apparently the DeferredQueue works 
        without depending on the reactor running.
        """
        self.setRecv(self.inject)
        self.fired = False
        self.game.dataReceived(b"\r\nTWGS v2.20b   www.eisonline.com\n\r\n\r\n\r")
        self.assertEquals(self.fired, True, "Data was not received from queue_game.")

    def path_part3(self, *_):
        self.assertIsInstance(self.game.gamedata, GameData)
        warps = [2565, 5468, 5128, 238]
        pprint(self.game.gamedata.warps)
        for w in warps:
            self.assertIn(w, self.game.gamedata.warps)

        # Test completed, clear the deferred.
        self.d.callback(0)

    def path_part2(self, *_):
        self.assertEquals(self.fired, True, "reactor.callLater is not working.")
        self.assertIsInstance(self.game.gamedata, GameData)

        self.game.dataReceived(
            b"The shortest path (9 hops, 27 turns) from sector 2565 to sector 2957 is:\n\r"
        )
        self.game.dataReceived(
            b"2565 > 5468 > 5128 > 238 > (3957) > (2531) > (1292) > (1892) > (4439) > (2957)\n\r"
        )
        reactor.callLater(0.1, self.path_part3)

    def path_user(self, user):
        self.fired = True

    def test_path(self):
        """ Verify that path detection is working. 
        
        This needs a deferred(), in order for reactor.CallLater events
        to fire.  (See: Observer.emit)
        """

        self.fired = False
        self.game.observer.connect("user-game", self.path_user)
        # emit user-game to init the game.gamedata (GameData) object.
        self.game.observer.emit("user-game", ("test", "A"))

        d = defer.Deferred()
        self.d = d
        reactor.callLater(0.1, self.path_part2)
        return d

    def trade1_part3(self, *_):
        print("How about now?")
        self.d.callback(0)

    def trade1_part2(self, *_):
        for line in self.p.trade_report:
            noascii = "".join([c if ord(c) < 0x7F else "" for c in line])
            print(cleanANSI(noascii))

        # print(self.p.trade_report)
        self.game.observer.load(self.p.save)
        self.p.keepalive.stop()
        self.p = None

        self.d.callback(0)
        # reactor.callLater(0.5, self.trade1_part3)

    def trade1_part1(self, *_):
        self.assertEquals(self.fired, True, "user-game emit ok.")

        ProxyMenu.awake = Mock()
        self.p = ProxyMenu(self.game)
        # self.p.keepalive.stop()
        # # stop the keepalive so the reactor is clean.

        z = coiterate(self.p.make_trade_report())
        z.addCallback(self.trade1_part2)

    def test_trade1(self):
        """ Attempt to run the trade script. 
        Oh boy.

        """
        self.fired = False
        self.game.observer.connect("user-game", self.path_user)
        # emit user-game to init the game.gamedata (GameData) object.
        copyfile("../data_Z.json", "data_Z.json")
        self.game.observer.emit("user-game", ("data", "Z"))

        d = defer.Deferred()
        self.d = d
        reactor.callLater(1.5, self.trade1_part1)
        return d