test_appservice.py 19 KB

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