def setup_test_homeserver(name="test", datastore=None, config=None, **kargs):
    """Setup a homeserver suitable for running tests against. Keyword arguments
    are passed to the Homeserver constructor. If no datastore is supplied a
    datastore backed by an in-memory sqlite db will be given to the HS.
    """
    if config is None:
        config = Mock()
        config.signing_key = [MockKey()]
        config.event_cache_size = 1
        config.disable_registration = False

    if "clock" not in kargs:
        kargs["clock"] = MockClock()

    if datastore is None:
        db_pool = SQLiteMemoryDbPool()
        yield db_pool.prepare()
        hs = HomeServer(
            name, db_pool=db_pool, config=config,
            version_string="Synapse/tests",
            database_engine=create_engine("sqlite3"),
            **kargs
        )
    else:
        hs = HomeServer(
            name, db_pool=None, datastore=datastore, config=config,
            version_string="Synapse/tests",
            database_engine=create_engine("sqlite3"),
            **kargs
        )

    defer.returnValue(hs)

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse synchrotron", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + str(e) + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.synchrotron"

    setup_logging(config, use_worker_options=True)

    synapse.events.USE_FROZEN_DICTS = config.use_frozen_dicts

    database_engine = create_engine(config.database_config)

    ss = SynchrotronServer(
        config.server_name,
        db_config=config.database_config,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
        application_service_handler=SynchrotronApplicationService(),
    )

    ss.setup()
    ss.start_listening(config.worker_listeners)

    def start():
        ss.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

    _base.start_worker_reactor("synapse-synchrotron", config)

    def setUp(self):
        self.db_pool = Mock(spec=["runInteraction"])
        self.mock_txn = Mock()
        self.mock_conn = Mock(spec_set=["cursor", "rollback", "commit"])
        self.mock_conn.cursor.return_value = self.mock_txn
        self.mock_conn.rollback.return_value = None
        # Our fake runInteraction just runs synchronously inline

        def runInteraction(func, *args, **kwargs):
            return defer.succeed(func(self.mock_txn, *args, **kwargs))

        self.db_pool.runInteraction = runInteraction

        def runWithConnection(func, *args, **kwargs):
            return defer.succeed(func(self.mock_conn, *args, **kwargs))

        self.db_pool.runWithConnection = runWithConnection

        config = Mock()
        config._disable_native_upserts = True
        config.event_cache_size = 1
        config.database_config = {"name": "sqlite3"}
        engine = create_engine(config.database_config)
        fake_engine = Mock(wraps=engine)
        fake_engine.can_native_upsert = False
        hs = TestHomeServer(
            "test", db_pool=self.db_pool, config=config, database_engine=fake_engine
        )

        self.datastore = SQLBaseStore(None, hs)

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse client reader", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + str(e) + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.client_reader"

    setup_logging(config, use_worker_options=True)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    database_engine = create_engine(config.database_config)

    ss = ClientReaderServer(
        config.server_name,
        db_config=config.database_config,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ss.setup()
    reactor.callWhenRunning(_base.start, ss, config.worker_listeners)

    _base.start_worker_reactor("synapse-client-reader", config)

    def setUp(self):
        self.db_pool = Mock(spec=["runInteraction"])
        self.mock_txn = Mock()
        self.mock_conn = Mock(spec_set=["cursor"])
        self.mock_conn.cursor.return_value = self.mock_txn
        # Our fake runInteraction just runs synchronously inline

        def runInteraction(func, *args, **kwargs):
            return defer.succeed(func(self.mock_txn, *args, **kwargs))
        self.db_pool.runInteraction = runInteraction

        def runWithConnection(func, *args, **kwargs):
            return defer.succeed(func(self.mock_conn, *args, **kwargs))
        self.db_pool.runWithConnection = runWithConnection

        config = Mock()
        config.event_cache_size = 1
        hs = HomeServer(
            "test",
            db_pool=self.db_pool,
            config=config,
            database_engine=create_engine("sqlite3"),
        )

        self.datastore = SQLBaseStore(hs)

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse frontend proxy", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + str(e) + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.frontend_proxy"

    assert config.worker_main_http_uri is not None

    setup_logging(config, use_worker_options=True)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    database_engine = create_engine(config.database_config)

    ss = FrontendProxyServer(
        config.server_name,
        db_config=config.database_config,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ss.setup()
    reactor.callWhenRunning(_base.start, ss, config.worker_listeners)

    _base.start_worker_reactor("synapse-frontend-proxy", config)

def setup_test_homeserver(name="test", datastore=None, config=None, **kargs):
    """Setup a homeserver suitable for running tests against. Keyword arguments
    are passed to the Homeserver constructor. If no datastore is supplied a
    datastore backed by an in-memory sqlite db will be given to the HS.
    """
    if config is None:
        config = Mock()
        config.signing_key = [MockKey()]
        config.event_cache_size = 1
        config.disable_registration = False
        config.macaroon_secret_key = "not even a little secret"
        config.server_name = "server.under.test"

    if "clock" not in kargs:
        kargs["clock"] = MockClock()

    if datastore is None:
        db_pool = SQLiteMemoryDbPool()
        yield db_pool.prepare()
        hs = HomeServer(
            name, db_pool=db_pool, config=config,
            version_string="Synapse/tests",
            database_engine=create_engine("sqlite3"),
            **kargs
        )
    else:
        hs = HomeServer(
            name, db_pool=None, datastore=datastore, config=config,
            version_string="Synapse/tests",
            database_engine=create_engine("sqlite3"),
            **kargs
        )

    # bcrypt is far too slow to be doing in unit tests
    def swap_out_hash_for_testing(old_build_handlers):
        def build_handlers():
            handlers = old_build_handlers()
            auth_handler = handlers.auth_handler
            auth_handler.hash = lambda p: hashlib.md5(p).hexdigest()
            auth_handler.validate_hash = lambda p, h: hashlib.md5(p).hexdigest() == h
            return handlers
        return build_handlers

    hs.build_handlers = swap_out_hash_for_testing(hs.build_handlers)

    defer.returnValue(hs)

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse federation reader", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + e.message + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.federation_reader"

    setup_logging(config.worker_log_config, config.worker_log_file)

    database_engine = create_engine(config.database_config)

    tls_server_context_factory = context_factory.ServerContextFactory(config)

    ss = FederationReaderServer(
        config.server_name,
        db_config=config.database_config,
        tls_server_context_factory=tls_server_context_factory,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ss.setup()
    ss.get_handlers()
    ss.start_listening(config.worker_listeners)

    def run():
        with LoggingContext("run"):
            logger.info("Running")
            change_resource_limit(config.soft_file_limit)
            if config.gc_thresholds:
                gc.set_threshold(*config.gc_thresholds)
            reactor.run()

    def start():
        ss.get_state_handler().start_caching()
        ss.get_datastore().start_profiling()
        ss.replicate()

    reactor.callWhenRunning(start)

    if config.worker_daemonize:
        daemon = Daemonize(
            app="synapse-federation-reader",
            pid=config.worker_pid_file,
            action=run,
            auto_close_fds=False,
            verbose=True,
            logger=logger,
        )
        daemon.start()
    else:
        run()

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse user directory", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + str(e) + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.user_dir"

    setup_logging(config, use_worker_options=True)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    database_engine = create_engine(config.database_config)

    if config.update_user_directory:
        sys.stderr.write(
            "\nThe update_user_directory must be disabled in the main synapse process"
            "\nbefore they can be run in a separate worker."
            "\nPlease add ``update_user_directory: false`` to the main config"
            "\n"
        )
        sys.exit(1)

    # Force the pushers to start since they will be disabled in the main config
    config.update_user_directory = True

    tls_server_context_factory = context_factory.ServerContextFactory(config)
    tls_client_options_factory = context_factory.ClientTLSOptionsFactory(config)

    ps = UserDirectoryServer(
        config.server_name,
        db_config=config.database_config,
        tls_server_context_factory=tls_server_context_factory,
        tls_client_options_factory=tls_client_options_factory,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ps.setup()
    ps.start_listening(config.worker_listeners)

    def start():
        ps.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

    _base.start_worker_reactor("synapse-user-dir", config)

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse federation sender", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + e.message + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.federation_sender"

    setup_logging(config, use_worker_options=True)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    database_engine = create_engine(config.database_config)

    if config.send_federation:
        sys.stderr.write(
            "\nThe send_federation must be disabled in the main synapse process"
            "\nbefore they can be run in a separate worker."
            "\nPlease add ``send_federation: false`` to the main config"
            "\n"
        )
        sys.exit(1)

    # Force the pushers to start since they will be disabled in the main config
    config.send_federation = True

    tls_server_context_factory = context_factory.ServerContextFactory(config)

    ps = FederationSenderServer(
        config.server_name,
        db_config=config.database_config,
        tls_server_context_factory=tls_server_context_factory,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ps.setup()
    ps.start_listening(config.worker_listeners)

    def start():
        ps.get_datastore().start_profiling()
        ps.get_state_handler().start_caching()

    reactor.callWhenRunning(start)
    _base.start_worker_reactor("synapse-federation-sender", config)

def setupdb():
    # If we're using PostgreSQL, set up the db once
    if USE_POSTGRES_FOR_TESTS:
        # create a PostgresEngine
        db_engine = create_engine({"name": "psycopg2", "args": {}})

        # connect to postgres to create the base database.
        db_conn = db_engine.module.connect(
            user=POSTGRES_USER,
            host=POSTGRES_HOST,
            password=POSTGRES_PASSWORD,
            dbname=POSTGRES_DBNAME_FOR_INITIAL_CREATE,
        )
        db_conn.autocommit = True
        cur = db_conn.cursor()
        cur.execute("DROP DATABASE IF EXISTS %s;" % (POSTGRES_BASE_DB,))
        cur.execute("CREATE DATABASE %s;" % (POSTGRES_BASE_DB,))
        cur.close()
        db_conn.close()

        # Set up in the db
        db_conn = db_engine.module.connect(
            database=POSTGRES_BASE_DB,
            user=POSTGRES_USER,
            host=POSTGRES_HOST,
            password=POSTGRES_PASSWORD,
        )
        cur = db_conn.cursor()
        _get_or_create_schema_state(cur, db_engine)
        _setup_new_database(cur, db_engine)
        db_conn.commit()
        cur.close()
        db_conn.close()

        def _cleanup():
            db_conn = db_engine.module.connect(
                user=POSTGRES_USER,
                host=POSTGRES_HOST,
                password=POSTGRES_PASSWORD,
                dbname=POSTGRES_DBNAME_FOR_INITIAL_CREATE,
            )
            db_conn.autocommit = True
            cur = db_conn.cursor()
            cur.execute("DROP DATABASE IF EXISTS %s;" % (POSTGRES_BASE_DB,))
            cur.close()
            db_conn.close()

        atexit.register(_cleanup)

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse pusher", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + str(e) + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.pusher"

    setup_logging(config, use_worker_options=True)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    if config.start_pushers:
        sys.stderr.write(
            "\nThe pushers must be disabled in the main synapse process"
            "\nbefore they can be run in a separate worker."
            "\nPlease add ``start_pushers: false`` to the main config"
            "\n"
        )
        sys.exit(1)

    # Force the pushers to start since they will be disabled in the main config
    config.start_pushers = True

    database_engine = create_engine(config.database_config)

    ps = PusherServer(
        config.server_name,
        db_config=config.database_config,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ps.setup()
    ps.start_listening(config.worker_listeners)

    def start():
        ps.get_pusherpool().start()
        ps.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

    _base.start_worker_reactor("synapse-pusher", config)

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse media repository", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + str(e) + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.media_repository"

    if config.enable_media_repo:
        _base.quit_with_error(
            "enable_media_repo must be disabled in the main synapse process\n"
            "before the media repo can be run in a separate worker.\n"
            "Please add ``enable_media_repo: false`` to the main config\n"
        )

    setup_logging(config, use_worker_options=True)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    database_engine = create_engine(config.database_config)

    tls_server_context_factory = context_factory.ServerContextFactory(config)
    tls_client_options_factory = context_factory.ClientTLSOptionsFactory(config)

    ss = MediaRepositoryServer(
        config.server_name,
        db_config=config.database_config,
        tls_server_context_factory=tls_server_context_factory,
        tls_client_options_factory=tls_client_options_factory,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ss.setup()
    ss.start_listening(config.worker_listeners)

    def start():
        ss.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

    _base.start_worker_reactor("synapse-media-repository", config)

def setupdb():

    # If we're using PostgreSQL, set up the db once
    if USE_POSTGRES_FOR_TESTS:
        pgconfig = {
            "name": "psycopg2",
            "args": {
                "database": POSTGRES_BASE_DB,
                "user": POSTGRES_USER,
                "cp_min": 1,
                "cp_max": 5,
            },
        }
        config = Mock()
        config.password_providers = []
        config.database_config = pgconfig
        db_engine = create_engine(pgconfig)
        db_conn = db_engine.module.connect(user=POSTGRES_USER)
        db_conn.autocommit = True
        cur = db_conn.cursor()
        cur.execute("DROP DATABASE IF EXISTS %s;" % (POSTGRES_BASE_DB,))
        cur.execute("CREATE DATABASE %s;" % (POSTGRES_BASE_DB,))
        cur.close()
        db_conn.close()

        # Set up in the db
        db_conn = db_engine.module.connect(
            database=POSTGRES_BASE_DB, user=POSTGRES_USER
        )
        cur = db_conn.cursor()
        _get_or_create_schema_state(cur, db_engine)
        _setup_new_database(cur, db_engine)
        db_conn.commit()
        cur.close()
        db_conn.close()

        def _cleanup():
            db_conn = db_engine.module.connect(user=POSTGRES_USER)
            db_conn.autocommit = True
            cur = db_conn.cursor()
            cur.execute("DROP DATABASE IF EXISTS %s;" % (POSTGRES_BASE_DB,))
            cur.close()
            db_conn.close()

        atexit.register(_cleanup)

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse frontend proxy", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + e.message + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.frontend_proxy"

    assert config.worker_main_http_uri is not None

    setup_logging(config, use_worker_options=True)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    database_engine = create_engine(config.database_config)

    tls_server_context_factory = context_factory.ServerContextFactory(config)

    ss = FrontendProxyServer(
        config.server_name,
        db_config=config.database_config,
        tls_server_context_factory=tls_server_context_factory,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ss.setup()
    ss.start_listening(config.worker_listeners)

    def start():
        ss.get_state_handler().start_caching()
        ss.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

    _base.start_worker_reactor("synapse-frontend-proxy", config)

def setup(config_options):
    try:
        config = PusherSlaveConfig.load_config(
            "Synapse pusher", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + e.message + "\n")
        sys.exit(1)

    if not config:
        sys.exit(0)

    config.setup_logging()

    database_engine = create_engine(config.database_config)

    ps = PusherServer(
        config.server_name,
        db_config=config.database_config,
        config=config,
        version_string=get_version_string("Synapse", synapse),
        database_engine=database_engine,
    )

    ps.setup()
    ps.start_listening()

    change_resource_limit(ps.config.soft_file_limit)
    if ps.config.gc_thresholds:
        gc.set_threshold(*ps.config.gc_thresholds)

    def start():
        ps.replicate()
        ps.get_pusherpool().start()
        ps.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

    return ps

def start(config_options):
    try:
        config = HomeServerConfig.load_config(
            "Synapse federation reader", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + str(e) + "\n")
        sys.exit(1)

    assert config.worker_app == "synapse.app.federation_reader"

    setup_logging(config, use_worker_options=True)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    database_engine = create_engine(config.database_config)

    tls_server_context_factory = context_factory.ServerContextFactory(config)
    tls_client_options_factory = context_factory.ClientTLSOptionsFactory(config)

    ss = FederationReaderServer(
        config.server_name,
        db_config=config.database_config,
        tls_server_context_factory=tls_server_context_factory,
        tls_client_options_factory=tls_client_options_factory,
        config=config,
        version_string="Synapse/" + get_version_string(synapse),
        database_engine=database_engine,
    )

    ss.setup()
    ss.start_listening(config.worker_listeners)

    def start():
        ss.get_datastore().start_profiling()

    reactor.callWhenRunning(start)

    _base.start_worker_reactor("synapse-federation-reader", config)

def setup(config_options):
    try:
        config = SynchrotronConfig.load_config(
            "Synapse synchrotron", config_options
        )
    except ConfigError as e:
        sys.stderr.write("\n" + e.message + "\n")
        sys.exit(1)

    if not config:
        sys.exit(0)

    config.setup_logging()

    database_engine = create_engine(config.database_config)

    ss = SynchrotronServer(
        config.server_name,
        db_config=config.database_config,
        config=config,
        version_string=get_version_string("Synapse", synapse),
        database_engine=database_engine,
        application_service_handler=SynchrotronApplicationService(),
    )

    ss.setup()
    ss.start_listening()

    change_resource_limit(ss.config.soft_file_limit)
    if ss.config.gc_thresholds:
        ss.set_threshold(*ss.config.gc_thresholds)

    def start():
        ss.get_datastore().start_profiling()
        ss.replicate()

    reactor.callWhenRunning(start)

    return ss

 def prepare(self):
     engine = create_engine("sqlite3")
     return self.runWithConnection(
         lambda conn: prepare_database(conn, engine)
     )

def setup(config_options):
    """
    Args:
        config_options_options: The options passed to Synapse. Usually
            `sys.argv[1:]`.

    Returns:
        HomeServer
    """
    try:
        config = HomeServerConfig.load_or_generate_config(
            "Synapse Homeserver",
            config_options,
        )
    except ConfigError as e:
        sys.stderr.write("\n" + e.message + "\n")
        sys.exit(1)

    if not config:
        # If a config isn't returned, and an exception isn't raised, we're just
        # generating config files and shouldn't try to continue.
        sys.exit(0)

    config.setup_logging()

    # check any extra requirements we have now we have a config
    check_requirements(config)

    version_string = get_version_string("Synapse", synapse)

    logger.info("Server hostname: %s", config.server_name)
    logger.info("Server version: %s", version_string)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts

    tls_server_context_factory = context_factory.ServerContextFactory(config)

    database_engine = create_engine(config.database_config)
    config.database_config["args"]["cp_openfun"] = database_engine.on_new_connection

    hs = SynapseHomeServer(
        config.server_name,
        db_config=config.database_config,
        tls_server_context_factory=tls_server_context_factory,
        config=config,
        content_addr=config.content_addr,
        version_string=version_string,
        database_engine=database_engine,
    )

    logger.info("Preparing database: %s...", config.database_config['name'])

    try:
        db_conn = hs.get_db_conn(run_new_connection=False)
        prepare_database(db_conn, database_engine, config=config)
        database_engine.on_new_connection(db_conn)

        hs.run_startup_checks(db_conn, database_engine)

        db_conn.commit()
    except UpgradeDatabaseException:
        sys.stderr.write(
            "\nFailed to upgrade database.\n"
            "Have you checked for version specific instructions in"
            " UPGRADES.rst?\n"
        )
        sys.exit(1)

    logger.info("Database prepared in %s.", config.database_config['name'])

    hs.setup()
    hs.start_listening()

    def start():
        hs.get_pusherpool().start()
        hs.get_state_handler().start_caching()
        hs.get_datastore().start_profiling()
        hs.get_datastore().start_doing_background_updates()
        hs.get_replication_layer().start_get_pdu_cache()

    reactor.callWhenRunning(start)

    return hs