tests.py 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143
  1. #! /usr/bin/env python3
  2. #
  3. # Copyright (C) 2018 Garmin Ltd.
  4. #
  5. # SPDX-License-Identifier: GPL-2.0-only
  6. #
  7. # This program is free software; you can redistribute it and/or modify
  8. # it under the terms of the GNU General Public License version 2 as
  9. # published by the Free Software Foundation.
  10. #
  11. # This program is distributed in the hope that it will be useful,
  12. # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13. # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  14. # GNU General Public License for more details.
  15. #
  16. # You should have received a copy of the GNU General Public License along
  17. # with this program; if not, write to the Free Software Foundation, Inc.,
  18. # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  19. import unittest
  20. import threading
  21. import sqlite3
  22. import hashlib
  23. import urllib.request
  24. import json
  25. from . import create_server
  26. class TestHashEquivalenceServer(unittest.TestCase):
  27. def setUp(self):
  28. # Start an in memory hash equivalence server in the background bound to
  29. # an ephemeral port
  30. db = sqlite3.connect(':memory:', check_same_thread=False)
  31. self.server = create_server(('localhost', 0), db)
  32. self.server_addr = 'http://localhost:%d' % self.server.socket.getsockname()[1]
  33. self.server_thread = threading.Thread(target=self.server.serve_forever)
  34. self.server_thread.start()
  35. def tearDown(self):
  36. # Shutdown server
  37. s = getattr(self, 'server', None)
  38. if s is not None:
  39. self.server.shutdown()
  40. self.server_thread.join()
  41. self.server.server_close()
  42. def send_get(self, path):
  43. url = '%s/%s' % (self.server_addr, path)
  44. request = urllib.request.Request(url)
  45. response = urllib.request.urlopen(request)
  46. return json.loads(response.read().decode('utf-8'))
  47. def send_post(self, path, data):
  48. headers = {'content-type': 'application/json'}
  49. url = '%s/%s' % (self.server_addr, path)
  50. request = urllib.request.Request(url, json.dumps(data).encode('utf-8'), headers)
  51. response = urllib.request.urlopen(request)
  52. return json.loads(response.read().decode('utf-8'))
  53. def test_create_hash(self):
  54. # Simple test that hashes can be created
  55. taskhash = '35788efcb8dfb0a02659d81cf2bfd695fb30faf9'
  56. outhash = '2765d4a5884be49b28601445c2760c5f21e7e5c0ee2b7e3fce98fd7e5970796f'
  57. unihash = 'f46d3fbb439bd9b921095da657a4de906510d2cd'
  58. d = self.send_get('v1/equivalent?method=TestMethod&taskhash=%s' % taskhash)
  59. self.assertIsNone(d, msg='Found unexpected task, %r' % d)
  60. d = self.send_post('v1/equivalent', {
  61. 'taskhash': taskhash,
  62. 'method': 'TestMethod',
  63. 'outhash': outhash,
  64. 'unihash': unihash,
  65. })
  66. self.assertEqual(d['unihash'], unihash, 'Server returned bad unihash')
  67. def test_create_equivalent(self):
  68. # Tests that a second reported task with the same outhash will be
  69. # assigned the same unihash
  70. taskhash = '53b8dce672cb6d0c73170be43f540460bfc347b4'
  71. outhash = '5a9cb1649625f0bf41fc7791b635cd9c2d7118c7f021ba87dcd03f72b67ce7a8'
  72. unihash = 'f37918cc02eb5a520b1aff86faacbc0a38124646'
  73. d = self.send_post('v1/equivalent', {
  74. 'taskhash': taskhash,
  75. 'method': 'TestMethod',
  76. 'outhash': outhash,
  77. 'unihash': unihash,
  78. })
  79. self.assertEqual(d['unihash'], unihash, 'Server returned bad unihash')
  80. # Report a different task with the same outhash. The returned unihash
  81. # should match the first task
  82. taskhash2 = '3bf6f1e89d26205aec90da04854fbdbf73afe6b4'
  83. unihash2 = 'af36b199320e611fbb16f1f277d3ee1d619ca58b'
  84. d = self.send_post('v1/equivalent', {
  85. 'taskhash': taskhash2,
  86. 'method': 'TestMethod',
  87. 'outhash': outhash,
  88. 'unihash': unihash2,
  89. })
  90. self.assertEqual(d['unihash'], unihash, 'Server returned bad unihash')
  91. def test_duplicate_taskhash(self):
  92. # Tests that duplicate reports of the same taskhash with different
  93. # outhash & unihash always return the unihash from the first reported
  94. # taskhash
  95. taskhash = '8aa96fcffb5831b3c2c0cb75f0431e3f8b20554a'
  96. outhash = 'afe240a439959ce86f5e322f8c208e1fedefea9e813f2140c81af866cc9edf7e'
  97. unihash = '218e57509998197d570e2c98512d0105985dffc9'
  98. d = self.send_post('v1/equivalent', {
  99. 'taskhash': taskhash,
  100. 'method': 'TestMethod',
  101. 'outhash': outhash,
  102. 'unihash': unihash,
  103. })
  104. d = self.send_get('v1/equivalent?method=TestMethod&taskhash=%s' % taskhash)
  105. self.assertEqual(d['unihash'], unihash)
  106. outhash2 = '0904a7fe3dc712d9fd8a74a616ddca2a825a8ee97adf0bd3fc86082c7639914d'
  107. unihash2 = 'ae9a7d252735f0dafcdb10e2e02561ca3a47314c'
  108. d = self.send_post('v1/equivalent', {
  109. 'taskhash': taskhash,
  110. 'method': 'TestMethod',
  111. 'outhash': outhash2,
  112. 'unihash': unihash2
  113. })
  114. d = self.send_get('v1/equivalent?method=TestMethod&taskhash=%s' % taskhash)
  115. self.assertEqual(d['unihash'], unihash)
  116. outhash3 = '77623a549b5b1a31e3732dfa8fe61d7ce5d44b3370f253c5360e136b852967b4'
  117. unihash3 = '9217a7d6398518e5dc002ed58f2cbbbc78696603'
  118. d = self.send_post('v1/equivalent', {
  119. 'taskhash': taskhash,
  120. 'method': 'TestMethod',
  121. 'outhash': outhash3,
  122. 'unihash': unihash3
  123. })
  124. d = self.send_get('v1/equivalent?method=TestMethod&taskhash=%s' % taskhash)
  125. self.assertEqual(d['unihash'], unihash)