from datetime import datetime from decimal import Decimal from typing import cast, List import unittest from geniusweb.actions.Accept import Accept from geniusweb.actions.Action import Action from geniusweb.actions.Offer import Offer from geniusweb.actions.PartyId import PartyId from geniusweb.connection.ConnectionEnd import ConnectionEnd from geniusweb.inform.ActionDone import ActionDone from geniusweb.inform.Inform import Inform from geniusweb.inform.Settings import Settings from geniusweb.inform.YourTurn import YourTurn from geniusweb.issuevalue.Bid import Bid from geniusweb.issuevalue.NumberValue import NumberValue from geniusweb.progress.ProgressTime import ProgressTime from geniusweb.references.Parameters import Parameters from geniusweb.references.ProfileRef import ProfileRef from geniusweb.references.ProtocolRef import ProtocolRef from geniusweb.references.Reference import Reference from randomparty.RandomParty import RandomParty from tudelft.utilities.listener.DefaultListenable import DefaultListenable from uri import URI # type: ignore class MyConn(ConnectionEnd[Inform, Action], DefaultListenable): def __init__(self): super().__init__() self._actions:List[Action]=[] def send(self,data:Action ): self._actions.append(data) def getReference(self) -> Reference: return cast(Reference,None) def getRemoteURI(self)->URI: return None def close(self): pass def getError(self) -> Exception: return cast(Exception, None) def getActions(self)-> List[Action]: return self._actions class RandomPartyTest(unittest.TestCase): id = PartyId("party1") profileref = ProfileRef(URI("file:test/resources/japantrip1.json")) protocolref = ProtocolRef(URI("SAOP")) progress=ProgressTime(1000, datetime.fromtimestamp(12345)) parameters=Parameters() def test_smoke(self): RandomParty() def testConnect(self): connection = MyConn() party=RandomParty() party.connect(connection) party.disconnect() def testSendInfo(self): settings = Settings(self.id, self.profileref, self.protocolref, self.progress, self.parameters ) connection = MyConn() party=RandomParty() party.connect(connection) connection.notifyListeners(settings) party.disconnect() self.assertEquals([], connection.getActions()) def testSendYourTurn(self): settings = Settings(self.id, self.profileref, self.protocolref, self.progress, self.parameters ) connection = MyConn() party=RandomParty() party.connect(connection) connection.notifyListeners(settings) connection.notifyListeners(YourTurn()) party.disconnect() actions = connection.getActions() self.assertEquals(1, len(actions)) self.assertTrue(isinstance(actions[0], Offer)) print("party did an offer: "+repr(actions[0])) def testSendOfferAndYourTurn(self): settings = Settings(self.id, self.profileref, self.protocolref, self.progress, self.parameters ) # nonsense bid, party should not accept bid=Bid({'a':NumberValue(Decimal(1))}) offer = Offer(PartyId('other'), bid) connection = MyConn() party=RandomParty() party.connect(connection) connection.notifyListeners(settings) connection.notifyListeners(ActionDone(offer)) connection.notifyListeners(YourTurn()) party.disconnect() actions = connection.getActions() self.assertEquals(1, len(actions)) self.assertTrue(isinstance(actions[0], Offer))