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