1
0

test_commands.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. # Copyright 2020 The Matrix.org Foundation C.I.C.
  2. #
  3. # Licensed under the Apache License, Version 2.0 (the "License");
  4. # you may not use this file except in compliance with the License.
  5. # You may obtain a copy of the License at
  6. #
  7. # http://www.apache.org/licenses/LICENSE-2.0
  8. #
  9. # Unless required by applicable law or agreed to in writing, software
  10. # distributed under the License is distributed on an "AS IS" BASIS,
  11. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. # See the License for the specific language governing permissions and
  13. # limitations under the License.
  14. from synapse.replication.tcp.commands import (
  15. RdataCommand,
  16. ReplicateCommand,
  17. parse_command_from_line,
  18. )
  19. from tests.unittest import TestCase
  20. class ParseCommandTestCase(TestCase):
  21. def test_parse_one_word_command(self) -> None:
  22. line = "REPLICATE"
  23. cmd = parse_command_from_line(line)
  24. self.assertIsInstance(cmd, ReplicateCommand)
  25. def test_parse_rdata(self) -> None:
  26. line = 'RDATA events master 6287863 ["ev", ["$eventid", "!roomid", "type", null, null, null]]'
  27. cmd = parse_command_from_line(line)
  28. assert isinstance(cmd, RdataCommand)
  29. self.assertEqual(cmd.stream_name, "events")
  30. self.assertEqual(cmd.instance_name, "master")
  31. self.assertEqual(cmd.token, 6287863)
  32. def test_parse_rdata_batch(self) -> None:
  33. line = 'RDATA presence master batch ["@foo:example.com", "online"]'
  34. cmd = parse_command_from_line(line)
  35. assert isinstance(cmd, RdataCommand)
  36. self.assertEqual(cmd.stream_name, "presence")
  37. self.assertEqual(cmd.instance_name, "master")
  38. self.assertIsNone(cmd.token)