1
0

server.py 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904
  1. # Copyright 2018-2021 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. import hashlib
  15. import json
  16. import logging
  17. import os
  18. import os.path
  19. import time
  20. import uuid
  21. import warnings
  22. from collections import deque
  23. from io import SEEK_END, BytesIO
  24. from typing import (
  25. Callable,
  26. Dict,
  27. Iterable,
  28. MutableMapping,
  29. Optional,
  30. Tuple,
  31. Type,
  32. Union,
  33. )
  34. from unittest.mock import Mock
  35. import attr
  36. from typing_extensions import Deque
  37. from zope.interface import implementer
  38. from twisted.internet import address, threads, udp
  39. from twisted.internet._resolver import SimpleResolverComplexifier
  40. from twisted.internet.defer import Deferred, fail, maybeDeferred, succeed
  41. from twisted.internet.error import DNSLookupError
  42. from twisted.internet.interfaces import (
  43. IAddress,
  44. IHostnameResolver,
  45. IProtocol,
  46. IPullProducer,
  47. IPushProducer,
  48. IReactorPluggableNameResolver,
  49. IReactorTime,
  50. IResolverSimple,
  51. ITransport,
  52. )
  53. from twisted.python.failure import Failure
  54. from twisted.test.proto_helpers import (
  55. AccumulatingProtocol,
  56. MemoryReactor,
  57. MemoryReactorClock,
  58. )
  59. from twisted.web.http_headers import Headers
  60. from twisted.web.resource import IResource
  61. from twisted.web.server import Request, Site
  62. from synapse.config.database import DatabaseConnectionConfig
  63. from synapse.http.site import SynapseRequest
  64. from synapse.logging.context import ContextResourceUsage
  65. from synapse.server import HomeServer
  66. from synapse.storage import DataStore
  67. from synapse.storage.engines import PostgresEngine, create_engine
  68. from synapse.types import JsonDict
  69. from synapse.util import Clock
  70. from tests.utils import (
  71. LEAVE_DB,
  72. POSTGRES_BASE_DB,
  73. POSTGRES_HOST,
  74. POSTGRES_PASSWORD,
  75. POSTGRES_PORT,
  76. POSTGRES_USER,
  77. SQLITE_PERSIST_DB,
  78. USE_POSTGRES_FOR_TESTS,
  79. MockClock,
  80. default_config,
  81. )
  82. logger = logging.getLogger(__name__)
  83. # the type of thing that can be passed into `make_request` in the headers list
  84. CustomHeaderType = Tuple[Union[str, bytes], Union[str, bytes]]
  85. class TimedOutException(Exception):
  86. """
  87. A web query timed out.
  88. """
  89. @attr.s(auto_attribs=True)
  90. class FakeChannel:
  91. """
  92. A fake Twisted Web Channel (the part that interfaces with the
  93. wire).
  94. """
  95. site: Union[Site, "FakeSite"]
  96. _reactor: MemoryReactor
  97. result: dict = attr.Factory(dict)
  98. _ip: str = "127.0.0.1"
  99. _producer: Optional[Union[IPullProducer, IPushProducer]] = None
  100. resource_usage: Optional[ContextResourceUsage] = None
  101. _request: Optional[Request] = None
  102. @property
  103. def request(self) -> Request:
  104. assert self._request is not None
  105. return self._request
  106. @request.setter
  107. def request(self, request: Request) -> None:
  108. assert self._request is None
  109. self._request = request
  110. @property
  111. def json_body(self):
  112. return json.loads(self.text_body)
  113. @property
  114. def text_body(self) -> str:
  115. """The body of the result, utf-8-decoded.
  116. Raises an exception if the request has not yet completed.
  117. """
  118. if not self.is_finished:
  119. raise Exception("Request not yet completed")
  120. return self.result["body"].decode("utf8")
  121. def is_finished(self) -> bool:
  122. """check if the response has been completely received"""
  123. return self.result.get("done", False)
  124. @property
  125. def code(self):
  126. if not self.result:
  127. raise Exception("No result yet.")
  128. return int(self.result["code"])
  129. @property
  130. def headers(self) -> Headers:
  131. if not self.result:
  132. raise Exception("No result yet.")
  133. h = Headers()
  134. for i in self.result["headers"]:
  135. h.addRawHeader(*i)
  136. return h
  137. def writeHeaders(self, version, code, reason, headers):
  138. self.result["version"] = version
  139. self.result["code"] = code
  140. self.result["reason"] = reason
  141. self.result["headers"] = headers
  142. def write(self, content):
  143. assert isinstance(content, bytes), "Should be bytes! " + repr(content)
  144. if "body" not in self.result:
  145. self.result["body"] = b""
  146. self.result["body"] += content
  147. def registerProducer(self, producer, streaming):
  148. self._producer = producer
  149. self.producerStreaming = streaming
  150. def _produce():
  151. if self._producer:
  152. self._producer.resumeProducing()
  153. self._reactor.callLater(0.1, _produce)
  154. if not streaming:
  155. self._reactor.callLater(0.0, _produce)
  156. def unregisterProducer(self):
  157. if self._producer is None:
  158. return
  159. self._producer = None
  160. def requestDone(self, _self):
  161. self.result["done"] = True
  162. if isinstance(_self, SynapseRequest):
  163. self.resource_usage = _self.logcontext.get_resource_usage()
  164. def getPeer(self):
  165. # We give an address so that getClientAddress/getClientIP returns a non null entry,
  166. # causing us to record the MAU
  167. return address.IPv4Address("TCP", self._ip, 3423)
  168. def getHost(self):
  169. # this is called by Request.__init__ to configure Request.host.
  170. return address.IPv4Address("TCP", "127.0.0.1", 8888)
  171. def isSecure(self):
  172. return False
  173. @property
  174. def transport(self):
  175. return self
  176. def await_result(self, timeout_ms: int = 1000) -> None:
  177. """
  178. Wait until the request is finished.
  179. """
  180. end_time = self._reactor.seconds() + timeout_ms / 1000.0
  181. self._reactor.run()
  182. while not self.is_finished():
  183. # If there's a producer, tell it to resume producing so we get content
  184. if self._producer:
  185. self._producer.resumeProducing()
  186. if self._reactor.seconds() > end_time:
  187. raise TimedOutException("Timed out waiting for request to finish.")
  188. self._reactor.advance(0.1)
  189. def extract_cookies(self, cookies: MutableMapping[str, str]) -> None:
  190. """Process the contents of any Set-Cookie headers in the response
  191. Any cookines found are added to the given dict
  192. """
  193. headers = self.headers.getRawHeaders("Set-Cookie")
  194. if not headers:
  195. return
  196. for h in headers:
  197. parts = h.split(";")
  198. k, v = parts[0].split("=", maxsplit=1)
  199. cookies[k] = v
  200. class FakeSite:
  201. """
  202. A fake Twisted Web Site, with mocks of the extra things that
  203. Synapse adds.
  204. """
  205. server_version_string = b"1"
  206. site_tag = "test"
  207. access_logger = logging.getLogger("synapse.access.http.fake")
  208. def __init__(self, resource: IResource, reactor: IReactorTime):
  209. """
  210. Args:
  211. resource: the resource to be used for rendering all requests
  212. """
  213. self._resource = resource
  214. self.reactor = reactor
  215. def getResourceFor(self, request):
  216. return self._resource
  217. def make_request(
  218. reactor,
  219. site: Union[Site, FakeSite],
  220. method: Union[bytes, str],
  221. path: Union[bytes, str],
  222. content: Union[bytes, str, JsonDict] = b"",
  223. access_token: Optional[str] = None,
  224. request: Type[Request] = SynapseRequest,
  225. shorthand: bool = True,
  226. federation_auth_origin: Optional[bytes] = None,
  227. content_is_form: bool = False,
  228. await_result: bool = True,
  229. custom_headers: Optional[Iterable[CustomHeaderType]] = None,
  230. client_ip: str = "127.0.0.1",
  231. ) -> FakeChannel:
  232. """
  233. Make a web request using the given method, path and content, and render it
  234. Returns the fake Channel object which records the response to the request.
  235. Args:
  236. reactor:
  237. site: The twisted Site to use to render the request
  238. method: The HTTP request method ("verb").
  239. path: The HTTP path, suitably URL encoded (e.g. escaped UTF-8 & spaces and such).
  240. content: The body of the request. JSON-encoded, if a str of bytes.
  241. access_token: The access token to add as authorization for the request.
  242. request: The request class to create.
  243. shorthand: Whether to try and be helpful and prefix the given URL
  244. with the usual REST API path, if it doesn't contain it.
  245. federation_auth_origin: if set to not-None, we will add a fake
  246. Authorization header pretenting to be the given server name.
  247. content_is_form: Whether the content is URL encoded form data. Adds the
  248. 'Content-Type': 'application/x-www-form-urlencoded' header.
  249. await_result: whether to wait for the request to complete rendering. If true,
  250. will pump the reactor until the the renderer tells the channel the request
  251. is finished.
  252. custom_headers: (name, value) pairs to add as request headers
  253. client_ip: The IP to use as the requesting IP. Useful for testing
  254. ratelimiting.
  255. Returns:
  256. channel
  257. """
  258. if not isinstance(method, bytes):
  259. method = method.encode("ascii")
  260. if not isinstance(path, bytes):
  261. path = path.encode("ascii")
  262. # Decorate it to be the full path, if we're using shorthand
  263. if (
  264. shorthand
  265. and not path.startswith(b"/_matrix")
  266. and not path.startswith(b"/_synapse")
  267. ):
  268. if path.startswith(b"/"):
  269. path = path[1:]
  270. path = b"/_matrix/client/r0/" + path
  271. if not path.startswith(b"/"):
  272. path = b"/" + path
  273. if isinstance(content, dict):
  274. content = json.dumps(content).encode("utf8")
  275. if isinstance(content, str):
  276. content = content.encode("utf8")
  277. channel = FakeChannel(site, reactor, ip=client_ip)
  278. req = request(channel, site)
  279. channel.request = req
  280. req.content = BytesIO(content)
  281. # Twisted expects to be at the end of the content when parsing the request.
  282. req.content.seek(0, SEEK_END)
  283. if access_token:
  284. req.requestHeaders.addRawHeader(
  285. b"Authorization", b"Bearer " + access_token.encode("ascii")
  286. )
  287. if federation_auth_origin is not None:
  288. req.requestHeaders.addRawHeader(
  289. b"Authorization",
  290. b"X-Matrix origin=%s,key=,sig=" % (federation_auth_origin,),
  291. )
  292. if content:
  293. if content_is_form:
  294. req.requestHeaders.addRawHeader(
  295. b"Content-Type", b"application/x-www-form-urlencoded"
  296. )
  297. else:
  298. # Assume the body is JSON
  299. req.requestHeaders.addRawHeader(b"Content-Type", b"application/json")
  300. if custom_headers:
  301. for k, v in custom_headers:
  302. req.requestHeaders.addRawHeader(k, v)
  303. req.parseCookies()
  304. req.requestReceived(method, path, b"1.1")
  305. if await_result:
  306. channel.await_result()
  307. return channel
  308. @implementer(IReactorPluggableNameResolver)
  309. class ThreadedMemoryReactorClock(MemoryReactorClock):
  310. """
  311. A MemoryReactorClock that supports callFromThread.
  312. """
  313. def __init__(self):
  314. self.threadpool = ThreadPool(self)
  315. self._tcp_callbacks: Dict[Tuple[str, int], Callable] = {}
  316. self._udp = []
  317. self.lookups: Dict[str, str] = {}
  318. self._thread_callbacks: Deque[Callable[[], None]] = deque()
  319. lookups = self.lookups
  320. @implementer(IResolverSimple)
  321. class FakeResolver:
  322. def getHostByName(self, name, timeout=None):
  323. if name not in lookups:
  324. return fail(DNSLookupError("OH NO: unknown %s" % (name,)))
  325. return succeed(lookups[name])
  326. self.nameResolver = SimpleResolverComplexifier(FakeResolver())
  327. super().__init__()
  328. def installNameResolver(self, resolver: IHostnameResolver) -> IHostnameResolver:
  329. raise NotImplementedError()
  330. def listenUDP(self, port, protocol, interface="", maxPacketSize=8196):
  331. p = udp.Port(port, protocol, interface, maxPacketSize, self)
  332. p.startListening()
  333. self._udp.append(p)
  334. return p
  335. def callFromThread(self, callback, *args, **kwargs):
  336. """
  337. Make the callback fire in the next reactor iteration.
  338. """
  339. cb = lambda: callback(*args, **kwargs)
  340. # it's not safe to call callLater() here, so we append the callback to a
  341. # separate queue.
  342. self._thread_callbacks.append(cb)
  343. def getThreadPool(self):
  344. return self.threadpool
  345. def add_tcp_client_callback(self, host: str, port: int, callback: Callable):
  346. """Add a callback that will be invoked when we receive a connection
  347. attempt to the given IP/port using `connectTCP`.
  348. Note that the callback gets run before we return the connection to the
  349. client, which means callbacks cannot block while waiting for writes.
  350. """
  351. self._tcp_callbacks[(host, port)] = callback
  352. def connectTCP(self, host: str, port: int, factory, timeout=30, bindAddress=None):
  353. """Fake L{IReactorTCP.connectTCP}."""
  354. conn = super().connectTCP(
  355. host, port, factory, timeout=timeout, bindAddress=None
  356. )
  357. callback = self._tcp_callbacks.get((host, port))
  358. if callback:
  359. callback()
  360. return conn
  361. def advance(self, amount):
  362. # first advance our reactor's time, and run any "callLater" callbacks that
  363. # makes ready
  364. super().advance(amount)
  365. # now run any "callFromThread" callbacks
  366. while True:
  367. try:
  368. callback = self._thread_callbacks.popleft()
  369. except IndexError:
  370. break
  371. callback()
  372. # check for more "callLater" callbacks added by the thread callback
  373. # This isn't required in a regular reactor, but it ends up meaning that
  374. # our database queries can complete in a single call to `advance` [1] which
  375. # simplifies tests.
  376. #
  377. # [1]: we replace the threadpool backing the db connection pool with a
  378. # mock ThreadPool which doesn't really use threads; but we still use
  379. # reactor.callFromThread to feed results back from the db functions to the
  380. # main thread.
  381. super().advance(0)
  382. class ThreadPool:
  383. """
  384. Threadless thread pool.
  385. """
  386. def __init__(self, reactor):
  387. self._reactor = reactor
  388. def start(self):
  389. pass
  390. def stop(self):
  391. pass
  392. def callInThreadWithCallback(self, onResult, function, *args, **kwargs):
  393. def _(res):
  394. if isinstance(res, Failure):
  395. onResult(False, res)
  396. else:
  397. onResult(True, res)
  398. d = Deferred()
  399. d.addCallback(lambda x: function(*args, **kwargs))
  400. d.addBoth(_)
  401. self._reactor.callLater(0, d.callback, True)
  402. return d
  403. def _make_test_homeserver_synchronous(server: HomeServer) -> None:
  404. """
  405. Make the given test homeserver's database interactions synchronous.
  406. """
  407. clock = server.get_clock()
  408. for database in server.get_datastores().databases:
  409. pool = database._db_pool
  410. def runWithConnection(func, *args, **kwargs):
  411. return threads.deferToThreadPool(
  412. pool._reactor,
  413. pool.threadpool,
  414. pool._runWithConnection,
  415. func,
  416. *args,
  417. **kwargs,
  418. )
  419. def runInteraction(interaction, *args, **kwargs):
  420. return threads.deferToThreadPool(
  421. pool._reactor,
  422. pool.threadpool,
  423. pool._runInteraction,
  424. interaction,
  425. *args,
  426. **kwargs,
  427. )
  428. pool.runWithConnection = runWithConnection
  429. pool.runInteraction = runInteraction
  430. # Replace the thread pool with a threadless 'thread' pool
  431. pool.threadpool = ThreadPool(clock._reactor)
  432. pool.running = True
  433. # We've just changed the Databases to run DB transactions on the same
  434. # thread, so we need to disable the dedicated thread behaviour.
  435. server.get_datastores().main.USE_DEDICATED_DB_THREADS_FOR_EVENT_FETCHING = False
  436. def get_clock() -> Tuple[ThreadedMemoryReactorClock, Clock]:
  437. clock = ThreadedMemoryReactorClock()
  438. hs_clock = Clock(clock)
  439. return clock, hs_clock
  440. @implementer(ITransport)
  441. @attr.s(cmp=False)
  442. class FakeTransport:
  443. """
  444. A twisted.internet.interfaces.ITransport implementation which sends all its data
  445. straight into an IProtocol object: it exists to connect two IProtocols together.
  446. To use it, instantiate it with the receiving IProtocol, and then pass it to the
  447. sending IProtocol's makeConnection method:
  448. server = HTTPChannel()
  449. client.makeConnection(FakeTransport(server, self.reactor))
  450. If you want bidirectional communication, you'll need two instances.
  451. """
  452. other = attr.ib()
  453. """The Protocol object which will receive any data written to this transport.
  454. :type: twisted.internet.interfaces.IProtocol
  455. """
  456. _reactor = attr.ib()
  457. """Test reactor
  458. :type: twisted.internet.interfaces.IReactorTime
  459. """
  460. _protocol = attr.ib(default=None)
  461. """The Protocol which is producing data for this transport. Optional, but if set
  462. will get called back for connectionLost() notifications etc.
  463. """
  464. _peer_address: Optional[IAddress] = attr.ib(default=None)
  465. """The value to be returned by getPeer"""
  466. _host_address: Optional[IAddress] = attr.ib(default=None)
  467. """The value to be returned by getHost"""
  468. disconnecting = False
  469. disconnected = False
  470. connected = True
  471. buffer = attr.ib(default=b"")
  472. producer = attr.ib(default=None)
  473. autoflush = attr.ib(default=True)
  474. def getPeer(self) -> Optional[IAddress]:
  475. return self._peer_address
  476. def getHost(self) -> Optional[IAddress]:
  477. return self._host_address
  478. def loseConnection(self, reason=None):
  479. if not self.disconnecting:
  480. logger.info("FakeTransport: loseConnection(%s)", reason)
  481. self.disconnecting = True
  482. if self._protocol:
  483. self._protocol.connectionLost(reason)
  484. # if we still have data to write, delay until that is done
  485. if self.buffer:
  486. logger.info(
  487. "FakeTransport: Delaying disconnect until buffer is flushed"
  488. )
  489. else:
  490. self.connected = False
  491. self.disconnected = True
  492. def abortConnection(self):
  493. logger.info("FakeTransport: abortConnection()")
  494. if not self.disconnecting:
  495. self.disconnecting = True
  496. if self._protocol:
  497. self._protocol.connectionLost(None)
  498. self.disconnected = True
  499. def pauseProducing(self):
  500. if not self.producer:
  501. return
  502. self.producer.pauseProducing()
  503. def resumeProducing(self):
  504. if not self.producer:
  505. return
  506. self.producer.resumeProducing()
  507. def unregisterProducer(self):
  508. if not self.producer:
  509. return
  510. self.producer = None
  511. def registerProducer(self, producer, streaming):
  512. self.producer = producer
  513. self.producerStreaming = streaming
  514. def _produce():
  515. if not self.producer:
  516. # we've been unregistered
  517. return
  518. # some implementations of IProducer (for example, FileSender)
  519. # don't return a deferred.
  520. d = maybeDeferred(self.producer.resumeProducing)
  521. d.addCallback(lambda x: self._reactor.callLater(0.1, _produce))
  522. if not streaming:
  523. self._reactor.callLater(0.0, _produce)
  524. def write(self, byt):
  525. if self.disconnecting:
  526. raise Exception("Writing to disconnecting FakeTransport")
  527. self.buffer = self.buffer + byt
  528. # always actually do the write asynchronously. Some protocols (notably the
  529. # TLSMemoryBIOProtocol) get very confused if a read comes back while they are
  530. # still doing a write. Doing a callLater here breaks the cycle.
  531. if self.autoflush:
  532. self._reactor.callLater(0.0, self.flush)
  533. def writeSequence(self, seq):
  534. for x in seq:
  535. self.write(x)
  536. def flush(self, maxbytes=None):
  537. if not self.buffer:
  538. # nothing to do. Don't write empty buffers: it upsets the
  539. # TLSMemoryBIOProtocol
  540. return
  541. if self.disconnected:
  542. return
  543. if maxbytes is not None:
  544. to_write = self.buffer[:maxbytes]
  545. else:
  546. to_write = self.buffer
  547. logger.info("%s->%s: %s", self._protocol, self.other, to_write)
  548. try:
  549. self.other.dataReceived(to_write)
  550. except Exception as e:
  551. logger.exception("Exception writing to protocol: %s", e)
  552. return
  553. self.buffer = self.buffer[len(to_write) :]
  554. if self.buffer and self.autoflush:
  555. self._reactor.callLater(0.0, self.flush)
  556. if not self.buffer and self.disconnecting:
  557. logger.info("FakeTransport: Buffer now empty, completing disconnect")
  558. self.disconnected = True
  559. def connect_client(
  560. reactor: ThreadedMemoryReactorClock, client_id: int
  561. ) -> Tuple[IProtocol, AccumulatingProtocol]:
  562. """
  563. Connect a client to a fake TCP transport.
  564. Args:
  565. reactor
  566. factory: The connecting factory to build.
  567. """
  568. factory = reactor.tcpClients.pop(client_id)[2]
  569. client = factory.buildProtocol(None)
  570. server = AccumulatingProtocol()
  571. server.makeConnection(FakeTransport(client, reactor))
  572. client.makeConnection(FakeTransport(server, reactor))
  573. return client, server
  574. class TestHomeServer(HomeServer):
  575. DATASTORE_CLASS = DataStore
  576. def setup_test_homeserver(
  577. cleanup_func,
  578. name="test",
  579. config=None,
  580. reactor=None,
  581. homeserver_to_use: Type[HomeServer] = TestHomeServer,
  582. **kwargs,
  583. ):
  584. """
  585. Setup a homeserver suitable for running tests against. Keyword arguments
  586. are passed to the Homeserver constructor.
  587. If no datastore is supplied, one is created and given to the homeserver.
  588. Args:
  589. cleanup_func : The function used to register a cleanup routine for
  590. after the test.
  591. Calling this method directly is deprecated: you should instead derive from
  592. HomeserverTestCase.
  593. """
  594. if reactor is None:
  595. from twisted.internet import reactor
  596. if config is None:
  597. config = default_config(name, parse=True)
  598. config.caches.resize_all_caches()
  599. config.ldap_enabled = False
  600. if "clock" not in kwargs:
  601. kwargs["clock"] = MockClock()
  602. if USE_POSTGRES_FOR_TESTS:
  603. test_db = "synapse_test_%s" % uuid.uuid4().hex
  604. database_config = {
  605. "name": "psycopg2",
  606. "args": {
  607. "database": test_db,
  608. "host": POSTGRES_HOST,
  609. "password": POSTGRES_PASSWORD,
  610. "user": POSTGRES_USER,
  611. "port": POSTGRES_PORT,
  612. "cp_min": 1,
  613. "cp_max": 5,
  614. },
  615. }
  616. else:
  617. if SQLITE_PERSIST_DB:
  618. # The current working directory is in _trial_temp, so this gets created within that directory.
  619. test_db_location = os.path.abspath("test.db")
  620. logger.debug("Will persist db to %s", test_db_location)
  621. # Ensure each test gets a clean database.
  622. try:
  623. os.remove(test_db_location)
  624. except FileNotFoundError:
  625. pass
  626. else:
  627. logger.debug("Removed existing DB at %s", test_db_location)
  628. else:
  629. test_db_location = ":memory:"
  630. database_config = {
  631. "name": "sqlite3",
  632. "args": {"database": test_db_location, "cp_min": 1, "cp_max": 1},
  633. }
  634. if "db_txn_limit" in kwargs:
  635. database_config["txn_limit"] = kwargs["db_txn_limit"]
  636. database = DatabaseConnectionConfig("master", database_config)
  637. config.database.databases = [database]
  638. db_engine = create_engine(database.config)
  639. # Create the database before we actually try and connect to it, based off
  640. # the template database we generate in setupdb()
  641. if isinstance(db_engine, PostgresEngine):
  642. db_conn = db_engine.module.connect(
  643. database=POSTGRES_BASE_DB,
  644. user=POSTGRES_USER,
  645. host=POSTGRES_HOST,
  646. port=POSTGRES_PORT,
  647. password=POSTGRES_PASSWORD,
  648. )
  649. db_conn.autocommit = True
  650. cur = db_conn.cursor()
  651. cur.execute("DROP DATABASE IF EXISTS %s;" % (test_db,))
  652. cur.execute(
  653. "CREATE DATABASE %s WITH TEMPLATE %s;" % (test_db, POSTGRES_BASE_DB)
  654. )
  655. cur.close()
  656. db_conn.close()
  657. hs = homeserver_to_use(
  658. name,
  659. config=config,
  660. version_string="Synapse/tests",
  661. reactor=reactor,
  662. )
  663. # Install @cache_in_self attributes
  664. for key, val in kwargs.items():
  665. setattr(hs, "_" + key, val)
  666. # Mock TLS
  667. hs.tls_server_context_factory = Mock()
  668. hs.tls_client_options_factory = Mock()
  669. hs.setup()
  670. if homeserver_to_use == TestHomeServer:
  671. hs.setup_background_tasks()
  672. if isinstance(db_engine, PostgresEngine):
  673. database = hs.get_datastores().databases[0]
  674. # We need to do cleanup on PostgreSQL
  675. def cleanup():
  676. import psycopg2
  677. # Close all the db pools
  678. database._db_pool.close()
  679. dropped = False
  680. # Drop the test database
  681. db_conn = db_engine.module.connect(
  682. database=POSTGRES_BASE_DB,
  683. user=POSTGRES_USER,
  684. host=POSTGRES_HOST,
  685. port=POSTGRES_PORT,
  686. password=POSTGRES_PASSWORD,
  687. )
  688. db_conn.autocommit = True
  689. cur = db_conn.cursor()
  690. # Try a few times to drop the DB. Some things may hold on to the
  691. # database for a few more seconds due to flakiness, preventing
  692. # us from dropping it when the test is over. If we can't drop
  693. # it, warn and move on.
  694. for _ in range(5):
  695. try:
  696. cur.execute("DROP DATABASE IF EXISTS %s;" % (test_db,))
  697. db_conn.commit()
  698. dropped = True
  699. except psycopg2.OperationalError as e:
  700. warnings.warn(
  701. "Couldn't drop old db: " + str(e), category=UserWarning
  702. )
  703. time.sleep(0.5)
  704. cur.close()
  705. db_conn.close()
  706. if not dropped:
  707. warnings.warn("Failed to drop old DB.", category=UserWarning)
  708. if not LEAVE_DB:
  709. # Register the cleanup hook
  710. cleanup_func(cleanup)
  711. # bcrypt is far too slow to be doing in unit tests
  712. # Need to let the HS build an auth handler and then mess with it
  713. # because AuthHandler's constructor requires the HS, so we can't make one
  714. # beforehand and pass it in to the HS's constructor (chicken / egg)
  715. async def hash(p):
  716. return hashlib.md5(p.encode("utf8")).hexdigest()
  717. hs.get_auth_handler().hash = hash
  718. async def validate_hash(p, h):
  719. return hashlib.md5(p.encode("utf8")).hexdigest() == h
  720. hs.get_auth_handler().validate_hash = validate_hash
  721. # Make the threadpool and database transactions synchronous for testing.
  722. _make_test_homeserver_synchronous(hs)
  723. return hs