1
0

test_appservice.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530
  1. # Copyright 2015, 2016 OpenMarket Ltd
  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. import json
  15. import os
  16. import tempfile
  17. from typing import List, cast
  18. from unittest.mock import Mock
  19. import yaml
  20. from twisted.internet import defer
  21. from twisted.test.proto_helpers import MemoryReactor
  22. from synapse.appservice import ApplicationService, ApplicationServiceState
  23. from synapse.config._base import ConfigError
  24. from synapse.events import EventBase
  25. from synapse.server import HomeServer
  26. from synapse.storage.database import DatabasePool, LoggingDatabaseConnection, make_conn
  27. from synapse.storage.databases.main.appservice import (
  28. ApplicationServiceStore,
  29. ApplicationServiceTransactionStore,
  30. )
  31. from synapse.types import DeviceListUpdates
  32. from synapse.util import Clock
  33. from tests import unittest
  34. from tests.test_utils import make_awaitable
  35. class ApplicationServiceStoreTestCase(unittest.HomeserverTestCase):
  36. def setUp(self) -> None:
  37. super(ApplicationServiceStoreTestCase, self).setUp()
  38. self.as_yaml_files: List[str] = []
  39. self.hs.config.appservice.app_service_config_files = self.as_yaml_files
  40. self.hs.config.caches.event_cache_size = 1
  41. self.as_token = "token1"
  42. self.as_url = "some_url"
  43. self.as_id = "as1"
  44. self._add_appservice(
  45. self.as_token, self.as_id, self.as_url, "some_hs_token", "bob"
  46. )
  47. self._add_appservice("token2", "as2", "some_url", "some_hs_token", "bob")
  48. self._add_appservice("token3", "as3", "some_url", "some_hs_token", "bob")
  49. # must be done after inserts
  50. database = self.hs.get_datastores().databases[0]
  51. self.store = ApplicationServiceStore(
  52. database,
  53. make_conn(database._database_config, database.engine, "test"),
  54. self.hs,
  55. )
  56. def tearDown(self) -> None:
  57. # TODO: suboptimal that we need to create files for tests!
  58. for f in self.as_yaml_files:
  59. try:
  60. os.remove(f)
  61. except Exception:
  62. pass
  63. super(ApplicationServiceStoreTestCase, self).tearDown()
  64. def _add_appservice(
  65. self, as_token: str, id: str, url: str, hs_token: str, sender: str
  66. ) -> None:
  67. as_yaml = {
  68. "url": url,
  69. "as_token": as_token,
  70. "hs_token": hs_token,
  71. "id": id,
  72. "sender_localpart": sender,
  73. "namespaces": {},
  74. }
  75. # use the token as the filename
  76. with open(as_token, "w") as outfile:
  77. outfile.write(yaml.dump(as_yaml))
  78. self.as_yaml_files.append(as_token)
  79. def test_retrieve_unknown_service_token(self) -> None:
  80. service = self.store.get_app_service_by_token("invalid_token")
  81. self.assertEqual(service, None)
  82. def test_retrieval_of_service(self) -> None:
  83. stored_service = self.store.get_app_service_by_token(self.as_token)
  84. assert stored_service is not None
  85. self.assertEqual(stored_service.token, self.as_token)
  86. self.assertEqual(stored_service.id, self.as_id)
  87. self.assertEqual(stored_service.url, self.as_url)
  88. self.assertEqual(stored_service.namespaces[ApplicationService.NS_ALIASES], [])
  89. self.assertEqual(stored_service.namespaces[ApplicationService.NS_ROOMS], [])
  90. self.assertEqual(stored_service.namespaces[ApplicationService.NS_USERS], [])
  91. def test_retrieval_of_all_services(self) -> None:
  92. services = self.store.get_app_services()
  93. self.assertEqual(len(services), 3)
  94. class ApplicationServiceTransactionStoreTestCase(unittest.HomeserverTestCase):
  95. def setUp(self) -> None:
  96. super(ApplicationServiceTransactionStoreTestCase, self).setUp()
  97. self.as_yaml_files: List[str] = []
  98. self.hs.config.appservice.app_service_config_files = self.as_yaml_files
  99. self.hs.config.caches.event_cache_size = 1
  100. self.as_list = [
  101. {"token": "token1", "url": "https://matrix-as.org", "id": "id_1"},
  102. {"token": "alpha_tok", "url": "https://alpha.com", "id": "id_alpha"},
  103. {"token": "beta_tok", "url": "https://beta.com", "id": "id_beta"},
  104. {"token": "gamma_tok", "url": "https://gamma.com", "id": "id_gamma"},
  105. ]
  106. for s in self.as_list:
  107. self._add_service(s["url"], s["token"], s["id"])
  108. self.as_yaml_files = []
  109. # We assume there is only one database in these tests
  110. database = self.hs.get_datastores().databases[0]
  111. self.db_pool = database._db_pool
  112. self.engine = database.engine
  113. db_config = self.hs.config.database.get_single_database()
  114. self.store = TestTransactionStore(
  115. database, make_conn(db_config, self.engine, "test"), self.hs
  116. )
  117. def _add_service(self, url: str, as_token: str, id: str) -> None:
  118. as_yaml = {
  119. "url": url,
  120. "as_token": as_token,
  121. "hs_token": "something",
  122. "id": id,
  123. "sender_localpart": "a_sender",
  124. "namespaces": {},
  125. }
  126. # use the token as the filename
  127. with open(as_token, "w") as outfile:
  128. outfile.write(yaml.dump(as_yaml))
  129. self.as_yaml_files.append(as_token)
  130. def _set_state(self, id: str, state: ApplicationServiceState) -> defer.Deferred:
  131. return self.db_pool.runOperation(
  132. self.engine.convert_param_style(
  133. "INSERT INTO application_services_state(as_id, state) VALUES(?,?)"
  134. ),
  135. (id, state.value),
  136. )
  137. def _insert_txn(
  138. self, as_id: str, txn_id: int, events: List[Mock]
  139. ) -> "defer.Deferred[None]":
  140. return self.db_pool.runOperation(
  141. self.engine.convert_param_style(
  142. "INSERT INTO application_services_txns(as_id, txn_id, event_ids) "
  143. "VALUES(?,?,?)"
  144. ),
  145. (as_id, txn_id, json.dumps([e.event_id for e in events])),
  146. )
  147. def test_get_appservice_state_none(
  148. self,
  149. ) -> None:
  150. service = Mock(id="999")
  151. state = self.get_success(self.store.get_appservice_state(service))
  152. self.assertEqual(None, state)
  153. def test_get_appservice_state_up(
  154. self,
  155. ) -> None:
  156. self.get_success(
  157. self._set_state(self.as_list[0]["id"], ApplicationServiceState.UP)
  158. )
  159. service = Mock(id=self.as_list[0]["id"])
  160. state = self.get_success(
  161. defer.ensureDeferred(self.store.get_appservice_state(service))
  162. )
  163. self.assertEqual(ApplicationServiceState.UP, state)
  164. def test_get_appservice_state_down(
  165. self,
  166. ) -> None:
  167. self.get_success(
  168. self._set_state(self.as_list[0]["id"], ApplicationServiceState.UP)
  169. )
  170. self.get_success(
  171. self._set_state(self.as_list[1]["id"], ApplicationServiceState.DOWN)
  172. )
  173. self.get_success(
  174. self._set_state(self.as_list[2]["id"], ApplicationServiceState.DOWN)
  175. )
  176. service = Mock(id=self.as_list[1]["id"])
  177. state = self.get_success(self.store.get_appservice_state(service))
  178. self.assertEqual(ApplicationServiceState.DOWN, state)
  179. def test_get_appservices_by_state_none(
  180. self,
  181. ) -> None:
  182. services = self.get_success(
  183. self.store.get_appservices_by_state(ApplicationServiceState.DOWN)
  184. )
  185. self.assertEqual(0, len(services))
  186. def test_set_appservices_state_down(
  187. self,
  188. ) -> None:
  189. service = Mock(id=self.as_list[1]["id"])
  190. self.get_success(
  191. self.store.set_appservice_state(service, ApplicationServiceState.DOWN)
  192. )
  193. rows = self.get_success(
  194. self.db_pool.runQuery(
  195. self.engine.convert_param_style(
  196. "SELECT as_id FROM application_services_state WHERE state=?"
  197. ),
  198. (ApplicationServiceState.DOWN.value,),
  199. )
  200. )
  201. self.assertEqual(service.id, rows[0][0])
  202. def test_set_appservices_state_multiple_up(
  203. self,
  204. ) -> None:
  205. service = Mock(id=self.as_list[1]["id"])
  206. self.get_success(
  207. self.store.set_appservice_state(service, ApplicationServiceState.UP)
  208. )
  209. self.get_success(
  210. self.store.set_appservice_state(service, ApplicationServiceState.DOWN)
  211. )
  212. self.get_success(
  213. self.store.set_appservice_state(service, ApplicationServiceState.UP)
  214. )
  215. rows = self.get_success(
  216. self.db_pool.runQuery(
  217. self.engine.convert_param_style(
  218. "SELECT as_id FROM application_services_state WHERE state=?"
  219. ),
  220. (ApplicationServiceState.UP.value,),
  221. )
  222. )
  223. self.assertEqual(service.id, rows[0][0])
  224. def test_create_appservice_txn_first(
  225. self,
  226. ) -> None:
  227. service = Mock(id=self.as_list[0]["id"])
  228. events = cast(List[EventBase], [Mock(event_id="e1"), Mock(event_id="e2")])
  229. txn = self.get_success(
  230. defer.ensureDeferred(
  231. self.store.create_appservice_txn(
  232. service, events, [], [], {}, {}, DeviceListUpdates()
  233. )
  234. )
  235. )
  236. self.assertEqual(txn.id, 1)
  237. self.assertEqual(txn.events, events)
  238. self.assertEqual(txn.service, service)
  239. def test_complete_appservice_txn_first_txn(
  240. self,
  241. ) -> None:
  242. service = Mock(id=self.as_list[0]["id"])
  243. events = [Mock(event_id="e1"), Mock(event_id="e2")]
  244. txn_id = 1
  245. self.get_success(self._insert_txn(service.id, txn_id, events))
  246. self.get_success(
  247. self.store.complete_appservice_txn(txn_id=txn_id, service=service)
  248. )
  249. res = self.get_success(
  250. self.db_pool.runQuery(
  251. self.engine.convert_param_style(
  252. "SELECT * FROM application_services_txns WHERE txn_id=?"
  253. ),
  254. (txn_id,),
  255. )
  256. )
  257. self.assertEqual(0, len(res))
  258. def test_complete_appservice_txn_updates_last_txn_state(
  259. self,
  260. ) -> None:
  261. service = Mock(id=self.as_list[0]["id"])
  262. events = [Mock(event_id="e1"), Mock(event_id="e2")]
  263. txn_id = 5
  264. self._set_state(self.as_list[0]["id"], ApplicationServiceState.UP)
  265. self.get_success(self._insert_txn(service.id, txn_id, events))
  266. self.get_success(
  267. self.store.complete_appservice_txn(txn_id=txn_id, service=service)
  268. )
  269. res = self.get_success(
  270. self.db_pool.runQuery(
  271. self.engine.convert_param_style(
  272. "SELECT state FROM application_services_state WHERE as_id=?"
  273. ),
  274. (service.id,),
  275. )
  276. )
  277. self.assertEqual(1, len(res))
  278. self.assertEqual(ApplicationServiceState.UP.value, res[0][0])
  279. res = self.get_success(
  280. self.db_pool.runQuery(
  281. self.engine.convert_param_style(
  282. "SELECT * FROM application_services_txns WHERE txn_id=?"
  283. ),
  284. (txn_id,),
  285. )
  286. )
  287. self.assertEqual(0, len(res))
  288. def test_get_oldest_unsent_txn_none(
  289. self,
  290. ) -> None:
  291. service = Mock(id=self.as_list[0]["id"])
  292. txn = self.get_success(self.store.get_oldest_unsent_txn(service))
  293. self.assertEqual(None, txn)
  294. def test_get_oldest_unsent_txn(self) -> None:
  295. service = Mock(id=self.as_list[0]["id"])
  296. events = [Mock(event_id="e1"), Mock(event_id="e2")]
  297. other_events = [Mock(event_id="e5"), Mock(event_id="e6")]
  298. # we aren't testing store._base stuff here, so mock this out
  299. # (ignore needed because Mypy won't allow us to assign to a method otherwise)
  300. self.store.get_events_as_list = Mock(return_value=make_awaitable(events)) # type: ignore[assignment]
  301. self.get_success(self._insert_txn(self.as_list[1]["id"], 9, other_events))
  302. self.get_success(self._insert_txn(service.id, 10, events))
  303. self.get_success(self._insert_txn(service.id, 11, other_events))
  304. self.get_success(self._insert_txn(service.id, 12, other_events))
  305. txn = self.get_success(self.store.get_oldest_unsent_txn(service))
  306. assert txn is not None
  307. self.assertEqual(service, txn.service)
  308. self.assertEqual(10, txn.id)
  309. self.assertEqual(events, txn.events)
  310. def test_get_appservices_by_state_single(
  311. self,
  312. ) -> None:
  313. self.get_success(
  314. self._set_state(self.as_list[0]["id"], ApplicationServiceState.DOWN)
  315. )
  316. self.get_success(
  317. self._set_state(self.as_list[1]["id"], ApplicationServiceState.UP)
  318. )
  319. services = self.get_success(
  320. self.store.get_appservices_by_state(ApplicationServiceState.DOWN)
  321. )
  322. self.assertEqual(1, len(services))
  323. self.assertEqual(self.as_list[0]["id"], services[0].id)
  324. def test_get_appservices_by_state_multiple(
  325. self,
  326. ) -> None:
  327. self.get_success(
  328. self._set_state(self.as_list[0]["id"], ApplicationServiceState.DOWN)
  329. )
  330. self.get_success(
  331. self._set_state(self.as_list[1]["id"], ApplicationServiceState.UP)
  332. )
  333. self.get_success(
  334. self._set_state(self.as_list[2]["id"], ApplicationServiceState.DOWN)
  335. )
  336. self.get_success(
  337. self._set_state(self.as_list[3]["id"], ApplicationServiceState.UP)
  338. )
  339. services = self.get_success(
  340. self.store.get_appservices_by_state(ApplicationServiceState.DOWN)
  341. )
  342. self.assertEqual(2, len(services))
  343. self.assertEqual(
  344. {self.as_list[2]["id"], self.as_list[0]["id"]},
  345. {services[0].id, services[1].id},
  346. )
  347. class ApplicationServiceStoreTypeStreamIds(unittest.HomeserverTestCase):
  348. def prepare(
  349. self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
  350. ) -> None:
  351. self.service = Mock(id="foo")
  352. self.store = self.hs.get_datastores().main
  353. self.get_success(
  354. self.store.set_appservice_state(self.service, ApplicationServiceState.UP)
  355. )
  356. def test_get_type_stream_id_for_appservice_no_value(self) -> None:
  357. value = self.get_success(
  358. self.store.get_type_stream_id_for_appservice(self.service, "read_receipt")
  359. )
  360. self.assertEqual(value, 1)
  361. value = self.get_success(
  362. self.store.get_type_stream_id_for_appservice(self.service, "presence")
  363. )
  364. self.assertEqual(value, 1)
  365. def test_get_type_stream_id_for_appservice_invalid_type(self) -> None:
  366. self.get_failure(
  367. self.store.get_type_stream_id_for_appservice(self.service, "foobar"),
  368. ValueError,
  369. )
  370. def test_set_appservice_stream_type_pos(self) -> None:
  371. read_receipt_value = 1024
  372. self.get_success(
  373. self.store.set_appservice_stream_type_pos(
  374. self.service, "read_receipt", read_receipt_value
  375. )
  376. )
  377. result = self.get_success(
  378. self.store.get_type_stream_id_for_appservice(self.service, "read_receipt")
  379. )
  380. self.assertEqual(result, read_receipt_value)
  381. self.get_success(
  382. self.store.set_appservice_stream_type_pos(
  383. self.service, "presence", read_receipt_value
  384. )
  385. )
  386. result = self.get_success(
  387. self.store.get_type_stream_id_for_appservice(self.service, "presence")
  388. )
  389. self.assertEqual(result, read_receipt_value)
  390. def test_set_appservice_stream_type_pos_invalid_type(self) -> None:
  391. self.get_failure(
  392. self.store.set_appservice_stream_type_pos(self.service, "foobar", 1024),
  393. ValueError,
  394. )
  395. # required for ApplicationServiceTransactionStoreTestCase tests
  396. class TestTransactionStore(ApplicationServiceTransactionStore, ApplicationServiceStore):
  397. def __init__(
  398. self, database: DatabasePool, db_conn: LoggingDatabaseConnection, hs: HomeServer
  399. ) -> None:
  400. super().__init__(database, db_conn, hs)
  401. class ApplicationServiceStoreConfigTestCase(unittest.HomeserverTestCase):
  402. def _write_config(self, suffix: str, **kwargs: str) -> str:
  403. vals = {
  404. "id": "id" + suffix,
  405. "url": "url" + suffix,
  406. "as_token": "as_token" + suffix,
  407. "hs_token": "hs_token" + suffix,
  408. "sender_localpart": "sender_localpart" + suffix,
  409. "namespaces": {},
  410. }
  411. vals.update(kwargs)
  412. _, path = tempfile.mkstemp(prefix="as_config")
  413. with open(path, "w") as f:
  414. f.write(yaml.dump(vals))
  415. return path
  416. def test_unique_works(self) -> None:
  417. f1 = self._write_config(suffix="1")
  418. f2 = self._write_config(suffix="2")
  419. self.hs.config.appservice.app_service_config_files = [f1, f2]
  420. self.hs.config.caches.event_cache_size = 1
  421. database = self.hs.get_datastores().databases[0]
  422. ApplicationServiceStore(
  423. database,
  424. make_conn(database._database_config, database.engine, "test"),
  425. self.hs,
  426. )
  427. def test_duplicate_ids(self) -> None:
  428. f1 = self._write_config(id="id", suffix="1")
  429. f2 = self._write_config(id="id", suffix="2")
  430. self.hs.config.appservice.app_service_config_files = [f1, f2]
  431. self.hs.config.caches.event_cache_size = 1
  432. with self.assertRaises(ConfigError) as cm:
  433. database = self.hs.get_datastores().databases[0]
  434. ApplicationServiceStore(
  435. database,
  436. make_conn(database._database_config, database.engine, "test"),
  437. self.hs,
  438. )
  439. e = cm.exception
  440. self.assertIn(f1, str(e))
  441. self.assertIn(f2, str(e))
  442. self.assertIn("id", str(e))
  443. def test_duplicate_as_tokens(self) -> None:
  444. f1 = self._write_config(as_token="as_token", suffix="1")
  445. f2 = self._write_config(as_token="as_token", suffix="2")
  446. self.hs.config.appservice.app_service_config_files = [f1, f2]
  447. self.hs.config.caches.event_cache_size = 1
  448. with self.assertRaises(ConfigError) as cm:
  449. database = self.hs.get_datastores().databases[0]
  450. ApplicationServiceStore(
  451. database,
  452. make_conn(database._database_config, database.engine, "test"),
  453. self.hs,
  454. )
  455. e = cm.exception
  456. self.assertIn(f1, str(e))
  457. self.assertIn(f2, str(e))
  458. self.assertIn("as_token", str(e))