tests/test_reactors.py

   1 import unittest
   2 try:
   3     from unittest import mock
   4 except ImportError:
   5     import mock
   6
   7 from minecraft import SUPPORTED_PROTOCOL_VERSIONS
   8 from minecraft.networking.connection import (
   9     LoginReactor, PlayingReactor, ConnectionContext
  10 )
  11 from minecraft.networking.packets import clientbound
  12
  13
  14 latest_proto = SUPPORTED_PROTOCOL_VERSIONS[-1]
  15
  16
  17 class LoginReactorTest(unittest.TestCase):
  18
  19     @mock.patch('minecraft.networking.connection.encryption')
  20     def test_encryption_online_server(self, encrypt):
  21         connection = mock.MagicMock()
  22         connection.context = ConnectionContext(protocol_version=latest_proto)
  23         reactor = LoginReactor(connection)
  24
  25         packet = clientbound.login.EncryptionRequestPacket()
  26         packet.server_id = "123"
  27         packet.public_key = b"asdf"
  28         packet.verify_token = b"23"
  29
  30         secret = b"secret"
  31         encrypt.generate_shared_secret.return_value = secret
  32         encrypt.encrypt_token_and_secret.return_value = (b"a", b"b")
  33         encrypt.generate_verification_hash.return_value = b"hash"
  34
  35         reactor.react(packet)
  36
  37         encrypt.encrypt_token_and_secret.assert_called_once_with(
  38             packet.public_key, packet.verify_token, secret
  39         )
  40         connection.auth_token.join.assert_called_once_with(b"hash")
  41         self.assertEqual(connection.write_packet.call_count, 1)
  42
  43     @mock.patch('minecraft.networking.connection.encryption')
  44     def test_encryption_offline_server(self, encrypt):
  45         connection = mock.MagicMock()
  46         connection.context = ConnectionContext(protocol_version=latest_proto)
  47         reactor = LoginReactor(connection)
  48
  49         packet = clientbound.login.EncryptionRequestPacket()
  50         packet.server_id = "-"
  51         packet.public_key = b"asdf"
  52         packet.verify_token = b"23"
  53
  54         secret = b"secret"
  55         encrypt.generate_shared_secret.return_value = secret
  56         encrypt.encrypt_token_and_secret.return_value = (b"a", b"b")
  57
  58         reactor.react(packet)
  59
  60         encrypt.encrypt_token_and_secret.assert_called_once_with(
  61             packet.public_key, packet.verify_token, secret
  62         )
  63         self.assertEqual(connection.auth_token.join.call_count, 0)
  64         self.assertEqual(connection.write_packet.call_count, 1)
  65
  66
  67 class PlayingReactorTest(unittest.TestCase):
  68
  69     def get_position_packet(self):
  70         packet = clientbound.play.PlayerPositionAndLookPacket()
  71         packet.x = 1.0
  72         packet.y = 2.0
  73         packet.z = 3.0
  74         packet.yaw = 4.0
  75         packet.pitch = 5.0
  76
  77         packet.teleport_id = 42
  78
  79         return packet
  80
  81     def test_teleport_confirmation_old(self):
  82         connection = mock.MagicMock()
  83         connection.context = ConnectionContext(protocol_version=106)
  84         reactor = PlayingReactor(connection)
  85
  86         packet = self.get_position_packet()
  87         reactor.react(packet)
  88
  89         self.assertEqual(connection.write_packet.call_count, 1)
  90         response_packet = connection.write_packet.call_args[0][0]
  91
  92         self.assertEqual(response_packet.x, 1.0)
  93         self.assertEqual(response_packet.feet_y, 2.0)
  94         self.assertEqual(response_packet.z, 3.0)
  95         self.assertEqual(response_packet.yaw, 4.0)
  96         self.assertEqual(response_packet.pitch, 5.0)
  97         self.assertTrue(response_packet.on_ground)
  98
  99     def test_teleport_confirmation_new(self):
 100         connection = mock.MagicMock()
 101         connection.context = ConnectionContext(protocol_version=107)
 102         reactor = PlayingReactor(connection)
 103
 104         packet = self.get_position_packet()
 105         reactor.react(packet)
 106
 107         self.assertEqual(connection.write_packet.call_count, 1)
 108
 109         response_packet = connection.write_packet.call_args[0][0]
 110         self.assertEqual(response_packet.teleport_id, 42)