Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
event_loop,
database,
mock_protocol,
mock_games,
mock_players,
mock_player,
mock_geoip,
mock_nts_client
):
lc = LobbyConnection(
database=database,
geoip=mock_geoip,
game_service=mock_games,
players=mock_players,
nts_client=mock_nts_client,
ladder_service=asynctest.create_autospec(LadderService)
)
lc.player = mock_player
lc.protocol = mock_protocol
lc.player_service.fetch_player_data = CoroutineMock()
lc.peer_address = Address('127.0.0.1', 1234)
lc._authenticated = True
return lc
def mock_protocol():
return asynctest.create_autospec(QDataStreamProtocol(mock.Mock(), mock.Mock()))
async def test_launch_game(lobbyconnection, game, player_factory):
old_game_conn = asynctest.create_autospec(GameConnection)
lobbyconnection.player = player_factory()
lobbyconnection.game_connection = old_game_conn
lobbyconnection.send = CoroutineMock()
await lobbyconnection.launch_game(game)
# Verify all side effects of launch_game here
old_game_conn.abort.assert_called_with("Player launched a new game")
assert lobbyconnection.game_connection is not None
assert lobbyconnection.game_connection.game == game
assert lobbyconnection.player.game == game
assert lobbyconnection.player.game_connection == lobbyconnection.game_connection
assert lobbyconnection.game_connection.player == lobbyconnection.player
assert lobbyconnection.player.state == PlayerState.JOINING
lobbyconnection.send.assert_called_once()
async def test_process_game_stats(
game_stats_service, event_service, achievement_service, player, game
):
with open("tests/data/game_stats_full_example.json", "r") as stats_file:
stats = json.loads(stats_file.read())["stats"]
mock_lconn = asynctest.create_autospec(LobbyConnection)
player.lobby_connection = mock_lconn
await game_stats_service.process_game_stats(player, game, stats)
event_service.record_event.assert_any_call(ev.EVENT_LOST_ACUS, 0, [])
event_service.record_event.assert_any_call(ev.EVENT_BUILT_AIR_UNITS, 1, [])
event_service.record_event.assert_any_call(ev.EVENT_LOST_AIR_UNITS, 2, [])
event_service.record_event.assert_any_call(
ev.EVENT_BUILT_LAND_UNITS, 4, []
)
event_service.record_event.assert_any_call(ev.EVENT_LOST_LAND_UNITS, 5, [])
event_service.record_event.assert_any_call(
ev.EVENT_BUILT_NAVAL_UNITS, 33, []
)
event_service.record_event.assert_any_call(
ev.EVENT_LOST_NAVAL_UNITS, 11, []
def mock_nts_client():
return asynctest.create_autospec(TwilioNTS)
async def test_connect_to_peer_disconnected(game_connection):
# Weak reference has dissapeared
await game_connection.connect_to_peer(None)
peer = asynctest.create_autospec(GameConnection)
peer.send_ConnectToPeer.side_effect = DisconnectedError("Test error")
# The client disconnects right as we send the message
await game_connection.connect_to_peer(peer)
def mock_games():
return asynctest.create_autospec(GameService)
async def test_handle_action_IceMsg(
game_connection: GameConnection,
player_service,
player_factory
):
peer = player_factory(player_id=2)
peer.game_connection = asynctest.create_autospec(GameConnection)
player_service[peer.id] = peer
await game_connection.handle_action("IceMsg", [2, "the message"])
peer.game_connection.send.assert_called_once_with({
"command": "IceMsg",
"args": [game_connection.player.id, "the message"]
})
async def test_broadcast_shutdown_error(player_factory, player_service):
player = player_factory()
lconn = asynctest.create_autospec(LobbyConnection)
lconn.send_warning.side_effect = ValueError
player.lobby_connection = lconn
player_service[0] = player
await player_service.shutdown()
player.lobby_connection.send_warning.assert_called_once()